Method _build_base_samples_cte

xtop/core/query_builder.py:457–608

Build base_samples CTE with JOINs and filters

(self, required_sources: Set[str],
                                where_clause: str,
                                low_time: Optional[datetime],
                                high_time: Optional[datetime],
                                need_sc_histogram: bool,
                                need_io_histogram: bool)

    FROM ({base_samples}) AS samples"""
    
    def _build_base_samples_cte(self, required_sources: Set[str],
                                where_clause: str,
                                low_time: Optional[datetime],
                                high_time: Optional[datetime],
                                need_sc_histogram: bool,
                                need_io_histogram: bool) -> str:
        """Build base_samples CTE with JOINs and filters"""
        # Start with SELECT columns
        select_cols = ["es.*"]
        
        # Determine join availability based on schema info
        syscend_join = 'syscend' in required_sources and self._has_columns('syscend', ['tid', 'sysc_seq_num'])
        iorq_join = 'iorqend' in required_sources and self._has_columns('iorqend', ['insert_tid', 'iorq_seq_num'])
        kstack_join = 'kstacks' in required_sources and self._has_column('kstacks', 'kstack_hash')
        ustack_join = 'ustacks' in required_sources and self._has_column('ustacks', 'ustack_hash')
        partitions_join = (
            'partitions' in required_sources
            and iorq_join
            and self._has_columns('iorqend', ['dev_maj', 'dev_min'])
            and self._has_columns('partitions', ['dev_maj', 'dev_min'])
        )

        def project(source: str, alias: str, column: str, output_alias: str, join_available: bool) -> str:
            if not join_available:
                return f"NULL AS {output_alias}"
            return self._column_expr(source, alias, column, output_alias)

        # Add columns from joined sources (use NULL fallbacks when missing)
        if 'syscend' in required_sources:
            select_cols.append(project('syscend', 'sc', 'duration_ns', 'sc_duration_ns', syscend_join and self._has_column('syscend', 'duration_ns')))
            select_cols.append(project('syscend', 'sc', 'type', 'sc_type', syscend_join))
            if need_sc_histogram and syscend_join and self._has_column('syscend', 'duration_ns'):
                bucket_calc = self.fragments.load('histogram_buckets')
                bucket_calc = bucket_calc.replace('#DURATION_COLUMN#', 'sc.duration_ns')
                bucket_calc = bucket_calc.replace('lat_bkt_us', 'sc_lat_bkt_us')
                select_cols.append(bucket_calc)
            elif need_sc_histogram:
                select_cols.append('NULL AS sc_lat_bkt_us')

        if 'iorqend' in required_sources:
            select_cols.append(project('iorqend', 'io', 'duration_ns', 'io_duration_ns', iorq_join and self._has_column('iorqend', 'duration_ns')))
            select_cols.append(project('iorqend', 'io', 'service_ns', 'io_service_ns', iorq_join and self._has_column('iorqend', 'service_ns')))
            select_cols.append(project('iorqend', 'io', 'queued_ns', 'io_queued_ns', iorq_join and self._has_column('iorqend', 'queued_ns')))
            select_cols.append(project('iorqend', 'io', 'bytes', 'io_bytes', iorq_join and self._has_column('iorqend', 'bytes')))
            select_cols.append(project('iorqend', 'io', 'dev_maj', 'io_dev_maj', iorq_join and self._has_column('iorqend', 'dev_maj')))
            select_cols.append(project('iorqend', 'io', 'dev_min', 'io_dev_min', iorq_join and self._has_column('iorqend', 'dev_min')))
            select_cols.append(project('iorqend', 'io', 'iorq_flags', 'iorq_flags', iorq_join and self._has_column('iorqend', 'iorq_flags')))
            if need_io_histogram and iorq_join and self._has_column('iorqend', 'duration_ns'):
                bucket_calc = self.fragments.load('histogram_buckets')
                bucket_calc = bucket_calc.replace('#DURATION_COLUMN#', 'io.duration_ns')
                bucket_calc = bucket_calc.replace('lat_bkt_us', 'io_lat_bkt_us')
                select_cols.append(bucket_calc)
            elif need_io_histogram:
                select_cols.append('NULL AS io_lat_bkt_us')

        if 'kstacks' in required_sources:
            select_cols.append(project('kstacks', 'ks', 'kstack_hash', 'KSTACK_HASH', kstack_join))
            select_cols.append(project('kstacks', 'ks', 'kstack_syms', 'KSTACK_SYMS', kstack_join and self._has_column('kstacks', 'kstack_syms')))