Method _discover_all_schemas

xtop/core/query_engine.py:323–376

Discover schemas for all data sources

(self)

        self.query_builder.use_materialized = use_materialized
    
    def _discover_all_schemas(self):
        """Discover schemas for all data sources"""
        for source_name, source_info in self.DATA_SOURCES.items():
            if source_name == 'partitions':
                # Special handling for partitions - hardcode the schema
                # The partitions file requires special parsing, so we can't auto-discover
                self.schema_cache[source_name] = [
                    ('dev_maj', 'INTEGER'),
                    ('dev_min', 'INTEGER'),
                    ('devname', 'VARCHAR')
                ]
                continue
                
            # Build a simple query to get schema
            fragment_path = self.fragments_path / source_info['fragment']
            if not fragment_path.exists():
                self.logger.warning(f"Fragment file not found: {fragment_path}")
                continue
                
            with open(fragment_path, 'r') as f:
                fragment = f.read()
            
            # Extract the SELECT statement from the fragment
            if source_info['is_base']:
                # Base query - use directly
                query = fragment.replace('#XTOP_DATADIR#', str(self.data_source.datadir))
            else:
                # Extract the SELECT from within the JOIN
                match = re.search(r'SELECT.*?FROM\s+read_csv_auto\([^)]+\)', fragment, re.DOTALL | re.IGNORECASE)
                if match:
                    query = match.group(0).replace('#XTOP_DATADIR#', str(self.data_source.datadir))
                else:
                    self.logger.warning(f"Could not extract SELECT from fragment: {source_name}")
                    continue
            
            # Get schema using DESCRIBE, with parquet fallback if CSV not available
            def _describe(q: str):
                conn = self.data_source.connect()
                return conn.execute(f"DESCRIBE ({q} LIMIT 0)").fetchall()
            try:
                result = _describe(query)
            except Exception as e:
                # Attempt parquet fallback by replacing CSV reader/patterns
                try:
                    pq_query = (query
                                .replace('read_csv_auto', 'read_parquet')
                                .replace('*.csv', '*.parquet'))
                    result = _describe(pq_query)
                except Exception as e2:
                    self.logger.error(f"Error discovering schema for {source_name}: {e2}")
                    result = []
            self.schema_cache[source_name] = [(row[0], row[1]) for row in result]
            if result:
                self.logger.debug(f"Discovered schema for {source_name}: {len(result)} columns")
    
    def get_column_to_source_mapping(self) -> Dict[str, str]:
        """Get mapping of column names to their source data files"""