featherdb_query/executor/

mod.rs

//! Query executor - executes logical plans

mod aggregate;
mod cte;
mod ddl;
mod dml;
mod explain;
mod filter;
mod fk;
mod join;
mod project;
mod scan;
mod sort;
mod subquery;
mod types;
mod window;

// Re-export public types
pub use types::{ExecutionContext, QueryResult, Row};

use crate::planner::LogicalPlan;
use featherdb_core::{Error, Permissions, Result};

/// Query executor
pub struct Executor;

impl Executor {
    /// Check if the current context has permission to perform an operation on a table
    ///
    /// This is the central permission check used by all scan and DML operations.
    /// When auth is disabled, all operations are allowed.
    /// When auth is enabled, the API key must have the required permission on the table.
    pub(crate) fn check_table_permission(
        ctx: &ExecutionContext,
        table_name: &str,
        required: Permissions,
    ) -> Result<()> {
        // If auth is not enabled, allow all operations
        if !ctx.auth_enabled {
            return Ok(());
        }

        // Get the API key (must exist if auth is enabled)
        let api_key_id = ctx.api_key_id.ok_or(Error::AuthenticationRequired)?;

        // Check permission via catalog
        if !ctx
            .catalog
            .check_permission(table_name, api_key_id, required)
        {
            return Err(Error::PermissionDenied {
                table: table_name.to_string(),
                operation: required.to_string(),
                api_key_id: api_key_id.to_string(),
            });
        }

        Ok(())
    }
}

impl Executor {
    /// Execute a logical plan
    pub fn execute(ctx: &ExecutionContext, plan: &LogicalPlan) -> Result<QueryResult> {
        match plan {
            LogicalPlan::Scan {
                table,
                alias,
                projection,
                filter,
            } => Self::execute_scan(
                ctx,
                table,
                alias.as_deref(),
                projection.as_deref(),
                filter.as_ref(),
            ),

            LogicalPlan::IndexScan {
                table,
                alias,
                index,
                index_column,
                range,
                residual_filter,
                projection,
            } => Self::execute_index_scan(
                ctx,
                table,
                alias.as_deref(),
                index,
                *index_column,
                range,
                residual_filter.as_ref(),
                projection.as_deref(),
            ),

            LogicalPlan::PkSeek {
                table,
                alias,
                key_values,
                residual_filter,
                projection,
            } => Self::execute_pk_seek(
                ctx,
                table,
                alias.as_deref(),
                key_values,
                residual_filter.as_ref(),
                projection.as_deref(),
            ),

            LogicalPlan::PkRangeScan {
                table,
                alias,
                range,
                residual_filter,
                projection,
            } => Self::execute_pk_range_scan(
                ctx,
                table,
                alias.as_deref(),
                range,
                residual_filter.as_ref(),
                projection.as_deref(),
            ),

            LogicalPlan::Filter { input, predicate } => Self::execute_filter(ctx, input, predicate),

            LogicalPlan::Project { input, exprs } => Self::execute_project(ctx, input, exprs),

            LogicalPlan::Join {
                left,
                right,
                join_type,
                condition,
            } => Self::execute_join(ctx, left, right, *join_type, condition.as_ref()),

            LogicalPlan::Aggregate {
                input,
                group_by,
                aggregates,
            } => Self::execute_aggregate(ctx, input, group_by, aggregates),

            LogicalPlan::Sort { input, order_by } => Self::execute_sort(ctx, input, order_by),

            LogicalPlan::Limit {
                input,
                limit,
                offset,
            } => Self::execute_limit(ctx, input, *limit, *offset),

            LogicalPlan::Distinct { input } => Self::execute_distinct(ctx, input),

            LogicalPlan::Insert {
                table,
                columns,
                values,
            } => Self::execute_insert(ctx, table, columns, values),

            LogicalPlan::Update {
                table,
                assignments,
                filter,
            } => Self::execute_update(ctx, table, assignments, filter.as_ref()),

            LogicalPlan::Delete { table, filter } => {
                Self::execute_delete(ctx, table, filter.as_ref())
            }

            LogicalPlan::CreateTable { table } => Self::execute_create_table(ctx, table),

            LogicalPlan::DropTable { name, if_exists } => {
                Self::execute_drop_table(ctx, name, *if_exists)
            }

            LogicalPlan::EmptyRelation => Ok(QueryResult::Rows(vec![Row::new(vec![], vec![])])),

            LogicalPlan::Explain {
                input,
                verbose,
                analyze,
            } => Self::execute_explain(ctx, input, *verbose, *analyze),

            LogicalPlan::AlterTable {
                table_name,
                operations,
            } => Self::execute_alter_table(ctx, table_name, operations),

            LogicalPlan::Window {
                input,
                window_exprs,
            } => Self::execute_window(ctx, input, window_exprs),

            LogicalPlan::WithCte { ctes, query } => Self::execute_with_cte(ctx, ctes, query),

            LogicalPlan::CteScan {
                cte_name,
                alias,
                columns: _,
            } => {
                // Retrieve materialized CTE results from context
                match ctx.cte_context.get(cte_name) {
                    Some(rows) => {
                        // If there's an alias, re-prefix all column names
                        let prefix = alias.as_ref().unwrap_or(cte_name);
                        let re_prefixed: Vec<Row> = rows
                            .iter()
                            .map(|row| {
                                let new_cols: Vec<String> = row
                                    .columns
                                    .iter()
                                    .map(|c| {
                                        let bare = c.split('.').next_back().unwrap_or(c);
                                        format!("{}.{}", prefix, bare)
                                    })
                                    .collect();
                                Row::new(row.values.clone(), new_cols)
                            })
                            .collect();
                        Ok(QueryResult::Rows(re_prefixed))
                    }
                    None => Err(Error::Internal(format!(
                        "CTE '{}' referenced but not in scope",
                        cte_name
                    ))),
                }
            }

            LogicalPlan::SubqueryScan {
                subquery, alias, ..
            } => Self::execute_subquery_scan(ctx, subquery, alias.as_deref()),

            LogicalPlan::SemiJoin {
                left,
                right,
                condition,
                positive,
            } => Self::execute_semi_join(ctx, left, right, condition.as_ref(), *positive),

            LogicalPlan::AntiJoin {
                left,
                right,
                condition,
            } => Self::execute_anti_join(ctx, left, right, condition.as_ref()),

            LogicalPlan::Grant {
                privileges,
                table,
                grantee,
            } => Self::execute_grant(ctx, *privileges, table.as_deref(), grantee),

            LogicalPlan::Revoke {
                privileges,
                table,
                grantee,
            } => Self::execute_revoke(ctx, *privileges, table.as_deref(), grantee),

            LogicalPlan::ShowGrants { target } => Self::execute_show_grants(ctx, target),

            LogicalPlan::Union { left, right, all } => Self::execute_union(ctx, left, right, *all),

            LogicalPlan::Intersect { left, right, all } => {
                Self::execute_intersect(ctx, left, right, *all)
            }

            LogicalPlan::Except { left, right, all } => {
                Self::execute_except(ctx, left, right, *all)
            }

            LogicalPlan::CreateView {
                name,
                query_sql,
                columns,
                ..
            } => Self::execute_create_view(ctx, name, query_sql, columns),

            LogicalPlan::DropView { name, if_exists } => {
                Self::execute_drop_view(ctx, name, *if_exists)
            }

            LogicalPlan::CreateIndex {
                index_name,
                table_name,
                columns,
                unique,
            } => Self::execute_create_index(ctx, index_name, table_name, columns, *unique),
        }
    }

    /// Execute a logical plan and track execution time metrics
    ///
    /// This wraps the execute method to record execution time when metrics are enabled.
    pub fn execute_with_timing(ctx: &ExecutionContext, plan: &LogicalPlan) -> Result<QueryResult> {
        if let Some(metrics) = &ctx.metrics {
            let start = std::time::Instant::now();
            let result = Self::execute(ctx, plan)?;
            let elapsed = start.elapsed().as_micros() as u64;

            metrics.record_exec_time(elapsed);

            // Track rows returned for SELECT queries
            if let QueryResult::Rows(ref rows) = result {
                metrics.record_rows_returned(rows.len() as u64);
            }

            Ok(result)
        } else {
            // No metrics, just execute normally
            Self::execute(ctx, plan)
        }
    }
}