vibesql-executor 0.2.0

//! Native Columnar Execution Pipeline
//!
//! Implements the `ExecutionPipeline` trait using zero-copy columnar execution
//! directly from storage. This pipeline keeps data in columnar format throughout
//! the execution to minimize materialization overhead.

use vibesql_ast::{Expression, SelectItem};
use vibesql_storage::Row;

use super::{ExecutionContext, ExecutionPipeline, PipelineInput, PipelineOutput};
use crate::{
    errors::ExecutorError,
    select::columnar::{
        extract_aggregates, extract_column_predicates, simd_filter_batch, AggregateOp,
        AggregateSource, AggregateSpec, ColumnarBatch,
    },
};

/// Native columnar execution pipeline.
///
/// This pipeline operates on data in columnar format throughout the execution,
/// only converting to rows at the final output stage. This provides maximum
/// performance for large-scale analytical queries.
///
/// # Performance
///
/// Uses LLVM auto-vectorization for SIMD-accelerated operations.
///
/// # Performance Characteristics
///
/// - **Filter**: SIMD-accelerated predicate evaluation (4-8x speedup)
/// - **Projection**: Zero-copy column selection
/// - **Aggregation**: SIMD reduction operations (10x speedup)
/// - **GROUP BY**: Hash-based grouping with columnar aggregation
///
/// # When to Use
///
/// - Large datasets (>100k rows)
/// - TPC-H style analytical queries
/// - Simple GROUP BY with column references
/// - Single table scans without complex JOINs
///
/// # Limitations
///
/// - No JOIN support (single table only)
/// - GROUP BY limited to simple column references
/// - No ROLLUP/CUBE/GROUPING SETS support
/// - Requires columnar storage format
pub struct NativeColumnarPipeline {
    /// Whether the pipeline has access to columnar storage
    #[allow(dead_code)]
    has_columnar_storage: bool,
}

impl NativeColumnarPipeline {
    /// Create a new native columnar pipeline.
    #[inline]
    pub fn new() -> Self {
        Self { has_columnar_storage: true }
    }

    /// Create a native columnar pipeline with explicit columnar storage availability.
    #[inline]
    #[allow(dead_code)]
    pub fn with_storage(has_columnar_storage: bool) -> Self {
        Self { has_columnar_storage }
    }
}

impl Default for NativeColumnarPipeline {
    fn default() -> Self {
        Self::new()
    }
}

impl ExecutionPipeline for NativeColumnarPipeline {
    /// Create an evaluator with context for native columnar execution.
    #[inline]
    fn create_evaluator<'a>(
        &self,
        ctx: &'a ExecutionContext<'a>,
    ) -> crate::evaluator::CombinedExpressionEvaluator<'a> {
        ctx.create_evaluator()
    }

    /// Apply WHERE clause filtering using SIMD-accelerated columnar operations.
    ///
    /// This is the most optimized path:
    /// 1. Extracts simple predicates from WHERE clause
    /// 2. Applies SIMD filtering directly on column arrays
    /// 3. Returns filtered columnar batch (converted to rows for output)
    fn apply_filter(
        &self,
        input: PipelineInput<'_>,
        predicate: Option<&Expression>,
        ctx: &ExecutionContext<'_>,
    ) -> Result<PipelineOutput, ExecutorError> {
        // Handle native columnar input specially
        if let PipelineInput::NativeColumnar { table_name, column_indices: _ } = &input {
            // Get columnar data directly from storage
            let columnar_table = match ctx.database.get_columnar(table_name) {
                Ok(Some(ct)) => ct,
                Ok(None) | Err(_) => {
                    // Fall back to row-based if columnar not available
                    return self.fallback_filter(input.into_rows(), predicate, ctx);
                }
            };

            // Skip empty tables
            if columnar_table.row_count() == 0 {
                return Ok(PipelineOutput::Empty);
            }

            // Convert to batch
            let batch = ColumnarBatch::from_storage_columnar(&columnar_table)?;

            // No predicate = return all rows
            if predicate.is_none() {
                let rows = batch.to_rows()?;
                return Ok(PipelineOutput::from_rows(rows));
            }

            let predicate = predicate.unwrap();

            // Extract simple predicates for SIMD filtering
            let predicates = match extract_column_predicates(predicate, ctx.schema, ctx.database.case_sensitive_like()) {
                Some(preds) => preds,
                None => {
                    // Complex predicate - fall back to row filtering
                    let rows = batch.to_rows()?;
                    return self.fallback_filter(rows, Some(predicate), ctx);
                }
            };

            // Apply SIMD-accelerated filtering
            let filtered_batch =
                if predicates.is_empty() { batch } else { simd_filter_batch(&batch, &predicates)? };

            // Return batch directly - avoid row conversion overhead
            // The batch will stay in columnar format through the pipeline
            return Ok(PipelineOutput::from_batch(filtered_batch));
        }

        // For row-based input, convert to columnar and filter
        let rows = input.into_rows();

        if predicate.is_none() {
            return Ok(PipelineOutput::from_rows(rows));
        }

        let predicate = predicate.unwrap();

        // Try to extract simple predicates
        let predicates = match extract_column_predicates(predicate, ctx.schema, ctx.database.case_sensitive_like()) {
            Some(preds) => preds,
            None => {
                return self.fallback_filter(rows, Some(predicate), ctx);
            }
        };

        if rows.is_empty() {
            return Ok(PipelineOutput::Empty);
        }

        // Convert to columnar batch
        let batch = ColumnarBatch::from_rows(&rows)?;

        // Apply SIMD filtering
        let filtered_batch =
            if predicates.is_empty() { batch } else { simd_filter_batch(&batch, &predicates)? };

        // Return batch directly - keep data in columnar format
        Ok(PipelineOutput::from_batch(filtered_batch))
    }

    /// Apply SELECT projection using columnar operations.
    ///
    /// In native columnar execution, projection is typically implicit -
    /// we only compute the required columns. For explicit projection,
    /// we fall back to row-oriented processing only when necessary.
    fn apply_projection(
        &self,
        input: PipelineInput<'_>,
        select_items: &[SelectItem],
        ctx: &ExecutionContext<'_>,
    ) -> Result<PipelineOutput, ExecutorError> {
        // For batch input, convert to rows for projection then convert back
        // This is less efficient but ensures correctness for complex projections
        // TODO: Implement native columnar projection for simple column selections
        let rows = match input {
            PipelineInput::Batch(batch) => {
                if batch.row_count() == 0 {
                    return Ok(PipelineOutput::Empty);
                }
                batch.to_rows()?
            }
            PipelineInput::Rows(rows) => {
                if rows.is_empty() {
                    return Ok(PipelineOutput::Empty);
                }
                rows.to_vec()
            }
            PipelineInput::RowsOwned(rows) => {
                if rows.is_empty() {
                    return Ok(PipelineOutput::Empty);
                }
                rows
            }
            PipelineInput::Empty => return Ok(PipelineOutput::Empty),
            PipelineInput::NativeColumnar { .. } => {
                // Should not reach here - native columnar goes through filter first
                return Ok(PipelineOutput::Empty);
            }
        };

        // Use row-oriented projection for now
        // Native columnar projection would require computing column indices
        // and extracting only those columns from the batch
        let evaluator = ctx.create_evaluator();
        let buffer_pool = vibesql_storage::QueryBufferPool::new();

        let mut projected_rows = Vec::with_capacity(rows.len());
        for row in rows {
            let projected = crate::select::projection::project_row_combined(
                &row,
                select_items,
                &evaluator,
                ctx.schema,
                &None,
                &buffer_pool,
            )?;
            projected_rows.push(projected);
        }

        Ok(PipelineOutput::from_rows(projected_rows))
    }

    /// Execute aggregation using SIMD-accelerated columnar operations.
    ///
    /// This is the most optimized aggregation path:
    /// 1. Extracts aggregate specifications from SELECT list
    /// 2. For simple aggregates without GROUP BY: Uses SIMD reductions
    /// 3. For GROUP BY: Uses hash-based grouping with columnar aggregation
    fn apply_aggregation(
        &self,
        input: PipelineInput<'_>,
        select_items: &[SelectItem],
        group_by: Option<&[Expression]>,
        having: Option<&Expression>,
        ctx: &ExecutionContext<'_>,
    ) -> Result<PipelineOutput, ExecutorError> {
        // HAVING not supported in native columnar pipeline - fall back to row-oriented
        if having.is_some() {
            return Err(ExecutorError::UnsupportedFeature(
                "HAVING not supported in native columnar pipeline".to_string(),
            ));
        }

        // Extract aggregate expressions from select items
        let agg_exprs: Vec<Expression> = select_items
            .iter()
            .filter_map(|item| {
                if let SelectItem::Expression { expr, .. } = item {
                    Some(expr.clone())
                } else {
                    None
                }
            })
            .collect();

        // Try to extract aggregate specs
        let agg_specs = match extract_aggregates(&agg_exprs, ctx.schema) {
            Some(specs) => specs,
            None => {
                return Err(ExecutorError::UnsupportedFeature(
                    "Complex aggregates not supported in native columnar pipeline".to_string(),
                ));
            }
        };

        // Check if there's a GROUP BY with non-empty expressions
        let has_group_by = group_by.is_some_and(|exprs| !exprs.is_empty());
        let group_by_count = group_by.map_or(0, |exprs| exprs.len());

        // Count non-aggregate expressions in SELECT list (should match GROUP BY columns)
        let select_non_agg_count = select_items
            .iter()
            .filter(|item| {
                matches!(item, SelectItem::Expression { expr, .. }
                    if !matches!(expr, Expression::AggregateFunction { .. }))
            })
            .count();

        // Issue #4233: columnar_group_by_batch returns [group_keys..., aggregates...] format.
        // If the SELECT list doesn't include all GROUP BY columns (e.g., SELECT AVG(col) FROM t
        // GROUP BY col), the result format won't match the SELECT list. Fall back to
        // row-oriented execution.
        if has_group_by && select_non_agg_count != group_by_count {
            return Err(ExecutorError::UnsupportedFeature(format!(
                "GROUP BY with SELECT list that doesn't include all group keys not supported in native columnar (SELECT has {} non-aggs, GROUP BY has {} keys)",
                select_non_agg_count, group_by_count
            )));
        }

        // Issue #4510: columnar_group_by_batch returns [group_keys..., aggregates...] format.
        // If the SELECT list has aggregates before GROUP BY columns (e.g., SELECT count(*), col
        // FROM t GROUP BY col), the result order won't match. Fall back to row-oriented.
        // The SELECT list must have all non-aggregates first, then all aggregates.
        if has_group_by {
            let mut first_agg_pos = None;
            let mut last_non_agg_pos = None;
            for (i, item) in select_items.iter().enumerate() {
                if let SelectItem::Expression { expr, .. } = item {
                    if matches!(expr, Expression::AggregateFunction { .. }) {
                        if first_agg_pos.is_none() {
                            first_agg_pos = Some(i);
                        }
                    } else {
                        last_non_agg_pos = Some(i);
                    }
                }
            }
            // If first aggregate comes before last non-aggregate, order doesn't match columnar
            if let (Some(first_agg), Some(last_non_agg)) = (first_agg_pos, last_non_agg_pos) {
                if first_agg < last_non_agg {
                    log::debug!(
                        "Native columnar fallback: aggregate at {} before non-aggregate at {}",
                        first_agg,
                        last_non_agg
                    );
                    return Err(ExecutorError::UnsupportedFeature(
                        "SELECT list with aggregates before GROUP BY columns not supported in native columnar".to_string(),
                    ));
                }
            }
        }

        // Helper to return empty result for GROUP BY with empty input
        // SQL semantics: GROUP BY on empty input returns 0 rows (no groups)
        // Without GROUP BY: aggregates on empty input return 1 row (COUNT=0, others=NULL)
        let return_empty_result = |has_gb: bool, specs: &[AggregateSpec]| -> PipelineOutput {
            if has_gb {
                // GROUP BY with empty input: return 0 rows
                PipelineOutput::from_rows(vec![])
            } else {
                // No GROUP BY: return 1 row with COUNT=0, others=NULL
                let values: Vec<vibesql_types::SqlValue> = specs
                    .iter()
                    .map(|spec| match spec.op {
                        AggregateOp::Count => vibesql_types::SqlValue::Integer(0),
                        _ => vibesql_types::SqlValue::Null,
                    })
                    .collect();
                PipelineOutput::from_rows(vec![Row::new(values)])
            }
        };

        // Get batch directly if input is already columnar, otherwise convert
        // This is the key optimization: avoid row conversion if data is already in batch format
        let batch = match input {
            PipelineInput::Batch(batch) => batch,
            PipelineInput::Rows(rows) => {
                if rows.is_empty() {
                    return Ok(return_empty_result(has_group_by, &agg_specs));
                }
                ColumnarBatch::from_rows(rows)?
            }
            PipelineInput::RowsOwned(rows) => {
                if rows.is_empty() {
                    return Ok(return_empty_result(has_group_by, &agg_specs));
                }
                ColumnarBatch::from_rows(&rows)?
            }
            PipelineInput::Empty => {
                return Ok(return_empty_result(has_group_by, &agg_specs));
            }
            PipelineInput::NativeColumnar { table_name, .. } => {
                // Get columnar data directly from storage
                let columnar_table = match ctx.database.get_columnar(&table_name) {
                    Ok(Some(ct)) => ct,
                    Ok(None) | Err(_) => {
                        return Err(ExecutorError::Other(format!(
                            "Table '{}' not found for columnar aggregation",
                            table_name
                        )));
                    }
                };
                ColumnarBatch::from_storage_columnar(&columnar_table)?
            }
        };

        // Handle empty batch
        if batch.row_count() == 0 {
            return Ok(return_empty_result(has_group_by, &agg_specs));
        }

        // Check if we have GROUP BY
        if let Some(group_exprs) = group_by {
            if !group_exprs.is_empty() {
                return self.execute_group_by(&batch, group_exprs, &agg_specs, ctx);
            }
        }

        // Simple aggregation without GROUP BY
        let needs_schema =
            agg_specs.iter().any(|spec| matches!(spec.source, AggregateSource::Expression(_)));
        let schema_ref = if needs_schema { Some(ctx.schema) } else { None };

        let results =
            crate::select::columnar::compute_aggregates_from_batch(&batch, &agg_specs, schema_ref)?;

        Ok(PipelineOutput::from_rows(vec![Row::new(results)]))
    }

    /// Native columnar supports simple aggregates with optional GROUP BY.
    fn supports_query_pattern(
        &self,
        has_aggregation: bool,
        _has_group_by: bool,
        has_joins: bool,
    ) -> bool {
        // Native columnar supports:
        // - Aggregates with or without GROUP BY (single table only)
        // - No JOINs (single table requirement)
        // GROUP BY is supported for simple column references
        has_aggregation && !has_joins
    }

    #[inline]
    fn name(&self) -> &'static str {
        "NativeColumnarPipeline (SIMD)"
    }
}

impl NativeColumnarPipeline {
    /// Fallback to row-oriented filtering for complex predicates.
    fn fallback_filter(
        &self,
        rows: Vec<Row>,
        predicate: Option<&Expression>,
        ctx: &ExecutionContext<'_>,
    ) -> Result<PipelineOutput, ExecutorError> {
        if predicate.is_none() {
            return Ok(PipelineOutput::from_rows(rows));
        }

        let predicate = predicate.unwrap();
        let evaluator = ctx.create_evaluator();

        let mut filtered = Vec::with_capacity(rows.len());
        for row in rows {
            let value = evaluator.eval(predicate, &row)?;
            let include = match value {
                vibesql_types::SqlValue::Boolean(true) => true,
                vibesql_types::SqlValue::Boolean(false) | vibesql_types::SqlValue::Null => false,
                vibesql_types::SqlValue::Integer(0) => false,
                vibesql_types::SqlValue::Integer(_) => true,
                _ => false,
            };
            if include {
                filtered.push(row);
            }
        }

        Ok(PipelineOutput::from_rows(filtered))
    }

    /// Execute GROUP BY using hash-based columnar aggregation.
    fn execute_group_by(
        &self,
        batch: &ColumnarBatch,
        group_exprs: &[Expression],
        agg_specs: &[AggregateSpec],
        ctx: &ExecutionContext<'_>,
    ) -> Result<PipelineOutput, ExecutorError> {
        // Extract group column indices (only simple column refs supported)
        let group_cols: Vec<usize> = group_exprs
            .iter()
            .filter_map(|expr| match expr {
                Expression::ColumnRef(col_id) => {
                    ctx.schema.get_column_index(col_id.table_canonical(), col_id.column_canonical())
                }
                _ => None,
            })
            .collect();

        if group_cols.len() != group_exprs.len() {
            return Err(ExecutorError::UnsupportedFeature(
                "GROUP BY with non-column expressions not supported in native columnar".to_string(),
            ));
        }

        // Convert aggregates to (column_idx, op) format
        let agg_cols: Vec<(usize, AggregateOp)> = agg_specs
            .iter()
            .filter_map(|spec| match &spec.source {
                AggregateSource::Column(idx) => Some((*idx, spec.op)),
                AggregateSource::CountStar => Some((0, AggregateOp::Count)),
                AggregateSource::Expression(_) => None,
            })
            .collect();

        if agg_cols.len() != agg_specs.len() {
            return Err(ExecutorError::UnsupportedFeature(
                "Expression aggregates not supported in native columnar GROUP BY".to_string(),
            ));
        }

        // Execute SIMD-accelerated hash-based GROUP BY directly on batch
        let result =
            crate::select::columnar::columnar_group_by_batch(batch, &group_cols, &agg_cols)?;

        Ok(PipelineOutput::from_rows(result))
    }
}

#[cfg(test)]
mod tests {
    use vibesql_catalog::TableSchema;
    use vibesql_types::SqlValue;

    use super::*;
    use crate::schema::CombinedSchema;

    fn create_test_setup() -> (vibesql_storage::Database, CombinedSchema) {
        let database = vibesql_storage::Database::new();
        let table_schema = TableSchema::new("test".to_string(), vec![]);
        let schema = CombinedSchema::from_table("test".to_string(), table_schema);
        (database, schema)
    }

    fn make_test_row(values: Vec<i64>) -> Row {
        Row::new(values.into_iter().map(SqlValue::Integer).collect::<Vec<_>>())
    }

    #[test]
    fn test_native_columnar_pipeline_name() {
        let pipeline = NativeColumnarPipeline::new();
        let name = pipeline.name();
        assert!(name.starts_with("NativeColumnarPipeline"));
    }

    #[test]
    fn test_native_columnar_pipeline_supports_aggregates() {
        let pipeline = NativeColumnarPipeline::new();

        // Supports aggregates without joins (SIMD always enabled via auto-vectorization)
        let supports_simple = pipeline.supports_query_pattern(true, false, false);
        assert!(supports_simple);

        // Does not support joins
        assert!(!pipeline.supports_query_pattern(true, false, true));
    }

    #[test]
    fn test_native_columnar_pipeline_filter_no_predicate() {
        let (database, schema) = create_test_setup();
        let pipeline = NativeColumnarPipeline::new();

        let rows = vec![make_test_row(vec![1, 2]), make_test_row(vec![3, 4])];
        let input = PipelineInput::from_rows_owned(rows);
        let ctx = ExecutionContext::new(&schema, &database);

        let result = pipeline.apply_filter(input, None, &ctx).unwrap();
        assert_eq!(result.row_count(), 2);
    }

    #[test]
    fn test_native_columnar_pipeline_default() {
        let pipeline = NativeColumnarPipeline::default();
        assert!(pipeline.name().starts_with("NativeColumnarPipeline"));
    }

    #[test]
    fn test_native_columnar_pipeline_limit_offset() {
        let pipeline = NativeColumnarPipeline::new();

        let rows = vec![
            make_test_row(vec![1]),
            make_test_row(vec![2]),
            make_test_row(vec![3]),
            make_test_row(vec![4]),
            make_test_row(vec![5]),
        ];
        let output = PipelineOutput::from_rows(rows);

        // Test limit
        let result = pipeline.apply_limit_offset(output, Some(3), None).unwrap();
        assert_eq!(result.len(), 3);
    }
}