Method _determine_required_sources

xtop/core/query_engine.py:416–459

Determine which data sources are needed based on requested columns

(self, columns: Set[str])

        return result
    
    def _determine_required_sources(self, columns: Set[str]) -> Set[str]:
        """Determine which data sources are needed based on requested columns"""
        required_sources = {'samples'}  # Always need the base source
        column_mapping = self.get_column_to_source_mapping()
        
        # Check which sources are needed for the requested columns
        for col in columns:
            # Handle prefixed columns (e.g., "sc.duration_ns")
            if '.' in col:
                prefix = col.split('.')[0]
                # Find source by alias
                for src_name, src_info in self.DATA_SOURCES.items():
                    if src_info['alias'] == prefix:
                        required_sources.add(src_name)
                        # Add dependencies
                        for dep in src_info['depends_on']:
                            required_sources.add(dep)
                        break
            else:
                source = column_mapping.get(col)
                if source:
                    required_sources.add(source)
                    # Add dependencies
                    for dep in self.DATA_SOURCES[source]['depends_on']:
                        required_sources.add(dep)
        
        # Add computed columns dependencies
        if 'kstack_current_func' in columns:
            required_sources.add('kstacks')
        if 'ustack_current_func' in columns:
            required_sources.add('ustacks')
        
        # Add dependencies for latency columns
        syscall_latency_cols = {'sc.min_lat_us', 'sc.avg_lat_us', 'sc.max_lat_us', 
                               'sc.p50_us', 'sc.p95_us', 'sc.p99_us', 'sc.p999_us', 'sclat_histogram'}
        io_latency_cols = {'io.min_lat_us', 'io.avg_lat_us', 'io.max_lat_us',
                          'io.p50_us', 'io.p95_us', 'io.p99_us', 'io.p999_us', 'iolat_histogram'}
        
        if any(col in columns for col in syscall_latency_cols):
            required_sources.add('syscend')
        if any(col in columns for col in io_latency_cols):
            required_sources.add('iorqend')
        
        return required_sources
    
    def _load_fragment(self, source_name: str) -> str:
        """Load SQL fragment for a data source"""