krishiv_sql/

lib.rs

#![forbid(unsafe_code)]

//! SQL planning and local execution seam for Krishiv.
//!
//! This crate owns the DataFusion integration for R1 while keeping DataFusion
//! out of the long-term public API exposed by `krishiv-api`.

use std::collections::{BTreeSet, HashMap, VecDeque};
use std::fmt;
use std::num::NonZeroUsize;
use std::ops::ControlFlow;
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};

use arrow::datatypes::SchemaRef;
use arrow::record_batch::RecordBatch;
use arrow::util::pretty::pretty_format_batches;
use catalog::{InMemoryCatalog, datafusion_bridge::DataFusionCatalogBridge};
use datafusion::dataframe::DataFrame as DataFusionDataFrame;
use datafusion::prelude::{ParquetReadOptions, SessionContext};
use datafusion::sql::sqlparser::{ast::visit_relations, dialect::GenericDialect, parser::Parser};
use object_store::aws::AmazonS3Builder;

use krishiv_plan::optimizer::{CostModel, Optimizer};
use krishiv_plan::{ExecutionKind, LogicalPlan, PlanNode};

/// Map an Arrow type name (as shipped in a Python-UDF directive) to a DataType.
/// Unknown names fall back to Utf8 (the safest catch-all for a scalar result).
fn python_udf_arrow_type(name: &str) -> arrow::datatypes::DataType {
    use arrow::datatypes::DataType;
    match name.trim().to_ascii_lowercase().as_str() {
        "double" | "float64" | "float" => DataType::Float64,
        "float32" | "real" => DataType::Float32,
        "int" | "int64" | "bigint" | "long" => DataType::Int64,
        "int32" | "integer" => DataType::Int32,
        "bool" | "boolean" => DataType::Boolean,
        "utf8" | "string" | "varchar" | "text" => DataType::Utf8,
        _ => DataType::Utf8,
    }
}

/// Build an `object_store` S3 client for `bucket` from the ambient AWS
/// environment (`AWS_ENDPOINT_URL` for MinIO, credentials, region).
///
/// Shared by the Iceberg FileIO [`catalog::object_store_io::KrishivStorage`]
/// (metadata reads/writes) *and* DataFusion's object-store registry (Parquet
/// data scans via `ListingTable`) so both hit the *same* S3/MinIO backend.
/// Reading `AWS_ENDPOINT_URL` here — the AWS-SDK convention prod sets — is what
/// makes MinIO reachable; `AmazonS3Builder::from_env` alone honours only
/// `AWS_ENDPOINT` and would silently target real AWS.
pub(crate) fn build_s3_object_store(
    bucket: &str,
) -> object_store::Result<std::sync::Arc<dyn object_store::ObjectStore>> {
    let mut builder = AmazonS3Builder::from_env().with_bucket_name(bucket);
    let mut has_endpoint = false;
    if let Ok(endpoint) = std::env::var("AWS_ENDPOINT_URL")
        && !endpoint.is_empty()
    {
        // MinIO / S3-compatible: path-style access over plain HTTP.
        builder = builder.with_endpoint(endpoint).with_allow_http(true);
        has_endpoint = true;
    }
    let has_key = std::env::var("AWS_ACCESS_KEY_ID")
        .map(|k| !k.is_empty())
        .unwrap_or(false);
    if let Ok(key) = std::env::var("AWS_ACCESS_KEY_ID")
        && !key.is_empty()
    {
        builder = builder.with_access_key_id(key);
    }
    if let Ok(secret) = std::env::var("AWS_SECRET_ACCESS_KEY")
        && !secret.is_empty()
    {
        builder = builder.with_secret_access_key(secret);
    }
    // A custom endpoint with no credentials means MinIO / an S3-compatible store
    // reached anonymously — NOT EC2. Skip request signing so the client reads
    // public objects directly instead of blocking for ~180s trying to fetch
    // instance credentials from the IMDS endpoint (169.254.169.254), which is
    // unreachable off-EC2. Real-AWS deployments (no endpoint) keep the default
    // credential chain, so EC2 instance-role auth is unaffected.
    if has_endpoint && !has_key {
        builder = builder.with_skip_signature(true);
    }
    let region = std::env::var("AWS_REGION")
        .or_else(|_| std::env::var("AWS_DEFAULT_REGION"))
        .unwrap_or_else(|_| "us-east-1".to_string());
    builder = builder.with_region(region);
    Ok(std::sync::Arc::new(builder.build()?))
}

pub mod analyze;
pub mod catalog;
pub mod cep_sql;

pub mod connector_table;
pub mod coop_amplifiers;
pub mod create_function_ddl;
pub mod distributed_plan;
pub mod grace_hash_join;
pub mod grammar;
pub mod incremental_view;
pub mod introspection_sql;

pub mod kafka_table;
pub mod lakehouse;
pub mod late_materialize;
pub mod live_table;
/// One reading of a join build side's size estimate, shared by the broadcast
/// override and the spillable-join choice so the two cannot disagree.
pub(crate) mod join_estimates;
pub mod object_store_registry;
pub mod pipeline_ddl;
pub mod pipe_syntax;
pub mod pivot_sql;
pub mod python_udf;
pub mod scalar_udf;
pub mod semi_join_reduction;
/// Spark SQL extensions: LATERAL VIEW, TABLESAMPLE, TRANSFORM, DESCRIBE EXTENDED, etc.
pub mod spark_sql_ext;
pub mod runtime_filter_exec;
pub mod spillable_join;
pub mod sqlstate;
pub mod subquery;
pub mod unnest_sql;
pub mod unspillable_headroom;

pub mod coverage;
mod higher_order_functions;
mod json_functions;
mod spark_functions;
pub mod statement_completion;
pub mod streaming;
pub mod streaming_table_ddl;
pub mod streaming_tvf;
pub mod streaming_window_plan;
mod udf;
mod window_functions;

pub use cep_sql::{
    MatchRecognizeStatement, execute_streaming_match_recognize, parse_match_recognize,
};
pub use lakehouse::{AsOfTableRef, MergeResult, MergeTargetUnsupportedError, preprocess_as_of_sql};

pub use grammar::{
    FeatureEntry, FeatureStatus, feature_matrix, features_by_status, features_for_category,
};
pub use sqlstate::{SqlStateError, sqlstate_for};
pub use streaming::{ContinuousInputError, ContinuousTableInput};

/// SQL result alias.
pub type SqlResult<T> = Result<T, SqlError>;

/// Pinned stream of record batches with typed [`SqlError`] items.
///
/// Previously this used `String` as the error type, which lost diagnostic
/// information at the stream boundary. Callers that need a `String` error can
/// map with `|e| e.to_string()`.
pub type SqlStream =
    std::pin::Pin<Box<dyn futures::stream::Stream<Item = Result<RecordBatch, SqlError>> + Send>>;

/// Global counter for unique ephemeral table names, preventing concurrent
/// MERGE/CEP queries from overwriting each other's result tables.
static EPHEMERAL_TABLE_COUNTER: AtomicU64 = AtomicU64::new(0);

fn next_ephemeral_name(prefix: &str) -> String {
    let id = EPHEMERAL_TABLE_COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("__{prefix}_{id}")
}

// ── Plan cache (single-lock, race-free) ──────────────────────────────────────

/// Whether the [`SqlEngine`] internal builder should attempt to register the
/// helper window UDFs (`tumble_start` / `tumble_end` / `hop_start` / `hop_end`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum WindowFnRegistration {
    /// Call `window_functions::register_window_functions`; propagate any error.
    Register,
    /// Skip registration entirely; infallible. Used as a fallback by
    /// [`SqlEngine::new`] when `Register` fails so the engine is still usable
    /// for non-window queries.
    Skip,
}

/// Bounded query-plan cache keyed by query text.
///
/// A single `Mutex<PlanCache>` replaces the previous two-structure approach
/// (`DashMap` + `Mutex<VecDeque>`) which had a TOCTOU race: two threads could
/// both see `len() < MAX` and both insert, growing the cache past the limit.
struct PlanCache {
    map: HashMap<String, (datafusion::logical_expr::LogicalPlan, std::time::Instant)>,
    order: VecDeque<String>,
    max: usize,
}

/// How long a cached logical plan stays valid.
///
/// A cached plan pins the table providers resolved at planning time —
/// including the exact data-file listing of Iceberg catalog tables. Local
/// writes clear the cache, but a table replaced by ANOTHER engine (a
/// coordinator-mode executor landing a pipeline pull, a second daemon) is
/// invisible here, and without a TTL a repeated query text served stale
/// results forever (found live 2026-07-19: the jdbc-pull e2e's COUNT read
/// 3 rows from a snapshot the executor had already replaced with 5).
/// 30s matches CATALOG_CACHE_TTL — the staleness bound the catalog layer
/// already accepts.
const PLAN_CACHE_TTL: std::time::Duration = std::time::Duration::from_secs(30);

impl PlanCache {
    fn new(max: usize) -> Self {
        Self {
            map: HashMap::new(),
            order: VecDeque::new(),
            max,
        }
    }

    /// Look up a plan, promoting it to most-recently-used on a hit.
    ///
    /// The promotion is what makes eviction LRU. Without it — `get(&self)` —
    /// `order` only ever reflects insertion, so eviction is FIFO and a burst of
    /// distinct query texts evicts the hot repeated ones that are the entire
    /// reason the cache exists. An expired entry is dropped here rather than
    /// left to age out, so a dead `LogicalPlan` is not held until eviction.
    fn get(&mut self, key: &str) -> Option<&datafusion::logical_expr::LogicalPlan> {
        match self.map.get(key).map(|(_, at)| at.elapsed() < PLAN_CACHE_TTL) {
            None => None,
            Some(false) => {
                self.map.remove(key);
                self.order.retain(|k| k != key);
                None
            }
            Some(true) => {
                if let Some(pos) = self.order.iter().position(|k| k == key)
                    && let Some(promoted) = self.order.remove(pos)
                {
                    self.order.push_back(promoted);
                }
                self.map.get(key).map(|(plan, _)| plan)
            }
        }
    }

    fn insert(&mut self, key: String, plan: datafusion::logical_expr::LogicalPlan) {
        if self.map.contains_key(&key) {
            // Remove the stale order entry so a repeated insert doesn't accumulate
            // duplicate references and corrupt LRU eviction order.
            self.order.retain(|k| k != &key);
        } else if self.map.len() >= self.max
            && let Some(oldest) = self.order.pop_front()
        {
            self.map.remove(&oldest);
        }
        self.order.push_back(key.clone());
        self.map.insert(key, (plan, std::time::Instant::now()));
    }

    fn clear(&mut self) {
        self.map.clear();
        self.order.clear();
    }

    #[cfg(test)]
    fn is_empty(&self) -> bool {
        self.map.is_empty()
    }
}

/// Typed options for Parquet reads (propagated into DataFusion).
#[derive(Debug, Clone, Default)]
pub struct ParquetReaderOptions {
    /// Maximum number of rows per output batch (None = DataFusion default 8192).
    pub batch_size: Option<usize>,
}

/// Typed options for CSV reads (propagated into DataFusion).
#[derive(Debug, Clone, Default)]
pub struct CsvReaderOptions {
    /// Field delimiter character (None = `,`).
    pub delimiter: Option<char>,
    /// Whether the first row is a header (None = true).
    pub has_header: Option<bool>,
}

/// Typed options for Parquet writes (propagated into the `ArrowWriter`).
#[derive(Debug, Clone, Default)]
pub struct ParquetWriterOptions {
    /// Compression codec: "snappy" | "zstd" | "gzip" | "lz4" | "brotli" | "uncompressed".
    pub compression: Option<String>,
    /// Maximum number of rows per row-group (None = `ArrowWriter` default 1 048 576).
    pub max_row_group_size: Option<usize>,
}

/// Typed options for CSV writes.
#[derive(Debug, Clone, Default)]
pub struct CsvWriterOptions {
    /// Field delimiter character (None = `,`).
    pub delimiter: Option<char>,
    /// Whether to emit a header row (None = true).
    pub has_header: Option<bool>,
}

/// SQL-layer errors.
#[non_exhaustive]
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum SqlError {
    /// Query was empty or whitespace only.
    #[error("SQL query is empty")]
    EmptyQuery,
    /// A table name was empty.
    #[error("table name is empty")]
    EmptyTableName,
    /// The requested SQL feature is not available in R1.
    #[error("unsupported SQL feature: {feature}")]
    Unsupported { feature: String },
    /// A table-function declaration or runtime registration was invalid.
    #[error("invalid table function: {message}")]
    InvalidTableFunction { message: String },
    /// DataFusion returned an error.
    #[error("DataFusion error: {message}")]
    DataFusion { message: String },
    /// Krishiv logical-plan optimization failed.
    #[error(transparent)]
    Optimizer(#[from] krishiv_plan::optimizer::OptimizerError),
    /// Access denied by auth or policy check.
    #[error("access denied: {reason}")]
    AccessDenied { reason: String },
    /// A running operation was cancelled by the caller.
    #[error("operation {operation_id} was cancelled")]
    OperationCancelled { operation_id: u64 },
    /// A query exceeded its configured execution timeout.
    #[error("query timed out after {timeout_ms} ms")]
    Timeout { timeout_ms: u64 },
}

impl From<datafusion::error::DataFusionError> for SqlError {
    fn from(value: datafusion::error::DataFusionError) -> Self {
        Self::DataFusion {
            message: value.to_string(),
        }
    }
}

/// SQL planning output.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SqlPlan {
    query: String,
    logical_plan: LogicalPlan,
}

impl SqlPlan {
    /// Original query.
    pub fn query(&self) -> &str {
        &self.query
    }

    /// Krishiv logical plan wrapper.
    pub fn logical_plan(&self) -> &LogicalPlan {
        &self.logical_plan
    }
}

/// Maximum number of query plans stored in the plan cache before LRU eviction.
const PLAN_CACHE_MAX_ENTRIES: usize = 256;

fn resolve_plan_cache_max_entries() -> usize {
    std::env::var("KRISHIV_PLAN_CACHE_MAX_ENTRIES")
        .ok()
        .and_then(|v| v.parse().ok())
        .filter(|&n| n > 0)
        .unwrap_or(PLAN_CACHE_MAX_ENTRIES)
}
const STREAMING_CEP_MAX_ROWS_DEFAULT: usize = 100_000;

/// Resolve the streaming MATCH_RECOGNIZE row cap from a raw env var value.
/// `None` and unparseable values fall back to the documented default of
/// 100_000. Zero is rejected because it would mean "scan zero rows".
pub fn resolve_streaming_match_recognize_limit(raw: Option<&str>) -> usize {
    raw.and_then(|s| s.parse::<usize>().ok())
        .filter(|n| *n > 0)
        .unwrap_or(STREAMING_CEP_MAX_ROWS_DEFAULT)
}

/// Resolve the streaming MATCH_RECOGNIZE row cap from the
/// `KRISHIV_MATCH_RECOGNIZE_STREAMING_LIMIT` environment variable.
pub fn streaming_match_recognize_limit_from_env() -> usize {
    resolve_streaming_match_recognize_limit(
        std::env::var("KRISHIV_MATCH_RECOGNIZE_STREAMING_LIMIT")
            .ok()
            .as_deref(),
    )
}

/// Resolve a per-engine DataFusion memory limit from a raw env var value.
/// `None`, unparseable, and zero values all mean "no limit" (the engine runs
/// with DataFusion's default unbounded pool).
pub fn resolve_query_memory_limit_bytes(raw: Option<&str>) -> Option<usize> {
    raw.and_then(|s| s.trim().parse::<usize>().ok())
        .filter(|n| *n > 0)
}

/// Resolve the default per-engine memory limit from the
/// `KRISHIV_QUERY_MEMORY_LIMIT_BYTES` environment variable, falling back to
/// a cgroup-derived default when the variable is unset.
///
/// Fallback: 25% of the container's cgroup memory limit (v2 `memory.max`,
/// v1 `memory.limit_in_bytes`). Several engines can be live in one process
/// (executor task slots, the Flight SQL host, IVM ticks), so a single
/// engine's pool must not claim the whole container. Explicit `0` disables
/// the limit entirely (DataFusion's default unbounded pool); an unlimited
/// cgroup (no limit / `max`) also yields `None`.
///
/// # This is the third copy of this decision, and it has no callers
///
/// Audited 2026-08-09: **zero references** across the workspace (both `*.rs`
/// and `*.rs.inc`). The live sizing path is
/// `krishiv_common::ExecutorCapacity`, which every engine constructor uses,
/// and `krishiv_ivm::spill::ivm_memory_limit_bytes` carries a byte-identical
/// third copy for the IVM tick.
///
/// Kept rather than deleted because it is `pub` and this crate's API is
/// consumed outside the workspace, but **do not add callers**: reach for
/// `ExecutorCapacity`, which also knows the slot count and so can answer
/// "this engine's share" rather than only "the process's ceiling". Three
/// independent readings of one env var is exactly how the spill threshold
/// and the bucket count came to disagree about what a byte count meant.
pub fn query_memory_limit_from_env() -> Option<usize> {
    match std::env::var("KRISHIV_QUERY_MEMORY_LIMIT_BYTES").ok() {
        // Set (including "0" and garbage): explicit-config semantics — no
        // cgroup fallback, unparseable/zero means unlimited.
        Some(raw) => resolve_query_memory_limit_bytes(Some(&raw)),
        None => cgroup_memory_limit_bytes()
            .map(|limit| (limit / 4) as usize)
            .filter(|&n| n > 0),
    }
}

pub use krishiv_common::cgroup_memory_limit_bytes;

/// DataFusion's memory-pool trait, re-exported so crates that only build
/// engines (the executor) can hold a shared pool without depending on
/// DataFusion directly.
pub use datafusion::execution::memory_pool::MemoryPool;

/// The consumer/reservation half of the same pool API.
///
/// Re-exported for the same reason as [`MemoryPool`]: the executor's map-side
/// shuffle-write buffer has to reserve its bytes through the task engine's
/// pool — a buffer the pool cannot see is a buffer the container limit cannot
/// govern — and it should not take a DataFusion dependency to do it.
pub use datafusion::execution::memory_pool::{MemoryConsumer, MemoryReservation};

/// The record-batch stream trait, re-exported for the same reason as
/// [`MemoryPool`].
///
/// A fragment's real output schema is the one its *stream* declares, not the
/// one its `DataFrame` does — physical planning re-types expressions, and
/// TPC-H q17 shipped batches of `Decimal128(30, 15)` under a logical schema
/// that said `Decimal128(15, 2)`. Reading it needs this trait in scope, and
/// the executor should not take a DataFusion dependency to do that.
pub use datafusion::physical_plan::RecordBatchStream;

/// The one query memory pool for this process, sized by
/// [`krishiv_common::ExecutorCapacity`] from the cgroup limit.
///
/// `None` when execution memory is unbounded: no cgroup limit (bare metal or
/// an unconstrained container), or an explicit
/// `KRISHIV_QUERY_MEMORY_LIMIT_BYTES=0`.
///
/// Built once. Every engine that does not carry an explicit per-job limit
/// draws on it, so the total execution memory of the process is this number
/// regardless of how many engines exist.
pub fn process_query_pool() -> Option<&'static Arc<dyn MemoryPool>> {
    static POOL: std::sync::LazyLock<Option<Arc<dyn MemoryPool>>> =
        std::sync::LazyLock::new(|| {
            let capacity = krishiv_common::ExecutorCapacity::detect();
            let bytes = capacity.query_pool_bytes?;
            tracing::info!(
                capacity = %capacity.summary(),
                "query memory: one shared FairSpillPool for this process"
            );
            Some(EngineMemory::shared_pool(
                usize::try_from(bytes).unwrap_or(usize::MAX),
            ))
        });
    POOL.as_ref()
}

/// A `FairSpillPool` of `bytes`, wrapped so spillable consumers cannot occupy
/// the whole of it.
///
/// The single constructor for every bounded pool in this crate. Bare
/// `FairSpillPool` lets N spillable consumers — each politely inside its own
/// fair share — take the entire budget between them, after which an operator
/// that *cannot* spill is refused even a few hundred bytes and nothing can
/// reclaim anything. TPC-H q10 and q11 died that way at SF100 (877 B refused by
/// a 2.3 GB pool). See [`crate::unspillable_headroom`].
///
/// Both `Private` and `Shared` route through here deliberately: the first
/// version guarded only the shared pool, and every executor task engine is
/// `Private`, so the protection was absent from the one deployment that needed
/// it. One constructor is what keeps that from being possible.
fn guarded_fair_pool(bytes: usize) -> Arc<dyn MemoryPool> {
    let inner: Arc<dyn MemoryPool> =
        Arc::new(datafusion::execution::memory_pool::FairSpillPool::new(bytes));
    Arc::new(crate::unspillable_headroom::UnspillableHeadroomPool::new(
        inner,
        bytes,
        crate::unspillable_headroom::headroom_bytes(bytes),
    ))
}

/// One engine's expected share of [`process_query_pool`] when every slot is
/// busy. Sizes spill reservations only; the pool itself is shared.
///
fn process_query_pool_fair_share_bytes() -> usize {
    krishiv_common::ExecutorCapacity::detect()
        .min_task_memory_share_bytes()
        .map_or(usize::MAX, |bytes| {
            usize::try_from(bytes).unwrap_or(usize::MAX)
        })
}

/// Where an engine's DataFusion execution memory comes from.
///
/// The distinction that matters is [`Private`](EngineMemory::Private) versus
/// [`Shared`](EngineMemory::Shared). An executor runs `slots` task fragments
/// at once; giving each its own pool means the process claims
/// `slots × pool_size`, which is how a container gets OOM-killed while every
/// individual pool still reports headroom. One pool shared by every task makes
/// the executor's total execution memory a single hard number, and
/// `FairSpillPool` divides it live across whoever is actually running — a task
/// alone gets all of it, four tasks get a quarter each, and nobody has to
/// predict the concurrency in advance.
#[derive(Clone)]
pub enum EngineMemory {
    /// DataFusion's default unbounded pool: no accounting, no spill.
    Unbounded,
    /// A `FairSpillPool` of this size belonging to this engine alone.
    /// Correct for one-engine-per-process deployments (embedded, gateway).
    Private(usize),
    /// A pool shared with every other engine in this process.
    ///
    /// `fair_share_bytes` is what this engine can expect when all slots are
    /// busy. It sizes spill *reservations* only — the pool itself is the
    /// shared object and is not bounded by this number.
    Shared {
        /// The process-wide pool.
        pool: Arc<dyn datafusion::execution::memory_pool::MemoryPool>,
        /// This engine's expected share, for reservation sizing.
        fair_share_bytes: usize,
    },
}

impl EngineMemory {
    /// A private pool of `bytes`, or [`Unbounded`](EngineMemory::Unbounded)
    /// when there is no limit.
    #[must_use]
    pub fn from_limit(bytes: Option<usize>) -> Self {
        bytes.map_or(Self::Unbounded, Self::Private)
    }

    /// Build a pool of `bytes` intended to be shared by several engines.
    ///
    /// The caller keeps the returned handle and passes clones of it in
    /// [`EngineMemory::Shared`], which is what bounds their combined execution
    /// memory rather than each engine's individually.
    #[must_use]
    pub fn shared_pool(bytes: usize) -> Arc<dyn MemoryPool> {
        guarded_fair_pool(bytes)
    }

    /// The default memory source for an engine built in this process: a share
    /// of [`process_query_pool`].
    ///
    /// This is what makes the bound hold no matter how many engines a process
    /// ends up with. A process can host the Flight SQL engine, several
    /// executor task slots, and IVM tick engines at once; when each built its
    /// own pool at a fraction of the container, the fractions summed past the
    /// container and the container was OOM-killed while every pool still
    /// reported headroom. One shared pool cannot oversubscribe, and
    /// `FairSpillPool` divides it live — an engine running alone gets all of
    /// it, so the bound costs nothing when there is no contention.
    #[must_use]
    pub fn for_this_process() -> Self {
        match process_query_pool() {
            Some(pool) => Self::Shared {
                pool: Arc::clone(pool),
                fair_share_bytes: process_query_pool_fair_share_bytes(),
            },
            None => Self::Unbounded,
        }
    }

    /// The byte figure that should size spill reservations and session config,
    /// or `None` when execution memory is unbounded.
    #[must_use]
    pub fn sizing_bytes(&self) -> Option<usize> {
        match self {
            Self::Unbounded => None,
            Self::Private(bytes) => Some(*bytes),
            Self::Shared {
                fair_share_bytes, ..
            } => Some(*fair_share_bytes),
        }
    }

    /// The pool to install on the runtime, or `None` for DataFusion's default.
    fn pool(&self) -> Option<Arc<dyn datafusion::execution::memory_pool::MemoryPool>> {
        match self {
            Self::Unbounded => None,
            // Guarded, like `shared_pool` — and this is the path that matters
            // on the cluster. Every executor task engine is built with
            // `EngineMemory::Private` (krishiv-executor `task_sql_engine`), so
            // wrapping only the shared pool would have left the headroom
            // switched off in exactly the deployment q10/q11 failed in.
            Self::Private(bytes) => Some(guarded_fair_pool(*bytes)),
            Self::Shared { pool, .. } => Some(Arc::clone(pool)),
        }
    }
}

impl fmt::Debug for EngineMemory {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Unbounded => f.write_str("EngineMemory::Unbounded"),
            Self::Private(bytes) => write!(f, "EngineMemory::Private({bytes})"),
            Self::Shared {
                fair_share_bytes, ..
            } => write!(f, "EngineMemory::Shared(share={fair_share_bytes})"),
        }
    }
}

/// Programmatic override for [`runtime_filters_enabled_from_env`]
/// (`u8::MAX` = unset). Exists because `std::env::set_var` is `unsafe`
/// under edition 2024 and the workspace forbids unsafe code, so the
/// corpus dual-run (AQE/runtime-filters off) cannot toggle the env var.
static RUNTIME_FILTERS_OVERRIDE: std::sync::atomic::AtomicU8 =
    std::sync::atomic::AtomicU8::new(u8::MAX);

/// Test/diagnostic hook: force runtime filters on (`true`) or off (`false`)
/// for every engine built in this process afterwards.
#[doc(hidden)]
pub fn set_runtime_filters_for_tests(enabled: bool) {
    RUNTIME_FILTERS_OVERRIDE.store(u8::from(enabled), std::sync::atomic::Ordering::Relaxed);
}

/// Phase 54: DataFusion's native dynamic ("runtime") filters — TopK, join,
/// and aggregate predicates pushed sideways into probe-side file scans at
/// execution time. On by default; `KRISHIV_RUNTIME_FILTERS=off` disables
/// them all (the AQE dual-run switch for result-identity verification).
pub fn runtime_filters_enabled_from_env() -> bool {
    match RUNTIME_FILTERS_OVERRIDE.load(std::sync::atomic::Ordering::Relaxed) {
        0 => return false,
        1 => return true,
        _ => {}
    }
    !matches!(
        std::env::var("KRISHIV_RUNTIME_FILTERS")
            .unwrap_or_default()
            .trim()
            .to_ascii_lowercase()
            .as_str(),
        "off" | "0" | "false" | "disabled"
    )
}

/// Resolve the batch size from `KRISHIV_BATCH_SIZE` env var.
///
/// Falls back to DataFusion's default (8192) if unset or invalid.
pub fn batch_size_from_env() -> usize {
    std::env::var("KRISHIV_BATCH_SIZE")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|n| *n > 0)
        .unwrap_or(8192)
}

/// Resolve the default parallelism from `KRISHIV_TARGET_PARALLELISM` env var.
///
/// Falls back to available parallelism if unset.
pub fn default_parallelism_from_env() -> NonZeroUsize {
    std::env::var("KRISHIV_TARGET_PARALLELISM")
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .and_then(NonZeroUsize::new)
        .unwrap_or_else(|| std::thread::available_parallelism().unwrap_or(NonZeroUsize::MIN))
}

/// DataFusion's own default `sort_spill_reservation_bytes` (the merge-phase
/// buffer an external sort reserves up front). Deployments that set a
/// `memory_limit_bytes` smaller than this would otherwise have every sort
/// fail immediately with "Not enough memory to continue external sort"
/// before a single byte spills — the reservation alone doesn't fit the pool.
const DEFAULT_SORT_SPILL_RESERVATION_BYTES: usize = 10 * 1024 * 1024;

/// Floor for the scaled-down reservation below which DataFusion's merge step
/// has too little room to make forward progress.
const MIN_SORT_SPILL_RESERVATION_BYTES: usize = 64 * 1024;

/// Install Krishiv's optimizer rules on a session-state builder.
///
/// **A6 (review 2026-07-27).** These rules used to be written out at each
/// construction site, with a comment at one of them warning that "a rule
/// installed on only one of them is indistinguishable from a rule that works
/// until you hit the other path". That warning was correct and the drift
/// happened anyway: there is a *third* site — `planning_session_context`, the
/// context the coordinator plans **every distributed query** on — and it
/// carried none of these. The consequence was that two shipped performance
/// fixes did not apply to the path being benchmarked:
///
/// - `SpillableJoinSelection` is what lets q18's oversized hash-join build
///   side become a spillable sort-merge join. Unregistered, q18 fails with
///   `Resources exhausted: HashJoinInput` on every distributed run.
/// - `SemiJoinReductionThroughAggregate` is 88 % of q17's runtime.
/// - `CooperativeAmplifiers` is what lets distributed cancellation preempt an
///   amplifying operator at all.
///
/// Registering them in one place is the actual fix: a new construction site
/// now has to *opt out* to be wrong, and `SessionStateBuilder` is consumed and
/// returned so this composes into an existing chain.
///
/// Note on `SpillableJoinSelection::from_capacity()`: it reads the *calling
/// process's* cgroup. On an executor that is right; on the coordinator it
/// describes the coordinator, not the executors the plan will run on. The
/// threshold is overridable via `KRISHIV_SPILL_JOIN_BUILD_BYTES`, which is the
/// supported way to make coordinator-side planning use the executor's real
/// per-task share until the capacity is plumbed through the stage builder.
#[must_use]
pub fn with_krishiv_optimizer_rules(
    builder: datafusion::execution::session_state::SessionStateBuilder,
) -> datafusion::execution::session_state::SessionStateBuilder {
    with_krishiv_optimizer_rules_with_join_threshold(builder, None)
}

/// As [`with_krishiv_optimizer_rules`], with the spillable-join build-side
/// threshold given explicitly; `None` derives it from this process's capacity.
#[must_use]
pub fn with_krishiv_optimizer_rules_with_join_threshold(
    builder: datafusion::execution::session_state::SessionStateBuilder,
    spill_join_build_bytes: Option<u64>,
) -> datafusion::execution::session_state::SessionStateBuilder {
    let spillable_join = match spill_join_build_bytes {
        // Deliberately NOT grace-aware: this builder plans the stages the
        // coordinator has to *encode*, and `GraceHashJoinExec` is a Krishiv
        // node that `datafusion-proto` cannot serialize. A grace join here made
        // the whole fragment fail to encode, and the scheduler's response to an
        // unencodable stage plan is to run the query as a SINGLE TASK — so
        // enabling the flag silently un-distributed q10 and q21 while looking
        // like a memory fix.
        //
        // Grace is applied on the executor instead, after decode, by
        // `distributed_plan::apply_local_spill_strategy`. Which algorithm an
        // operator uses to spill is a local execution decision; it has no
        // business on the wire.
        Some(bytes) => crate::spillable_join::SpillableJoinSelection::with_threshold(Some(bytes)),
        None => crate::spillable_join::SpillableJoinSelection::from_capacity(),
    }
    // ...except in a process that never encodes a stage plan. The paragraph
    // above is about the *coordinator*, and this builder serves the one-shot
    // CLI too, where the encodability argument simply does not apply — nothing
    // it plans goes on the wire. Gated on `is_single_query_process()`, the same
    // predicate that enables the degenerate-broadcast rescue, so the rescue and
    // the algorithm it wants stop being enabled in different processes. Still
    // off unless `KRISHIV_GRACE_HASH_JOIN` says otherwise.
    .with_grace_where_plans_are_never_encoded();
    builder
        .with_physical_optimizer_rule(std::sync::Arc::new(
            crate::coop_amplifiers::CooperativeAmplifiers::new(),
        ))
        // q18: a hash-join build side that exceeds the per-task memory share
        // fails under a cgroup cap because hash join cannot spill. This rule
        // converts exactly those joins — known-large build sides only — to
        // sort-merge, which can. See `spillable_join` for why this is per-join
        // and not a session config bit.
        .with_physical_optimizer_rule(std::sync::Arc::new(spillable_join))
        // Aggregates joined on their own grouping key only need the groups the
        // join keeps; see `semi_join_reduction`.
        .with_optimizer_rule(std::sync::Arc::new(
            crate::semi_join_reduction::SemiJoinReductionThroughAggregate,
        ))
        // q18: a decorrelated IN-subquery semi-join lands at the top of the
        // plan, so the most selective predicate runs after the joins it should
        // have shrunk. Push it into the join input instead.
        //
        // On by default: with it off, q18 at SF100 does not merely slow down,
        // it FAILS — the sort cannot get its first 2.1 MB because the joins
        // hold the whole 2.6 GB pool. With it on, 424 s. See
        // `SEMI_JOIN_PUSHDOWN_ENV`.
        .with_optimizer_rule(std::sync::Arc::new(
            crate::semi_join_reduction::SemiJoinPushdownThroughInnerJoin::default(),
        ))
        // q7: the two 25-row `nation` tables are last in the FROM clause, so a
        // left-deep plan applies `n_name IN (FRANCE, GERMANY)` above every big
        // join — after `s_nationkey` has ridden two 9.48 GB shuffles that were
        // measured at 81% of the query. Reduce the fact stream by the filtered
        // dimension first; the pushdown rule above then carries that reducer
        // down onto the `supplier` scan.
        //
        // Registered *after* the pushdown rule so the reducer it introduces is
        // picked up on the following pass rather than sitting at the top.
        .with_optimizer_rule(std::sync::Arc::new(
            crate::semi_join_reduction::SemiJoinReductionFromSelectiveDimension::default(),
        ))
        // q10: a `GROUP BY` that lists a key *and the columns that key
        // determines* drags those columns through every join and shuffle to
        // display twenty of them. Group on the key, take the top N, then
        // re-fetch. Measured at 14.8x on q10 at SF100 — and measured *not* to
        // be obtainable any cheaper; see `late_materialize` for the two local
        // rewrites that were tried first and lost.
        //
        // Registered last on purpose: it rewrites what the projection and
        // limit rules have already settled, and DataFusion's optimizer loops
        // (`max_passes`), so `optimize_projections` runs again afterwards and
        // is what actually prunes the deferred columns out of the scans.
        .with_optimizer_rule(std::sync::Arc::new(
            crate::late_materialize::LateMaterializeTopKAggregate::default(),
        ))
}

/// Build the DataFusion session config with a configurable parallelism level.
///
/// When `target_partitions > 1`, round-robin repartitioning is enabled so
/// DataFusion can balance work across threads for hash-join build,
/// aggregation spill, and parquet scan parallelism.
///
/// `execution.batch_size` is set from `KRISHIV_BATCH_SIZE` (default: 8192).
///
/// `memory_limit_bytes`, when `Some`, scales `sort_spill_reservation_bytes`
/// down proportionally so a tight memory pool can still spill instead of
/// failing outright because the reservation itself doesn't fit. Pools at or
/// above `4 * DEFAULT_SORT_SPILL_RESERVATION_BYTES` (40MB) are unaffected —
/// this only kicks in for genuinely memory-constrained deployments.
pub(crate) fn build_single_node_session_config(
    target_partitions: NonZeroUsize,
    memory_limit_bytes: Option<usize>,
) -> datafusion::prelude::SessionConfig {
    let tp = target_partitions.get();
    let batch_size = batch_size_from_env();
    let mut config = datafusion::prelude::SessionConfig::new()
        .with_target_partitions(tp)
        .with_batch_size(batch_size)
        .with_information_schema(true)
        .set_bool(
            "datafusion.optimizer.enable_round_robin_repartition",
            tp > 1,
        )
        // Phase 54 runtime filters: DataFusion's master switch does NOT
        // suppress the per-operator options when set to false (it only
        // force-enables them when true), so `KRISHIV_RUNTIME_FILTERS=off`
        // must clear all four together.
        .set_bool(
            "datafusion.optimizer.enable_dynamic_filter_pushdown",
            runtime_filters_enabled_from_env(),
        )
        .set_bool(
            "datafusion.optimizer.enable_join_dynamic_filter_pushdown",
            runtime_filters_enabled_from_env(),
        )
        .set_bool(
            "datafusion.optimizer.enable_topk_dynamic_filter_pushdown",
            runtime_filters_enabled_from_env(),
        )
        .set_bool(
            "datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown",
            runtime_filters_enabled_from_env(),
        );
    // Phase 60: use a lambda-capable parser dialect so Spark-style higher-order
    // functions (`transform(arr, x -> …)`, `filter`, `forall`) parse — the
    // default `Generic` dialect treats `->` as the JSON-arrow operator, so the
    // lambda variable is mis-planned as a column reference. DuckDB is the most
    // standards-compatible dialect that supports BOTH `x -> body` lambdas and
    // `[...]` array literals; it changes only parse syntax, not semantics, and
    // is validated against the full krishiv-sql suite.
    config.options_mut().sql_parser.dialect = datafusion::common::config::Dialect::DuckDB;
    // Parquet scan options stay at DataFusion's defaults (`pushdown_filters`
    // off, `enable_page_index` on). Forcing `pushdown_filters = true` here
    // cost ~2.2× on scan-heavy queries (Phase 52 #194 attribution probe,
    // TPC-H Q6 SF1: 268 ms → 121 ms); workloads that benefit from row-level
    // late materialization can opt in per session via
    // `SET datafusion.execution.parquet.pushdown_filters = true`.
    if let Some(limit) = memory_limit_bytes {
        let scaled = (limit / 4).clamp(
            MIN_SORT_SPILL_RESERVATION_BYTES,
            DEFAULT_SORT_SPILL_RESERVATION_BYTES,
        );
        config = config.with_sort_spill_reservation_bytes(scaled);
    }
    config
}

/// Iceberg catalogs registered via `with_iceberg_catalog`, paired with their
/// DataFusion catalog name, behind a shared lock for `CALL system.<proc>`
/// dispatch.
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
type IcebergCatalogRegistry =
    Arc<std::sync::RwLock<Vec<(Arc<catalog::unified::KrishivCatalog>, String)>>>;

/// Local SQL engine backed by DataFusion.
///
/// **Local-only**: all SQL execution is in-process via DataFusion. This crate
/// exposes no distributed execution path of its own — the distributed batch
/// runtime consumes [`distributed_plan`] instead.
///
/// Scoped to R1: DataFusion is expected to move behind the `KrishivDataFrameOps`
/// trait in a later release.
///
/// `register_parquet`, `read_delta` and `read_hudi` treat their path arguments
/// as local filesystem paths; S3/GCS URIs go through the object-store
/// connector layer (see [`object_store_registry`]).
#[derive(Clone)]
pub struct SqlEngine {
    context: SessionContext,
    target_parallelism: NonZeroUsize,
    krishiv_catalog: Option<Arc<RwLock<InMemoryCatalog>>>,
    udf_registry: Option<std::sync::Arc<std::sync::RwLock<krishiv_plan::udf::UdfRegistry>>>,
    /// Table names registered as unbounded streaming sources.
    /// Wrapped in `Arc<RwLock<>>` so that Session clones share the same set.
    streaming_sources: Arc<RwLock<std::collections::HashSet<String>>>,
    /// Serializes streaming table name validation and catalog registration.
    streaming_registration: Arc<Mutex<()>>,
    /// `true` once any streaming source has been registered.  Checked with a
    /// relaxed atomic load before acquiring `streaming_sources` so that the
    /// common case (no streaming sources, pure batch workload) avoids both the
    /// lock and the SQL parse inside `is_streaming_query`.
    has_streaming_sources: Arc<AtomicBool>,
    /// Optional UDF resource limits to apply when syncing UDFs for this engine.
    /// Set for job-specific engines so sandbox enforcement uses the job's budgets.
    udf_limits: Option<krishiv_plan::udf::ResourceLimits>,
    /// Monotonically increasing version counter; incremented on every UDF
    /// registration or removal. Used to skip `sync_all_udfs()` when nothing
    /// has changed since the last sync.
    udf_registry_version: Arc<AtomicU64>,
    /// The version at which the last `sync_all_udfs()` was performed.
    /// Compared against `udf_registry_version` to detect staleness.
    udf_last_synced_version: Arc<AtomicU64>,
    /// Bounded query plan cache: query text → DataFusion LogicalPlan.
    /// Skips re-parsing and re-optimising identical repeated queries.
    /// Max `PLAN_CACHE_MAX_ENTRIES` entries; oldest entry evicted when full.
    /// Single-lock design prevents the TOCTOU race of the previous two-structure
    /// (`DashMap` + `VecDeque`) implementation.
    plan_cache: Arc<Mutex<PlanCache>>,
    /// Override for shuffle partition count (`SET shuffle.partitions = N`).
    /// When `Some`, exchange nodes use this bucket count instead of auto-sizing.
    shuffle_partitions: Arc<std::sync::RwLock<Option<u32>>>,
    /// Estimated row counts for registered tables, keyed by table name.
    /// Populated by `register_parquet` and `register_record_batches`.
    /// Used by `krishiv_logical_plan` to annotate scan nodes for the
    /// `BroadcastAutoRule` optimizer.
    table_row_counts: Arc<std::sync::RwLock<HashMap<String, u64>>>,
    /// DataFusion memory pool limit in bytes for this engine, when bounded.
    /// `None` means the default unbounded pool. When `Some`, the engine runs
    /// with a `FairSpillPool` so sorts, hash joins, and aggregations spill to
    /// disk under memory pressure instead of growing without bound.
    memory_limit_bytes: Option<usize>,
    /// Iceberg catalogs registered via `with_iceberg_catalog`, keyed by their
    /// DataFusion catalog name. Stored so that `CALL system.<proc>` statements
    /// can dispatch maintenance operations to the right catalog.
    #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
    iceberg_catalogs: IcebergCatalogRegistry,
    /// Live-table DDL registry shared across SQL and session APIs.
    live_table_registry: Arc<live_table::LiveTableRegistry>,
    /// Incremental-view DDL registry shared across SQL and session APIs.
    incremental_view_registry: Arc<incremental_view::IncrementalViewRegistry>,
    /// Pipeline DDL registry (CREATE SOURCE / CREATE SINK metadata).
    pipeline_registry: Arc<pipeline_ddl::PipelineRegistry>,
    /// Cancelled operation IDs and progress snapshots for query lifecycle control.
    operation_registry: Arc<OperationRegistry>,
}

impl fmt::Debug for SqlEngine {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SqlEngine")
            .field("backend", &"datafusion")
            .finish_non_exhaustive()
    }
}

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

impl SqlEngine {
    /// Create a local SQL engine.
    ///
    /// Window helper UDFs (`tumble_start`, `tumble_end`, `hop_start`, `hop_end`)
    /// are registered as part of construction. If registration fails the
    /// engine is still returned — non-window queries work — and a
    /// `tracing::warn!` is emitted. Use [`SqlEngine::try_new`] when callers
    /// need to surface the registration error.
    ///
    /// DataFusion `target_partitions` defaults to the machine's available CPU
    /// parallelism (matching a bare `datafusion::SessionContext`), overridable
    /// via `KRISHIV_TARGET_PARALLELISM` or [`SqlEngine::with_target_parallelism`].
    /// Executor task engines deliberately scale this down to their per-slot
    /// share so concurrent tasks don't oversubscribe the machine.
    pub fn new() -> Self {
        Self::new_with_engine_memory(EngineMemory::for_this_process())
    }

    /// Create a local SQL engine whose DataFusion execution memory is capped
    /// at `memory_limit_bytes`.
    ///
    /// When `Some`, the engine runs with a `FairSpillPool` of that size plus
    /// the default disk manager, so memory-intensive operators (sort, hash
    /// join, aggregation) spill to disk under pressure and queries that cannot
    /// spill fail with a resources-exhausted error instead of exhausting
    /// process memory. `None` keeps DataFusion's default unbounded pool.
    ///
    /// Shares [`SqlEngine::new`]'s fallback behavior for window helper UDF
    /// registration failures.
    pub fn new_with_memory_limit(memory_limit_bytes: Option<usize>) -> Self {
        Self::new_with_engine_memory(EngineMemory::from_limit(memory_limit_bytes))
    }

    /// Create a local SQL engine over an explicit [`EngineMemory`] source.
    ///
    /// Prefer this over [`new_with_memory_limit`](Self::new_with_memory_limit)
    /// when several engines live in one process: passing each of them
    /// [`EngineMemory::Shared`] with the same pool bounds their *combined*
    /// execution memory, which per-engine limits cannot do.
    ///
    /// Shares [`SqlEngine::new`]'s fallback behavior for window helper UDF
    /// registration failures.
    pub fn new_with_engine_memory(engine_memory: EngineMemory) -> Self {
        let parallelism = default_parallelism_from_env();
        match Self::build_local(
            None,
            WindowFnRegistration::Register,
            parallelism,
            engine_memory.clone(),
        ) {
            Ok(engine) => engine,
            Err(err) => {
                tracing::warn!(
                    error = %err,
                    "SqlEngine::new: window helper UDF registration failed; \
                     window SQL functions will be unavailable, other queries are unaffected"
                );
                Self::build_local(
                    None,
                    WindowFnRegistration::Skip,
                    parallelism,
                    engine_memory.clone(),
                )
                .unwrap_or_else(|err| {
                    tracing::error!(
                        error = %err,
                        "memory-limited DataFusion runtime construction failed; \
                         falling back to an unbounded engine"
                    );
                    Self::build_local(
                        None,
                        WindowFnRegistration::Skip,
                        parallelism,
                        EngineMemory::Unbounded,
                    )
                    .unwrap_or_else(|_| Self::build_absolute_minimal(parallelism))
                })
            }
        }
    }

    /// Create a local SQL engine, propagating window helper registration errors.
    ///
    /// Callers that need to abort startup when window functions cannot be
    /// registered should use this constructor.
    pub fn try_new() -> SqlResult<Self> {
        Self::build_local(
            None,
            WindowFnRegistration::Register,
            default_parallelism_from_env(),
            EngineMemory::for_this_process(),
        )
    }

    /// Create an engine whose `krishiv` catalog resolves tables registered in `InMemoryCatalog` (P0-10).
    pub fn with_in_memory_catalog(catalog: Arc<RwLock<InMemoryCatalog>>) -> SqlResult<Self> {
        if krishiv_common::profile_requires_fail_closed_metadata(
            krishiv_common::resolve_durability_profile(),
        ) {
            return Err(SqlError::DataFusion {
                message: String::from(
                    "InMemoryCatalog is dev-only; configure a durable REST or file-backed \
                     catalog for production deployments",
                ),
            });
        }
        Self::build_local(
            Some(catalog),
            WindowFnRegistration::Register,
            default_parallelism_from_env(),
            EngineMemory::for_this_process(),
        )
    }

    /// Set the DataFusion `target_partitions` parallelism level for this engine.
    ///
    /// Higher values allow DataFusion to parallelise hash-join build,
    /// aggregation spilling, and parquet scans across more threads.
    /// Default: [`default_parallelism_from_env`] (available CPU parallelism,
    /// `KRISHIV_TARGET_PARALLELISM` override).
    ///
    /// The new level applies to queries planned after this call; plans already
    /// produced (or cached logical plans re-planned physically) pick it up at
    /// physical planning time.
    #[must_use]
    pub fn with_target_parallelism(mut self, n: NonZeroUsize) -> Self {
        self.target_parallelism = n;
        self.apply_target_partitions(n);
        self
    }

    /// Write `n` back into the live session state so DataFusion actually
    /// plans with it.
    ///
    /// `SessionConfig` is baked into the `SessionState` at construction;
    /// setting only the `target_parallelism` field would leave physical
    /// planning at the construction-time partition count (the Phase 51
    /// 4.5–8.9× embedded-overhead finding was exactly this: every
    /// `with_target_parallelism` caller silently kept running single-threaded).
    fn apply_target_partitions(&self, n: NonZeroUsize) {
        let state_ref = self.context.state_ref();
        let mut state = state_ref.write();
        let options = state.config_mut().options_mut();
        options.execution.target_partitions = n.get();
        options.optimizer.enable_round_robin_repartition = n.get() > 1;
    }

    /// Return the configured `target_partitions` parallelism level.
    pub fn target_parallelism(&self) -> NonZeroUsize {
        self.target_parallelism
    }

    /// Return the DataFusion memory pool limit for this engine, if bounded.
    pub fn memory_limit_bytes(&self) -> Option<usize> {
        self.memory_limit_bytes
    }

    /// Direct access to the underlying DataFusion session context.
    ///
    /// Used by the distributed stage builder (ADR-0003) to create and
    /// round-trip physical plans; general query execution should go through
    /// [`SqlEngine::sql`] so engine-level rewrites and governance apply.
    pub fn session_context(&self) -> &SessionContext {
        &self.context
    }

    /// Return the current `shuffle.partitions` override, if set via `SET shuffle.partitions = N`.
    pub fn shuffle_partitions(&self) -> Option<u32> {
        *self
            .shuffle_partitions
            .read()
            .unwrap_or_else(|e| e.into_inner())
    }

    /// Return access to the table row-count registry.
    ///
    /// Populated by `register_parquet` and `register_record_batches` with
    /// estimated row counts extracted from table-provider statistics. Used
    /// by `SqlDataFrame::krishiv_logical_plan` to annotate scan nodes.
    pub fn table_row_counts(&self) -> Arc<std::sync::RwLock<HashMap<String, u64>>> {
        Arc::clone(&self.table_row_counts)
    }

    /// Return table/view names registered in the live DataFusion catalog.
    ///
    /// Uses DataFusion's catalog provider API directly instead of routing
    /// through `SHOW TABLES`, which requires optional information-schema
    /// support in some DataFusion configurations.
    pub fn registered_table_names(&self) -> Vec<String> {
        let mut names = Vec::new();
        for catalog_name in self.context.catalog_names() {
            let Some(catalog) = self.context.catalog(&catalog_name) else {
                continue;
            };
            for schema_name in catalog.schema_names() {
                let Some(schema) = catalog.schema(&schema_name) else {
                    continue;
                };
                names.extend(schema.table_names());
            }
        }
        names.sort();
        names.dedup();
        names
    }

    /// Build a `SqlDataFrame` with this engine's shared session context attached
    /// so that `cache()` / `create_or_replace_temp_view()` work on the live session.
    fn make_sql_df(&self, name: &str, dataframe: DataFusionDataFrame) -> SqlDataFrame {
        SqlDataFrame::new(name, dataframe, self.table_row_counts())
            .with_context(self.context.clone())
    }

    /// Attach SQL text and execution kind derived from registered streaming sources.
    fn attach_query_metadata(&self, df: SqlDataFrame, query: &str) -> SqlDataFrame {
        let kind = if self.is_streaming_query(query).unwrap_or(false) {
            ExecutionKind::Streaming
        } else {
            ExecutionKind::Batch
        };
        df.with_query(query).with_execution_kind(kind)
    }

    /// Set an override for the shuffle partition count.
    ///
    /// When `n` is `Some`, exchange and shuffle-write operations use `n` buckets
    /// instead of auto-sizing. Pass `None` to restore auto-sizing.
    #[must_use]
    pub fn with_shuffle_partitions(self, n: Option<u32>) -> Self {
        if let Ok(mut guard) = self.shuffle_partitions.write() {
            *guard = n;
        }
        self
    }

    /// Internal builder shared by the public constructors.
    ///
    /// `krishiv_catalog` is `Some(...)` when the engine should bridge to an
    /// `InMemoryCatalog`; `None` for a default engine.
    ///
    /// `window_fn_registration` controls whether the helper UDFs
    /// (`tumble_start` / `tumble_end` / `hop_start` / `hop_end`) are
    /// registered. `Skip` is used as a fallback by [`SqlEngine::new`] when
    /// `Register` fails; it is infallible.
    fn build_local(
        krishiv_catalog: Option<Arc<RwLock<InMemoryCatalog>>>,
        window_fn_registration: WindowFnRegistration,
        target_partitions: NonZeroUsize,
        engine_memory: EngineMemory,
    ) -> SqlResult<Self> {
        let memory_limit_bytes = engine_memory.sizing_bytes();
        // Create streaming_sources first so it can be shared with KafkaTableFactory.
        // DDL-created Kafka tables (CREATE EXTERNAL TABLE … STORED AS KAFKA) then
        // correctly register in is_streaming_query.
        let streaming_sources: Arc<RwLock<std::collections::HashSet<String>>> =
            Arc::new(RwLock::new(std::collections::HashSet::new()));

        let mut state_builder = with_krishiv_optimizer_rules(
            datafusion::execution::session_state::SessionStateBuilder::new()
                .with_default_features(),
        )
        .with_config(build_single_node_session_config(
            target_partitions,
            memory_limit_bytes,
        ));
        {
            // The lazy registry is installed unconditionally, not only when a
            // memory limit is configured. An executor decoding a `dfplan:`
            // fragment discovers which buckets the plan touches only by
            // decoding it, and the decode is what needs the store — so there
            // is no earlier point at which the bucket could be registered.
            let mut runtime_builder = datafusion::execution::runtime_env::RuntimeEnvBuilder::new()
                .with_object_store_registry(Arc::new(
                    crate::object_store_registry::LazyCloudObjectStoreRegistry::new(),
                ));
            if let Some(pool) = engine_memory.pool() {
                // A FairSpillPool divides its capacity across concurrently
                // running consumers and lets spill-capable operators (sort,
                // hash join, aggregation) write to the default disk manager's
                // temp files instead of failing outright when the pool is
                // exhausted. Under `EngineMemory::Shared` that pool is the
                // executor's single process-wide pool, so the consumers being
                // divided across are every operator of every concurrent task.
                runtime_builder = runtime_builder.with_memory_pool(pool);
            }
            let runtime_env = runtime_builder
                .build_arc()
                .map_err(|e| SqlError::DataFusion {
                    message: format!(
                        "failed to build DataFusion runtime \
                     (memory limit {memory_limit_bytes:?} bytes): {e}"
                    ),
                })?;
            state_builder = state_builder.with_runtime_env(runtime_env);
        }
        let mut state = state_builder.build();
        // Connector factories layer on top of the defaults the builder already
        // installed; mutating in place avoids building a second throwaway
        // SessionState just to harvest the default factory map.
        crate::connector_table::register_connector_table_factories(
            state.table_factories_mut(),
            streaming_sources.clone(),
        );
        let context = SessionContext::new_with_state(state);
        if let Some(catalog) = &krishiv_catalog {
            context.register_catalog(
                "krishiv",
                Arc::new(DataFusionCatalogBridge::new(catalog.clone())),
            );
        }
        if matches!(window_fn_registration, WindowFnRegistration::Register) {
            window_functions::register_window_functions(&context).map_err(|e| {
                SqlError::DataFusion {
                    message: format!("failed to register window helper UDFs: {e}"),
                }
            })?;
        }
        // Phase 60: Spark-reference JSON scalar functions are always available on
        // the batch SQL front door (get_json_object, json_array_length).
        json_functions::register_json_functions(&context).map_err(|e| SqlError::DataFusion {
            message: format!("failed to register JSON UDFs: {e}"),
        })?;
        // Phase 60: Spark-parity higher-order array functions
        // (transform/filter/exists/forall) on the batch SQL front door.
        higher_order_functions::register_higher_order_spark_functions(&context).map_err(|e| {
            SqlError::DataFusion {
                message: format!("failed to register higher-order UDFs: {e}"),
            }
        })?;
        // Phase 60: Spark-parity scalar functions (Spark-pattern date_format, crc32).
        spark_functions::register_spark_scalar_functions(&context).map_err(|e| {
            SqlError::DataFusion {
                message: format!("failed to register Spark scalar UDFs: {e}"),
            }
        })?;
        Ok(Self {
            context,
            target_parallelism: target_partitions,
            krishiv_catalog,
            udf_registry: None,
            streaming_sources,
            streaming_registration: Arc::new(Mutex::new(())),
            has_streaming_sources: Arc::new(AtomicBool::new(false)),
            udf_limits: None,
            udf_registry_version: Arc::new(AtomicU64::new(0)),
            udf_last_synced_version: Arc::new(AtomicU64::new(u64::MAX)),
            plan_cache: Arc::new(Mutex::new(PlanCache::new(resolve_plan_cache_max_entries()))),
            shuffle_partitions: Arc::new(std::sync::RwLock::new(None)),
            table_row_counts: Arc::new(std::sync::RwLock::new(HashMap::new())),
            memory_limit_bytes,
            #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
            iceberg_catalogs: Arc::new(std::sync::RwLock::new(Vec::new())),
            live_table_registry: Arc::new(live_table::LiveTableRegistry::new()),
            incremental_view_registry: Arc::new(incremental_view::IncrementalViewRegistry::new()),
            pipeline_registry: Arc::new(pipeline_ddl::PipelineRegistry::new()),
            operation_registry: Arc::new(OperationRegistry::new()),
        })
    }

    /// Build the absolute minimal engine: no catalog, no window UDFs, no memory
    /// limit. Every step is infallible, so the return type is `Self`. Used as
    /// the last-resort fallback in `new_with_memory_limit`.
    fn build_absolute_minimal(target_partitions: NonZeroUsize) -> Self {
        let streaming_sources: Arc<RwLock<std::collections::HashSet<String>>> =
            Arc::new(RwLock::new(std::collections::HashSet::new()));
        let mut state = with_krishiv_optimizer_rules(
            datafusion::execution::session_state::SessionStateBuilder::new()
                .with_default_features(),
        )
        .with_config(build_single_node_session_config(target_partitions, None))
        .build();
        crate::connector_table::register_connector_table_factories(
            state.table_factories_mut(),
            streaming_sources.clone(),
        );
        let context = SessionContext::new_with_state(state);
        Self {
            context,
            target_parallelism: target_partitions,
            krishiv_catalog: None,
            udf_registry: None,
            streaming_sources,
            streaming_registration: Arc::new(Mutex::new(())),
            has_streaming_sources: Arc::new(AtomicBool::new(false)),
            udf_limits: None,
            udf_registry_version: Arc::new(AtomicU64::new(0)),
            udf_last_synced_version: Arc::new(AtomicU64::new(u64::MAX)),
            plan_cache: Arc::new(Mutex::new(PlanCache::new(resolve_plan_cache_max_entries()))),
            shuffle_partitions: Arc::new(std::sync::RwLock::new(None)),
            table_row_counts: Arc::new(std::sync::RwLock::new(HashMap::new())),
            memory_limit_bytes: None,
            #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
            iceberg_catalogs: Arc::new(std::sync::RwLock::new(Vec::new())),
            live_table_registry: Arc::new(live_table::LiveTableRegistry::new()),
            incremental_view_registry: Arc::new(incremental_view::IncrementalViewRegistry::new()),
            pipeline_registry: Arc::new(pipeline_ddl::PipelineRegistry::new()),
            operation_registry: Arc::new(OperationRegistry::new()),
        }
    }

    /// Register an unbounded continuous table, returning its typed input.
    ///
    /// The returned input uses a bounded channel with capacity
    /// [`crate::streaming::CONTINUOUS_TABLE_CHANNEL_CAPACITY`]. When the
    /// consumer (the DataFusion query plan) is slower than the producer,
    /// `ContinuousTableInput::send(...).await` backpressures the producer,
    /// and `ContinuousTableInput::try_send(...)` returns a resource error
    /// rather than growing memory without limit. Use
    /// [`Self::register_streaming_table_with_capacity`] for a non-default
    /// capacity.
    pub fn register_streaming_table(
        &self,
        name: &str,
        schema: arrow::datatypes::SchemaRef,
    ) -> SqlResult<Arc<ContinuousTableInput>> {
        let _registration = self.lock_streaming_registration()?;
        self.validate_new_streaming_table(name, &schema)?;
        let (table, input) = crate::streaming::create_continuous_table(schema).map_err(|e| {
            SqlError::DataFusion {
                message: e.to_string(),
            }
        })?;
        self.register_new_streaming_provider(name, table)?;
        self.streaming_sources
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .insert(name.to_string());
        self.has_streaming_sources.store(true, Ordering::Release);
        self.invalidate_plan_cache();
        Ok(input)
    }

    /// Same as [`Self::register_streaming_table`] but with a caller-supplied
    /// channel capacity. Useful for tests that want to exercise the
    /// full/empty channel boundary without pushing
    /// `CONTINUOUS_TABLE_CHANNEL_CAPACITY` (64) batches.
    pub fn register_streaming_table_with_capacity(
        &self,
        name: &str,
        schema: arrow::datatypes::SchemaRef,
        capacity: usize,
    ) -> SqlResult<Arc<ContinuousTableInput>> {
        let _registration = self.lock_streaming_registration()?;
        self.validate_new_streaming_table(name, &schema)?;
        let (table, input) = crate::streaming::create_continuous_table_with_capacity(
            schema, capacity,
        )
        .map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })?;
        self.register_new_streaming_provider(name, table)?;
        self.streaming_sources
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .insert(name.to_string());
        self.has_streaming_sources.store(true, Ordering::Release);
        self.invalidate_plan_cache();
        Ok(input)
    }

    fn lock_streaming_registration(&self) -> SqlResult<std::sync::MutexGuard<'_, ()>> {
        self.streaming_registration
            .lock()
            .map_err(|error| SqlError::DataFusion {
                message: format!("streaming table registration lock poisoned: {error}"),
            })
    }

    fn validate_new_streaming_table(
        &self,
        name: &str,
        schema: &arrow::datatypes::SchemaRef,
    ) -> SqlResult<()> {
        if name.trim().is_empty() {
            return Err(SqlError::EmptyTableName);
        }
        if schema.fields().is_empty() {
            return Err(SqlError::DataFusion {
                message: "streaming table schema must contain at least one field".into(),
            });
        }
        if self
            .context
            .table_exist(name)
            .map_err(|error| SqlError::DataFusion {
                message: error.to_string(),
            })?
        {
            return Err(SqlError::DataFusion {
                message: format!("table '{name}' is already registered"),
            });
        }
        Ok(())
    }

    fn register_new_streaming_provider(
        &self,
        name: &str,
        table: Arc<dyn datafusion::catalog::TableProvider>,
    ) -> SqlResult<()> {
        let previous =
            self.context
                .register_table(name, table)
                .map_err(|error| SqlError::DataFusion {
                    message: error.to_string(),
                })?;
        if let Some(previous) = previous {
            self.context
                .register_table(name, previous)
                .map_err(|error| SqlError::DataFusion {
                    message: format!(
                        "table '{name}' was concurrently registered and could not be restored: \
                         {error}"
                    ),
                })?;
            return Err(SqlError::DataFusion {
                message: format!("table '{name}' was concurrently registered"),
            });
        }
        Ok(())
    }

    /// Register a live Kafka/Redpanda topic as an unbounded streaming table.
    ///
    /// This is the native Rust path — no Python bridge or external process
    /// required.  Under the hood it creates an `rdkafka` consumer and wraps it
    /// in a DataFusion `StreamingTable` so normal SQL queries (`SELECT`,
    /// `GROUP BY`, windowed aggregations) work against the live topic.
    ///
    /// Equivalent SQL DDL:
    /// ```sql
    /// CREATE EXTERNAL TABLE <name> (<cols>) STORED AS KAFKA
    ///   LOCATION '<topic>'
    ///   OPTIONS ('bootstrap.servers' = '…', 'group.id' = '…');
    /// ```
    pub fn register_kafka_source(
        &self,
        table_name: impl AsRef<str>,
        schema: arrow::datatypes::SchemaRef,
        bootstrap_servers: impl Into<String>,
        topic: impl Into<String>,
        group_id: impl Into<String>,
    ) -> SqlResult<()> {
        let table_name = table_name.as_ref();
        if table_name.trim().is_empty() {
            return Err(SqlError::EmptyTableName);
        }
        let config = krishiv_connectors::kafka::KafkaConfig {
            bootstrap_servers: bootstrap_servers.into(),
            topic: topic.into(),
            group_id: group_id.into(),
            auto_commit_interval_ms: {
                let profile = krishiv_common::resolve_durability_profile();
                if krishiv_common::requires_manual_kafka_commit(profile) {
                    None
                } else {
                    Some(1_000)
                }
            },
            security_protocol: None,
            ssl_ca_location: None,
            ssl_certificate_location: None,
            ssl_key_location: None,
            ssl_key_password: None,
            sasl_username: None,
            sasl_password: None,
            sasl_mechanisms: None,
            enable_idempotence: None,
            transactional_id: None,
        };
        let table =
            crate::kafka_table::create_kafka_streaming_table(schema, config).map_err(|e| {
                SqlError::DataFusion {
                    message: e.to_string(),
                }
            })?;
        if self
            .context
            .table_exist(table_name)
            .map_err(SqlError::from)?
        {
            let _ = self
                .context
                .deregister_table(table_name)
                .map_err(SqlError::from)?;
        }
        self.context
            .register_table(table_name, table)
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;
        self.streaming_sources
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .insert(table_name.to_string());
        self.has_streaming_sources.store(true, Ordering::Release);
        self.invalidate_plan_cache();
        Ok(())
    }

    /// Execute a SQL query and write every result row to a Kafka/Redpanda topic.
    ///
    /// Each row is serialised as a JSON object using the same format as
    /// `KafkaSink` (krishiv-proto).  The method blocks until the query stream ends and the
    /// producer queue is flushed, then returns the total number of rows written.
    ///
    /// **Note**: If `sql` targets an unbounded streaming table (e.g. one
    /// registered via [`Self::register_kafka_source`]) this call will never return.
    /// Use it with batch sources or add a `LIMIT` clause.
    pub async fn sql_to_kafka(
        &self,
        sql: impl AsRef<str>,
        bootstrap_servers: impl Into<String>,
        topic: impl Into<String>,
    ) -> SqlResult<u64> {
        use futures::StreamExt;
        use krishiv_connectors::Sink as _;
        use krishiv_connectors::kafka::{KafkaConfig, KafkaSink};

        let config = KafkaConfig {
            bootstrap_servers: bootstrap_servers.into(),
            topic: topic.into(),
            group_id: "krishiv-sql-writer".into(),
            auto_commit_interval_ms: None,
            security_protocol: None,
            ssl_ca_location: None,
            ssl_certificate_location: None,
            ssl_key_location: None,
            ssl_key_password: None,
            sasl_username: None,
            sasl_password: None,
            sasl_mechanisms: None,
            enable_idempotence: None,
            transactional_id: None,
        };
        let mut sink = KafkaSink::new(config).map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })?;

        let df = self.sql(sql.as_ref()).await?;
        let mut stream = df.execute_stream().await?;
        let mut total_rows = 0u64;

        while let Some(result) = stream.next().await {
            let batch = result.map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;
            if batch.num_rows() > 0 {
                total_rows += batch.num_rows() as u64;
                sink.write_batch(batch)
                    .await
                    .map_err(|e| SqlError::DataFusion {
                        message: e.to_string(),
                    })?;
            }
        }
        sink.flush().await.map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })?;
        Ok(total_rows)
    }

    /// Configure this engine with explicit UDF resource limits (Track E).
    /// When set, calls to `sql()` and direct UDF syncs will use these budgets
    /// instead of unlimited defaults. Intended for job-specific engines.
    pub fn with_udf_limits(mut self, limits: krishiv_plan::udf::ResourceLimits) -> Self {
        self.udf_limits = Some(limits);
        self
    }

    /// Returns `true` if `table_name` is registered as an unbounded streaming source.
    pub fn is_streaming_source(&self, table_name: &str) -> bool {
        self.streaming_sources
            .read()
            .unwrap_or_else(|e| e.into_inner())
            .contains(table_name)
    }

    /// Register a table name as a streaming source without creating a live connector.
    ///
    /// This is the test-safe alternative to [`Self::register_kafka_source`]: it marks
    /// `table_name` in the `streaming_sources` set so that `is_streaming_query`
    /// returns `true` for queries that reference it, without constructing any
    /// broker connection. Useful for unit tests where a live Kafka broker is not
    /// available and rdkafka's log subsystem is not initialised.
    /// Returns [`SqlError::EmptyTableName`] if `table_name` is blank.
    pub fn register_streaming_source_name(&self, table_name: impl Into<String>) -> SqlResult<()> {
        let name: String = table_name.into();
        if name.trim().is_empty() {
            return Err(SqlError::EmptyTableName);
        }
        self.streaming_sources
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .insert(name);
        self.has_streaming_sources.store(true, Ordering::Release);
        self.invalidate_plan_cache();
        Ok(())
    }

    /// Remove a streaming source registration.
    ///
    /// Deregisters the table from DataFusion and removes it from the streaming-
    /// sources set. Invalidates the plan cache. Idempotent — deregistering a
    /// name that was never registered is not an error.
    pub fn deregister_streaming_source(&self, name: &str) -> SqlResult<()> {
        if name.trim().is_empty() {
            return Err(SqlError::EmptyTableName);
        }
        // Idempotent: ignore the Option return (None when table wasn't registered).
        let _ = self
            .context
            .deregister_table(name)
            .map_err(SqlError::from)?;
        {
            let mut sources = self
                .streaming_sources
                .write()
                .unwrap_or_else(|e| e.into_inner());
            sources.remove(name);
            if sources.is_empty() {
                self.has_streaming_sources.store(false, Ordering::Release);
            }
            // Invalidate while still holding the write lock so there is no window
            // between source removal and cache invalidation where a concurrent
            // is_streaming_query returns false but serves a stale cached plan (N5).
            self.invalidate_plan_cache();
        }
        Ok(())
    }

    /// Shared live-table registry for `CREATE LIVE TABLE` DDL.
    pub fn live_table_registry(&self) -> &Arc<live_table::LiveTableRegistry> {
        &self.live_table_registry
    }

    /// Shared incremental-view registry for `CREATE INCREMENTAL VIEW` DDL.
    pub fn incremental_view_registry(&self) -> &Arc<incremental_view::IncrementalViewRegistry> {
        &self.incremental_view_registry
    }

    /// Shared pipeline registry for `CREATE SOURCE` / `CREATE SINK` DDL.
    pub fn pipeline_registry(&self) -> &Arc<pipeline_ddl::PipelineRegistry> {
        &self.pipeline_registry
    }

    /// Shared operation registry for cancellation and progress reporting.
    pub fn operation_registry(&self) -> &Arc<OperationRegistry> {
        &self.operation_registry
    }

    /// Drop a named table from the session context.
    ///
    /// Idempotent — dropping a name that was never registered is not an error.
    ///
    /// Also clears any streaming-source registration for `name`. This function
    /// used to drop the table from DataFusion and leave the name in
    /// `streaming_sources` forever, with `has_streaming_sources` latched true.
    /// Two consequences, and the second is the bad one:
    ///
    /// * the set grew without bound for the life of the engine, and
    /// * a name **re-registered as an ordinary batch table** stayed classified
    ///   as streaming. `DROP TABLE events; CREATE TABLE events AS SELECT …`
    ///   followed by a plain `SELECT` over `events` would be routed down the
    ///   streaming path by [`Self::is_streaming_query`], which decides execution mode
    ///   purely from this set.
    ///
    /// `deregister_streaming_source` always did this cleanup; the two paths
    /// simply disagreed, and the generic one is the one most callers reach for.
    pub fn deregister_table(&self, name: &str) -> SqlResult<()> {
        if name.trim().is_empty() {
            return Err(SqlError::EmptyTableName);
        }
        let _ = self
            .context
            .deregister_table(name)
            .map_err(SqlError::from)?;
        {
            let mut sources = self
                .streaming_sources
                .write()
                .unwrap_or_else(|e| e.into_inner());
            sources.remove(name);
            if sources.is_empty() {
                self.has_streaming_sources.store(false, Ordering::Release);
            }
            // Invalidated under the write lock, as in
            // `deregister_streaming_source`: otherwise there is a window where
            // a concurrent `is_streaming_query` sees the name gone but still
            // serves a stale cached plan built for the streaming shape (N5).
            self.invalidate_plan_cache();
        }
        Ok(())
    }

    /// Register a table UDF backed by a Rust closure.
    ///
    /// The closure receives literal arguments passed by the SQL caller as
    /// `ScalarValue` values and returns an Arrow `RecordBatch`. Non-literal
    /// arguments are rejected because they cannot be evaluated safely at the
    /// synchronous DataFusion table-function boundary. `schema` describes the
    /// output columns.
    ///
    /// # Example
    /// ```ignore
    /// engine.register_table_udf_fn(
    ///     "generate_ints",
    ///     Schema::new(vec![Field::new("n", DataType::Int64, false)]),
    ///     |args| {
    ///         let count = match args.first() {
    ///             Some(ScalarValue::Int64(n)) => *n,
    ///             _ => 10,
    ///         };
    ///         let arr = Int64Array::from((0..count).collect::<Vec<_>>());
    ///         Ok(RecordBatch::try_from_iter([("n", Arc::new(arr) as _)])?)
    ///     },
    /// )?;
    /// ```
    pub fn register_table_udf_fn(
        &self,
        name: impl Into<String>,
        schema: arrow::datatypes::Schema,
        f: impl Fn(
            &[krishiv_plan::udf::ScalarValue],
        ) -> Result<arrow::record_batch::RecordBatch, krishiv_plan::udf::UdfError>
        + Send
        + Sync
        + 'static,
    ) -> SqlResult<()> {
        let udf =
            create_function_ddl::ClosureTableUdf::try_new(name, schema, std::sync::Arc::new(f))
                .map_err(|error| SqlError::InvalidTableFunction {
                    message: error.to_string(),
                })?;
        if let Some(registry) = &self.udf_registry {
            let mut guard = registry.write().map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;
            guard.register_table(std::sync::Arc::new(udf.clone()));
        }
        udf::register_single_table_udf(&self.context, std::sync::Arc::new(udf))
            .map_err(SqlError::from)?;
        self.bump_udf_version();
        Ok(())
    }

    /// Returns `true` if any table referenced in `sql` is a registered streaming source.
    pub fn is_streaming_query(&self, sql: &str) -> SqlResult<bool> {
        // Fast-path: avoid the RwLock acquire and SQL parse for the common case
        // where no streaming sources have ever been registered (pure batch engines).
        if !self.has_streaming_sources.load(Ordering::Acquire) {
            return Ok(false);
        }
        let sources = self
            .streaming_sources
            .read()
            .unwrap_or_else(|e| e.into_inner());
        if sources.is_empty() {
            return Ok(false);
        }
        let dialect = GenericDialect {};
        let statements = Parser::parse_sql(&dialect, sql).map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })?;
        for stmt in &statements {
            let mut is_streaming = false;
            let _ = visit_relations(stmt, |relation| {
                // relation.to_string() yields the fully-qualified name (e.g. "schema.table").
                // Extract the unqualified table name (last segment after dot).
                let full = relation.to_string();
                let table_name = full.split('.').next_back().unwrap_or(&full);
                if sources.contains(table_name) {
                    is_streaming = true;
                    return ControlFlow::Break(());
                }
                ControlFlow::Continue(())
            });
            if is_streaming {
                return Ok(true);
            }
        }
        Ok(false)
    }

    /// Shared Krishiv catalog backing this engine, if configured.
    pub fn krishiv_catalog(&self) -> Option<&Arc<RwLock<InMemoryCatalog>>> {
        self.krishiv_catalog.as_ref()
    }

    /// Register an Iceberg [`KrishivCatalog`] as a DataFusion catalog provider.
    ///
    /// Tables in the catalog are resolved automatically by DataFusion when SQL
    /// queries reference `<catalog_name>.<namespace>.<table>`. The bridge uses
    /// `plan_files()` to enumerate Parquet files and wraps them in a
    /// `ListingTable`, giving DataFusion native projection/filter pushdown.
    ///
    /// Multiple catalogs can be registered under different names.
    #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
    #[must_use]
    pub fn with_iceberg_catalog(
        self,
        catalog: std::sync::Arc<catalog::unified::KrishivCatalog>,
        catalog_name: impl Into<String>,
    ) -> Self {
        self.register_iceberg_catalog(catalog, catalog_name);
        self
    }

    /// Register an Iceberg [`KrishivCatalog`] on an already-built engine.
    ///
    /// Non-consuming twin of [`Self::with_iceberg_catalog`] for callers that
    /// only hold a shared reference (e.g. the Flight SQL daemon attaching a
    /// platform REST catalog at startup). Invalidates the plan cache so
    /// statements planned before registration cannot pin the old schema view.
    #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
    pub fn register_iceberg_catalog(
        &self,
        catalog: std::sync::Arc<catalog::unified::KrishivCatalog>,
        catalog_name: impl Into<String>,
    ) {
        let catalog_name = catalog_name.into();
        let bridge = catalog::iceberg_catalog_bridge::IcebergCatalogBridge::new(
            Arc::clone(&catalog),
            catalog_name.clone(),
        );
        self.context
            .register_catalog(catalog_name.clone(), Arc::new(bridge));
        self.iceberg_catalogs
            .write()
            .unwrap_or_else(|e| e.into_inner())
            .push((catalog, catalog_name));
        self.invalidate_plan_cache();
    }

    /// Register a platform Iceberg REST catalog from the `KRISHIV_ICEBERG_REST_*`
    /// environment (URI / WAREHOUSE / TOKEN / NAME) when `KRISHIV_ICEBERG_REST_URI`
    /// is set, so governed `catalog.namespace.table` references resolve on *this*
    /// engine. Returns whether a catalog was registered.
    ///
    /// Unlike the Flight host's registration (which only fires in InProcess mode),
    /// this is callable from the per-task engines the executor builds for
    /// coordinated batch SQL — closing the coordinator-mode catalog gap so
    /// `SELECT … FROM krishiv.<ns>.<table>` works on the split (coordinator +
    /// platformd) topology. `s3://` warehouses are read through `KrishivStorage`.
    /// No-op returning `Ok(false)` when built without the `rest-catalog` feature.
    pub async fn register_iceberg_rest_catalog_from_env(&self) -> Result<bool, String> {
        #[cfg(feature = "rest-catalog")]
        {
            let uri = match std::env::var("KRISHIV_ICEBERG_REST_URI") {
                Ok(uri) => uri,
                Err(_) => return Ok(false),
            };
            let warehouse = std::env::var("KRISHIV_ICEBERG_REST_WAREHOUSE").unwrap_or_default();
            let token = std::env::var("KRISHIV_ICEBERG_REST_TOKEN").ok();
            // The platform's canonical governed catalog is `main` (every pipeline
            // SQL, guard, and permission references `main.<ns>.<table>`). Default
            // to that so coordinator-mode `SELECT … FROM main.…` resolves without
            // per-deploy env; `KRISHIV_ICEBERG_REST_NAME` still overrides.
            let name =
                std::env::var("KRISHIV_ICEBERG_REST_NAME").unwrap_or_else(|_| String::from("main"));
            // When the warehouse is object-store-backed, register an S3 store on
            // this engine's DataFusion runtime so the `ListingTable` the catalog
            // bridge builds can *scan* the Parquet data files (iceberg FileIO
            // only covers metadata reads). Without this a correctly-named S3
            // table resolves but fails at scan with "no object store for s3://".
            self.register_s3_object_store_for_warehouse(&warehouse)?;
            let catalog = std::sync::Arc::new(
                catalog::unified::KrishivCatalog::rest(&uri, &warehouse, token.as_deref())
                    .await
                    .map_err(|e| format!("iceberg REST catalog at {uri}: {e}"))?,
            );
            self.register_iceberg_catalog(std::sync::Arc::clone(&catalog), &name);
            // Back-compat alias: parts of the surface (console sample queries,
            // some example jobs, older docs) still qualify governed tables as
            // `krishiv.<ns>.<table>` from a half-finished `krishiv`→`main` rename.
            // Register the same catalog under `krishiv` too so both resolve.
            // Harmless when the primary name already is `krishiv`.
            if name != "krishiv" {
                self.register_iceberg_catalog(catalog, "krishiv");
            }
            Ok(true)
        }
        #[cfg(not(feature = "rest-catalog"))]
        {
            Ok(false)
        }
    }

    /// Share a session UDF registry so scalar UDFs are visible in SQL.
    #[must_use]
    pub fn with_udf_registry(
        mut self,
        registry: std::sync::Arc<std::sync::RwLock<krishiv_plan::udf::UdfRegistry>>,
    ) -> Self {
        self.udf_registry = Some(registry);
        // Mark UDFs as dirty so the first sql() call syncs them.
        self.bump_udf_version();
        self
    }

    /// Signal that the UDF set has changed: `sync_all_udfs()` is needed on the
    /// next `sql()` call, **and** the plan cache is now stale.
    ///
    /// Both consequences belong to the one event, which is why they live in one
    /// place. A cached [`LogicalPlan`](datafusion::logical_expr::LogicalPlan)
    /// pins the UDF it resolved at plan time — `Expr::ScalarFunction` holds an
    /// `Arc<ScalarUDF>`, and a table function is resolved all the way to the
    /// `TableProvider` it returned. So after `CREATE OR REPLACE FUNCTION f …`,
    /// re-running an identical query text would hit the cache and execute the
    /// *old* `f`. Redefinition is the entire point of `OR REPLACE`, so this is
    /// reachable from ordinary SQL.
    ///
    /// Deliberately not placed in the `sync_*_udfs` methods: those are public
    /// and are called when the UDF set has *not* changed (krishiv-api calls
    /// them directly; the `_with_limits` variants re-sync the same set under
    /// different resource limits), and invalidating there would drop the cache
    /// on non-changes.
    pub(crate) fn bump_udf_version(&self) {
        self.udf_registry_version.fetch_add(1, Ordering::Release);
        self.invalidate_plan_cache();
    }

    /// Invalidate the plan cache after any schema change. Call this whenever a
    /// table is registered, replaced, or deregistered. Full invalidation is
    /// simpler and safer than per-table tracking: the cache refills quickly on
    /// the next few queries.
    fn invalidate_plan_cache(&self) {
        match self.plan_cache.lock() {
            Ok(mut cache) => cache.clear(),
            Err(poisoned) => poisoned.into_inner().clear(),
        }
    }

    /// Expose cache invalidation for tests and external callers that register
    /// tables through a different path.
    pub fn clear_plan_cache(&self) {
        self.invalidate_plan_cache();
    }

    /// Extract Python UDF/UDAF directive comments from `sql`, register each on
    /// this engine, and return `sql` with the directives removed. Handles both
    /// `/* krishiv-register-python-udf:name:in,…:out:pickle_b64 */` (scalar, via
    /// [`Self::register_python_udf`]) and
    /// `/* krishiv-register-python-udaf:… */` (aggregate, via
    /// [`Self::register_python_udaf`]). This is how a Python UDF shipped inside a
    /// fragment's SQL reaches the executor's fresh per-task engine. No-op for SQL
    /// that carries none.
    pub async fn register_python_udfs_from_sql(&self, sql: &str) -> SqlResult<String> {
        const SCALAR_PREFIX: &str = "/* krishiv-register-python-udf:";
        const AGG_PREFIX: &str = "/* krishiv-register-python-udaf:";
        if !sql.contains(SCALAR_PREFIX) && !sql.contains(AGG_PREFIX) {
            return Ok(sql.to_string());
        }
        let mut out = String::with_capacity(sql.len());
        let mut rest = sql;
        loop {
            // Find whichever directive appears first (aggregate and scalar
            // prefixes are disjoint: "udaf:" never contains "udf:").
            let agg = rest.find(AGG_PREFIX).map(|i| (i, true, AGG_PREFIX.len()));
            let scalar = rest
                .find(SCALAR_PREFIX)
                .map(|i| (i, false, SCALAR_PREFIX.len()));
            let next = match (agg, scalar) {
                (Some(a), Some(s)) => Some(if a.0 <= s.0 { a } else { s }),
                (Some(a), None) => Some(a),
                (None, Some(s)) => Some(s),
                (None, None) => None,
            };
            let Some((start, is_aggregate, prefix_len)) = next else {
                break;
            };
            out.push_str(&rest[..start]);
            let after = &rest[start + prefix_len..];
            let Some(end) = after.find(" */") else {
                // Unterminated — leave the remainder verbatim and stop.
                out.push_str(&rest[start..]);
                return Ok(out);
            };
            self.register_python_udf_directive(&after[..end], is_aggregate)
                .await?;
            rest = &after[end + " */".len()..];
        }
        out.push_str(rest);
        Ok(out)
    }

    /// Parse and register one `name:in1,in2:out:pickle_b64` directive body as a
    /// scalar or aggregate Python UDF.
    async fn register_python_udf_directive(&self, body: &str, is_aggregate: bool) -> SqlResult<()> {
        use base64::Engine as _;
        let mut parts = body.splitn(4, ':');
        let (name, in_types, out_type, pickle_b64) =
            match (parts.next(), parts.next(), parts.next(), parts.next()) {
                (Some(n), Some(i), Some(o), Some(p)) => (n, i, o, p),
                _ => {
                    return Err(SqlError::DataFusion {
                        message: "malformed python-udf directive".into(),
                    });
                }
            };
        let input_types: Vec<String> = if in_types.is_empty() {
            Vec::new()
        } else {
            in_types.split(',').map(str::to_string).collect()
        };
        let pickle = base64::engine::general_purpose::STANDARD
            .decode(pickle_b64)
            .map_err(|e| SqlError::DataFusion {
                message: format!("invalid python-udf pickle base64: {e}"),
            })?;
        if is_aggregate {
            self.register_python_udaf(name, &pickle, &input_types, out_type)
                .await
        } else {
            self.register_python_udf(name, &pickle, &input_types, out_type)
                .await
        }
    }

    /// Register a distributed Python UDF from cloudpickled bytes. The callable
    /// runs in a shared `python3` worker subprocess ([`crate::python_udf`]) — so
    /// it works on the Rust executors with no embedded interpreter. `input_types`
    /// / `output_type` are Arrow type names. Requires a UDF registry on this
    /// engine (executors' task engines set one).
    pub async fn register_python_udf(
        &self,
        name: &str,
        pickle: &[u8],
        input_types: &[String],
        output_type: &str,
    ) -> SqlResult<()> {
        use arrow::datatypes::{Field, Schema};
        let Some(registry) = &self.udf_registry else {
            return Err(SqlError::DataFusion {
                message: "cannot register a python UDF: engine has no UDF registry".into(),
            });
        };
        let input_fields: Vec<Field> = input_types
            .iter()
            .enumerate()
            .map(|(i, t)| Field::new(format!("a{i}"), python_udf_arrow_type(t), true))
            .collect();
        let input_schema = Schema::new(input_fields);
        let output_field = Field::new("out", python_udf_arrow_type(output_type), true);
        let pool = crate::python_udf::global_pool().map_err(|e| SqlError::DataFusion {
            message: format!("python UDF worker unavailable: {e:?}"),
        })?;
        let udf = std::sync::Arc::new(crate::python_udf::PythonWorkerUdf::new(
            name,
            pickle.to_vec(),
            input_schema,
            output_field,
            pool,
        ));
        registry
            .write()
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?
            .register_scalar(udf);
        self.bump_udf_version();
        self.sync_scalar_udfs().await
    }

    /// Register a distributed Python **aggregate** UDF (GROUPED_AGG) from
    /// cloudpickled bytes. The callable buffers a group's rows and runs once at
    /// finalize inside the shared `python3` worker, so it works on the Rust
    /// executors and merges correctly across partitions/executors (two-phase
    /// aggregation). `input_types`/`output_type` are Arrow type names. Requires
    /// a UDF registry on this engine (executors' task engines set one).
    pub async fn register_python_udaf(
        &self,
        name: &str,
        pickle: &[u8],
        input_types: &[String],
        output_type: &str,
    ) -> SqlResult<()> {
        use arrow::datatypes::{Field, Schema};
        let Some(registry) = &self.udf_registry else {
            return Err(SqlError::DataFusion {
                message: "cannot register a python UDAF: engine has no UDF registry".into(),
            });
        };
        let input_fields: Vec<Field> = input_types
            .iter()
            .enumerate()
            .map(|(i, t)| Field::new(format!("a{i}"), python_udf_arrow_type(t), true))
            .collect();
        let input_schema = Schema::new(input_fields);
        let output_field = Field::new("out", python_udf_arrow_type(output_type), true);
        let pool = crate::python_udf::global_pool().map_err(|e| SqlError::DataFusion {
            message: format!("python UDF worker unavailable: {e:?}"),
        })?;
        let udf = std::sync::Arc::new(crate::python_udf::PythonWorkerAggregateUdf::new(
            name,
            pickle.to_vec(),
            input_schema,
            output_field,
            pool,
        ));
        registry
            .write()
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?
            .register_aggregate(udf);
        self.bump_udf_version();
        self.sync_aggregate_udfs().await
    }

    pub async fn sync_scalar_udfs(&self) -> SqlResult<()> {
        let Some(registry) = &self.udf_registry else {
            return Ok(());
        };
        let guard = registry.read().map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })?;
        let limits = self.udf_limits.clone().unwrap_or_default();
        udf::sync_scalar_udfs_with_limits(&self.context, &guard, limits).map_err(|e| {
            SqlError::DataFusion {
                message: e.to_string(),
            }
        })
    }

    /// Register scalar UDFs with explicit ResourceLimits for sandbox enforcement.
    /// Callers that have a job context (scheduler, runner, api session for a job)
    /// should use this and pass limits derived from the JobSpec (memory + time cap).
    /// This is the concrete Track E seam from job limits to UDF execution.
    pub async fn sync_scalar_udfs_with_limits(
        &self,
        limits: krishiv_plan::udf::ResourceLimits,
    ) -> SqlResult<()> {
        self.sync_scalar_udfs_with_limits_for_profile(
            limits,
            krishiv_common::resolve_durability_profile(),
        )
        .await
    }

    /// Register scalar UDFs using a caller-resolved durability profile.
    pub async fn sync_scalar_udfs_with_limits_for_profile(
        &self,
        limits: krishiv_plan::udf::ResourceLimits,
        profile: krishiv_common::DurabilityProfile,
    ) -> SqlResult<()> {
        self.sync_scalar_udfs_with_limits_for_policy(
            limits,
            krishiv_common::NativeScalarUdfPolicy::resolve(profile),
        )
        .await
    }

    /// Register scalar UDFs using a caller-snapshotted policy decision.
    pub async fn sync_scalar_udfs_with_limits_for_policy(
        &self,
        limits: krishiv_plan::udf::ResourceLimits,
        policy: krishiv_common::NativeScalarUdfPolicy,
    ) -> SqlResult<()> {
        let Some(registry) = &self.udf_registry else {
            return Ok(());
        };
        let guard = registry.read().map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })?;
        udf::sync_scalar_udfs_with_limits_for_policy(&self.context, &guard, limits, policy).map_err(
            |e| SqlError::DataFusion {
                message: e.to_string(),
            },
        )
    }

    /// Register aggregate UDFs from the attached registry (P1-21).
    pub async fn sync_aggregate_udfs(&self) -> SqlResult<()> {
        let Some(registry) = &self.udf_registry else {
            return Ok(());
        };
        let guard = registry.read().map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })?;
        udf::sync_aggregate_udfs(&self.context, &guard).map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })
    }

    /// Register table UDFs from the attached registry (P1-21).
    pub async fn sync_table_udfs(&self) -> SqlResult<()> {
        let Some(registry) = &self.udf_registry else {
            return Ok(());
        };
        let guard = registry.read().map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })?;
        udf::sync_table_udfs(&self.context, &guard).map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })
    }

    /// Sync all UDF categories, respecting any limits configured on this engine (Track E).
    pub async fn sync_all_udfs(&self) -> SqlResult<()> {
        self.sync_scalar_udfs().await?;
        self.sync_aggregate_udfs().await?;
        self.sync_table_udfs().await?;
        Ok(())
    }

    /// Register an S3/MinIO object store on this engine's DataFusion runtime for
    /// an `s3://`/`s3a://`-scheme `path` (a warehouse root or a data-file URI), so
    /// `ListingTable` scans — including those the Iceberg catalog bridge builds
    /// for governed tables — can read Parquet from object storage. No-op for
    /// non-object-store paths. Idempotent: re-registering a bucket replaces the
    /// prior store.
    pub(crate) fn register_s3_object_store_for_warehouse(&self, path: &str) -> Result<(), String> {
        if !(path.starts_with("s3://") || path.starts_with("s3a://")) {
            return Ok(());
        }
        let url = url::Url::parse(path).map_err(|e| format!("invalid s3 url {path}: {e}"))?;
        let bucket = url.host_str().unwrap_or_default();
        // DataFusion keys object stores by scheme+authority (`s3://bucket`).
        let store_url = url::Url::parse(&format!("s3://{bucket}"))
            .map_err(|e| format!("invalid s3 bucket url: {e}"))?;
        let store = build_s3_object_store(bucket).map_err(|e| format!("s3 store init: {e}"))?;
        self.context.register_object_store(&store_url, store);
        Ok(())
    }

    /// Register a local Parquet path as a table.
    pub async fn register_parquet(
        &self,
        table_name: impl AsRef<str>,
        path: impl AsRef<Path>,
    ) -> SqlResult<()> {
        self.register_parquet_with_primary_key(table_name, path, &[] as &[String])
            .await
    }

    /// As [`Self::register_parquet`], declaring a primary key for the table.
    ///
    /// Parquet carries no key, so this declaration is the only way the optimizer
    /// can learn that one column determines the others — which is what unlocks
    /// functional-dependency group-by pruning and the late-materialisation
    /// join-back (`crate::late_materialize`, measured at 14.8x on TPC-H q10 at
    /// SF100).
    ///
    /// **Informational and unverified**, exactly as Spark/Databricks `RELY`:
    /// nothing reads the data to check it, and a key that is not unique or not
    /// non-null produces *wrong answers*, not slow ones.
    ///
    /// This exists so the embedded engine can be told the same thing the
    /// distributed one is told through `BatchSqlTable::primary_key`. A
    /// declaration that reached only one of the two would mean the two
    /// topologies silently plan different queries — the hardest kind of
    /// difference to notice, because both still return correct results.
    pub async fn register_parquet_with_primary_key<S: AsRef<str>>(
        &self,
        table_name: impl AsRef<str>,
        path: impl AsRef<Path>,
        primary_key: &[S],
    ) -> SqlResult<()> {
        let table_name = table_name.as_ref();
        if table_name.trim().is_empty() {
            return Err(SqlError::EmptyTableName);
        }

        let path = path.as_ref().to_string_lossy().into_owned();

        // Register an S3 ObjectStore when the path is an s3:// URL so DataFusion
        // can read remote Parquet files transparently.
        self.register_s3_object_store_for_warehouse(&path)
            .map_err(|message| SqlError::DataFusion { message })?;

        if self
            .context
            .table_exist(table_name)
            .map_err(SqlError::from)?
        {
            let _ = self
                .context
                .deregister_table(table_name)
                .map_err(SqlError::from)?;
        }
        // One registration path, not two. `register_parquet_table` is what the
        // coordinator uses for staged planning; routing the embedded engine
        // through the same function is what keeps "declared key" from meaning
        // two different things depending on where the query runs — including
        // its error for a key column that is not in the file's schema, which
        // would otherwise be a silent no-op here.
        let spec = crate::distributed_plan::ParquetTableSpec::new(table_name, path)
            .with_primary_key(primary_key.iter().map(|c| c.as_ref().to_owned()));
        crate::distributed_plan::register_parquet_table(&self.context, &spec).await?;
        // Extract estimated row count from table provider statistics.
        if let Ok(provider) = self.context.table_provider(table_name).await
            && let Some(stats) = provider.statistics()
            && let Some(n) = stats.num_rows.get_value()
            && let Ok(mut counts) = self.table_row_counts.write()
        {
            counts.insert(table_name.to_string(), *n as u64);
        }
        self.invalidate_plan_cache();
        Ok(())
    }

    /// Create a DataFrame by reading a local Parquet path directly.
    pub async fn read_parquet(&self, path: impl AsRef<Path>) -> SqlResult<SqlDataFrame> {
        let path = path.as_ref().to_string_lossy().into_owned();
        let dataframe = self
            .context
            .read_parquet(path, ParquetReadOptions::default())
            .await?;
        Ok(self.make_sql_df("parquet-read", dataframe))
    }

    /// Register an in-memory table from Arrow record batches.
    ///
    /// The schema is inferred from the first batch. An empty `batches` slice
    /// registers a table with no rows using the provided schema if the batches
    /// are non-empty, or is a no-op if empty.
    pub async fn register_record_batches(
        &self,
        table_name: impl AsRef<str>,
        batches: Vec<RecordBatch>,
    ) -> SqlResult<()> {
        use std::sync::Arc;
        let table_name = table_name.as_ref();
        if table_name.trim().is_empty() {
            return Err(SqlError::EmptyTableName);
        }
        if batches.is_empty() {
            return Ok(());
        }
        let total_rows: usize = batches.iter().map(|b| b.num_rows()).sum();
        let schema = batches
            .first()
            .ok_or_else(|| SqlError::DataFusion {
                message: "empty batch list".into(),
            })?
            .schema();
        let mem_table =
            datafusion::datasource::MemTable::try_new(schema, vec![batches]).map_err(|e| {
                SqlError::DataFusion {
                    message: e.to_string(),
                }
            })?;
        if self
            .context
            .table_exist(table_name)
            .map_err(SqlError::from)?
        {
            let _ = self
                .context
                .deregister_table(table_name)
                .map_err(SqlError::from)?;
        }
        self.context
            .register_table(table_name, Arc::new(mem_table))
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;
        if total_rows > 0
            && let Ok(mut counts) = self.table_row_counts.write()
        {
            counts.insert(table_name.to_string(), total_rows as u64);
        }
        self.invalidate_plan_cache();
        Ok(())
    }

    /// Create a DataFrame by reading a local Parquet path with typed options.
    pub async fn read_parquet_with_options(
        &self,
        path: impl AsRef<Path>,
        opts: &ParquetReaderOptions,
    ) -> SqlResult<SqlDataFrame> {
        let path = path.as_ref().to_string_lossy().into_owned();
        let mut options = datafusion::prelude::ParquetReadOptions::default();
        if opts.batch_size.is_some() {
            options = options.parquet_pruning(true);
        }
        // NOTE: `batch_size` is not yet propagated here because DataFusion's
        // ParquetReadOptions has no batch_size field — it lives on SessionConfig.
        // Callers should set batch_size on the SqlEngine's session config before
        // calling this method (via `SessionContext::new_with_state` with a config
        // that has `execution.batch_size` set).
        let dataframe = self.context.read_parquet(path, options).await?;
        Ok(self.make_sql_df("parquet-read", dataframe))
    }

    /// Create a DataFrame by reading a local CSV path directly.
    pub async fn read_csv(&self, path: impl AsRef<Path>) -> SqlResult<SqlDataFrame> {
        self.read_csv_with_options(path, &CsvReaderOptions::default())
            .await
    }

    /// Create a DataFrame by reading a local CSV path with typed options.
    pub async fn read_csv_with_options(
        &self,
        path: impl AsRef<Path>,
        opts: &CsvReaderOptions,
    ) -> SqlResult<SqlDataFrame> {
        let path = path.as_ref().to_string_lossy().into_owned();
        let mut options = datafusion::prelude::CsvReadOptions::new();
        if let Some(delim) = opts.delimiter {
            options = options.delimiter(delim as u8);
        }
        if let Some(has_header) = opts.has_header {
            options = options.has_header(has_header);
        }
        let dataframe = self.context.read_csv(path, options).await?;
        Ok(self.make_sql_df("csv-read", dataframe))
    }

    /// Create a DataFrame by reading a local JSON/NDJSON path directly.
    pub async fn read_json(&self, path: impl AsRef<Path>) -> SqlResult<SqlDataFrame> {
        let path = path.as_ref().to_string_lossy().into_owned();
        let dataframe = self
            .context
            .read_json(path, datafusion::prelude::JsonReadOptions::default())
            .await?;
        Ok(self.make_sql_df("json-read", dataframe))
    }

    /// Read a local Delta table directory into a DataFrame.
    pub async fn read_delta(
        &self,
        path: impl AsRef<str>,
        version: Option<i64>,
    ) -> SqlResult<SqlDataFrame> {
        let path = path.as_ref();
        let base = path.replace(['/', '.', '-'], "_");
        let table = match version {
            Some(v) => format!("delta_{base}_v{v}"),
            None => format!("delta_{base}"),
        };
        lakehouse::register_delta_uri(&self.context, &table, path, version).await?;
        self.sql(format!("SELECT * FROM {table}")).await
    }

    /// Read a Hudi table directory.
    pub async fn read_hudi(
        &self,
        path: impl AsRef<str>,
        query_type: krishiv_connectors::lakehouse::HudiQueryType,
        begin_instant: Option<&str>,
    ) -> SqlResult<SqlDataFrame> {
        let path = path.as_ref();
        let table = format!("hudi_{}", path.replace(['/', '.', '-'], "_"));
        lakehouse::register_hudi_uri(&self.context, &table, path, query_type, begin_instant)
            .await?;
        self.sql(format!("SELECT * FROM {table}")).await
    }

    /// Plan a SQL query with DataFusion.
    ///
    /// Returns a boxed future ON PURPOSE (do not revert to `async fn`): this
    /// body is one of the largest async state machines in the workspace, and
    /// as an opaque `impl Future` its `Send` proof leaked into every CONSUMER
    /// crate — rustc's old trait solver re-derived it per call site, which
    /// drove krishiv-executor's front-end to 20+ hour compiles (95% of CPU in
    /// `ObligationForest::process_obligations`, bisected 2026-07-19). Boxing
    /// at the definition pays that proof once, here.
    pub fn sql<'a>(
        &'a self,
        query: impl AsRef<str> + Send + 'a,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = SqlResult<SqlDataFrame>> + Send + 'a>>
    {
        let query: String = query.as_ref().to_owned();
        Box::pin(self.sql_boxed_body(query))
    }

    /// The real `sql` body — see the boxing note on [`Self::sql`].
    async fn sql_boxed_body(&self, query: String) -> SqlResult<SqlDataFrame> {
        let query = query.as_str();
        if query.trim().is_empty() {
            return Err(SqlError::EmptyQuery);
        }

        // ── Multi-statement scripts ──────────────────────────────────────────
        // Distributed batch fragments are re-planned on a fresh engine per
        // assignment, so session state registered by an earlier Flight SQL
        // call (e.g. a `STORED AS JDBC` pull table) does not exist there. A
        // caller that needs setup DDL plus the statement that reads it must
        // ship both in one body; statements run sequentially in this engine
        // and the last statement's result is returned. A failing statement
        // aborts the script (statements already executed are not rolled back).
        let script = split_sql_statements(query);
        if script.len() > 1
            && let [setup @ .., last_stmt] = script.as_slice()
        {
            for stmt in setup {
                // DDL executes eagerly inside `sql()`, but DML comes back as a
                // lazy frame that only runs on collect — force each setup
                // statement to completion before anything that depends on it.
                Box::pin(self.sql(stmt.as_str())).await?.collect().await?;
            }
            return Box::pin(self.sql(last_stmt.as_str())).await;
        }

        // Lazy UDF sync: only re-sync when the registry has changed since the
        // last sync. Avoids 3 RwLock reads per query when no UDFs are registered
        // or when the UDF set hasn't changed.
        {
            let current = self.udf_registry_version.load(Ordering::Acquire);
            let last = self.udf_last_synced_version.load(Ordering::Relaxed);
            if current != last {
                self.sync_all_udfs().await?;
                self.udf_last_synced_version
                    .store(current, Ordering::Release);
            }
        }

        // ── Intercept DESCRIBE / SHOW COLUMNS / EXPLAIN ──────────────────────
        if let Some(stmt) = introspection_sql::parse_introspection_statement(query)? {
            return match stmt {
                introspection_sql::IntrospectionStatement::Describe { table } => {
                    let batch = introspection_sql::describe_table(&self.context, &table).await?;
                    let describe_table_name = next_ephemeral_name("describe_result");
                    lakehouse::register_scan_batches(
                        &self.context,
                        &describe_table_name,
                        vec![batch],
                    )
                    .await?;
                    let dataframe = self
                        .context
                        .sql(&format!("SELECT * FROM {describe_table_name}"))
                        .await?;
                    Ok(self.attach_query_metadata(self.make_sql_df("describe", dataframe), query))
                }
                introspection_sql::IntrospectionStatement::Explain { mode, query: inner } => {
                    let text = introspection_sql::explain_query(&inner, mode)?;
                    let batch = introspection_sql::explain_result_batch(&text)?;
                    let explain_table = next_ephemeral_name("explain_result");
                    lakehouse::register_scan_batches(&self.context, &explain_table, vec![batch])
                        .await?;
                    let dataframe = self
                        .context
                        .sql(&format!("SELECT * FROM {explain_table}"))
                        .await?;
                    Ok(self.attach_query_metadata(self.make_sql_df("explain", dataframe), query))
                }
            };
        }

        // ── Intercept CREATE / REFRESH / DROP LIVE TABLE ─────────────────────
        if live_table::execute_live_table_ddl(&self.live_table_registry, query)?.is_some() {
            let empty = self.context.sql("SELECT 1 WHERE FALSE").await?;
            return Ok(self.attach_query_metadata(self.make_sql_df("live-table-ddl", empty), query));
        }

        // ── Intercept CREATE/DECLARE/REFRESH/DROP INCREMENTAL VIEW ───────────
        match incremental_view::execute_incremental_view_ddl(
            &self.incremental_view_registry,
            query,
        )? {
            Some(incremental_view::IncrementalViewResult::Refresh(_name)) => {
                // REFRESH requires the caller (Session) to re-run the pipeline.
                // Return a sentinel empty result so the caller knows to refresh.
                let empty = self.context.sql("SELECT 1 WHERE FALSE").await?;
                return Ok(self.attach_query_metadata(
                    self.make_sql_df("incremental-view-refresh", empty),
                    query,
                ));
            }
            Some(_) => {
                let empty = self.context.sql("SELECT 1 WHERE FALSE").await?;
                return Ok(self.attach_query_metadata(
                    self.make_sql_df("incremental-view-ddl", empty),
                    query,
                ));
            }
            None => {}
        }

        // ── Intercept CREATE STREAMING TABLE … AS <streaming SELECT> ─────────
        // The body is validated through the shared streaming front door, so an
        // unsupported query fails at the planner. Running the job needs the
        // streaming coordinator, which the pure SQL engine does not have, so we
        // surface a clear, actionable error rather than a cryptic parse failure;
        // a cluster-attached session submits the validated plan via the
        // continuous-stream registration API.
        if let Some(ddl) = streaming_table_ddl::parse_create_streaming_table(query) {
            let _plan = streaming_window_plan::compile_streaming_window_sql(&ddl.query)?;
            return Err(SqlError::Unsupported {
                feature: format!(
                    "CREATE STREAMING TABLE '{}' compiled to a continuous plan, but this session \
                     has no streaming coordinator to run it; submit it via the continuous-stream \
                     registration API or a cluster-attached session",
                    ddl.name
                ),
            });
        }

        // ── Intercept CREATE/DROP SOURCE / SINK (pipeline DDL) ───────────────
        // `START PIPELINE` is NOT handled here — it is executed by the
        // `krishiv-api` session, which can reach `Session::pipeline()`.
        if pipeline_ddl::execute_pipeline_ddl(&self.pipeline_registry, query)?.is_some() {
            let empty = self.context.sql("SELECT 1 WHERE FALSE").await?;
            return Ok(self.attach_query_metadata(self.make_sql_df("pipeline-ddl", empty), query));
        }

        // ── Intercept SET shuffle.partitions = N ─────────────────────────────
        // Krishiv-specific session config; DataFusion does not know about it.
        let trimmed = query.trim();
        if trimmed
            .to_ascii_uppercase()
            .starts_with("SET SHUFFLE.PARTITIONS")
        {
            let value = trimmed.split('=').nth(1).map(|s| s.trim()).unwrap_or("");
            match value.parse::<u32>() {
                Ok(n) if n > 0 => {
                    {
                        let mut guard =
                            self.shuffle_partitions
                                .write()
                                .map_err(|e| SqlError::DataFusion {
                                    message: e.to_string(),
                                })?;
                        *guard = Some(n);
                    }
                    let empty = self.context.sql("SELECT 1 WHERE FALSE").await?;
                    return Ok(self.make_sql_df("set-shuffle-partitions", empty));
                }
                Ok(_) => {
                    {
                        let mut guard =
                            self.shuffle_partitions
                                .write()
                                .map_err(|e| SqlError::DataFusion {
                                    message: e.to_string(),
                                })?;
                        *guard = None;
                    }
                    let empty = self.context.sql("SELECT 1 WHERE FALSE").await?;
                    return Ok(self.make_sql_df("set-shuffle-partitions", empty));
                }
                Err(_) => {
                    return Err(SqlError::DataFusion {
                        message: format!(
                            "invalid shuffle.partitions value '{value}'; expected a positive integer"
                        ),
                    });
                }
            }
        }

        // ── Intercept USE / SHOW DATABASES (Phase 60 statement completion) ───
        // DataFusion does not plan `USE` (session catalog/schema navigation) or
        // `SHOW DATABASES`/`SHOW SCHEMAS`; handle both before the fallthrough.
        if let Some(result) = statement_completion::apply_use(&self.context, query) {
            result.map_err(|message| SqlError::DataFusion { message })?;
            let empty = self.context.sql("SELECT 1 WHERE FALSE").await?;
            return Ok(self.attach_query_metadata(self.make_sql_df("use", empty), query));
        }
        if let Some(rewrite) = statement_completion::rewrite_show_databases(query) {
            let dataframe = self.context.sql(&rewrite).await?;
            return Ok(
                self.attach_query_metadata(self.make_sql_df("show-databases", dataframe), query)
            );
        }

        // ── Intercept CREATE FUNCTION … RETURNS TABLE ────────────────────────
        // DataFusion does not understand this extended DDL syntax. Parse and
        // register only executable LANGUAGE SQL definitions; unsupported
        // languages fail before any registry mutation.
        if create_function_ddl::is_create_function_returns_table(query) {
            let ddl = create_function_ddl::parse_create_function(query)
                .map_err(|message| SqlError::InvalidTableFunction { message })?;
            if ddl.language.as_deref() != Some("sql") {
                return Err(SqlError::Unsupported {
                    feature: format!(
                        "CREATE FUNCTION '{}' uses language {:?}; only LANGUAGE SQL AS '...' \
                         table functions are executable",
                        ddl.function_name, ddl.language
                    ),
                });
            }
            let body = ddl
                .body
                .as_deref()
                .filter(|body| !body.trim().is_empty())
                .ok_or_else(|| SqlError::InvalidTableFunction {
                    message: format!(
                        "SQL table function '{}' requires a non-empty AS body",
                        ddl.function_name
                    ),
                })?;
            let fields: Vec<_> = ddl
                .return_columns
                .iter()
                .map(|column| {
                    arrow::datatypes::Field::new(&column.name, column.data_type.clone(), true)
                })
                .collect();
            let schema = arrow::datatypes::Schema::new(fields);
            let udf: std::sync::Arc<dyn krishiv_plan::udf::TableUdf> = std::sync::Arc::new(
                create_function_ddl::SqlBodyTableUdf::try_new(
                    &ddl.function_name,
                    schema,
                    body,
                    ddl.arguments.len(),
                    std::sync::Arc::new(self.context.clone()),
                )
                .map_err(|error| SqlError::InvalidTableFunction {
                    message: error.to_string(),
                })?,
            );
            if let Some(registry) = &self.udf_registry {
                let mut guard = registry.write().map_err(|e| SqlError::DataFusion {
                    message: e.to_string(),
                })?;
                guard.register_table(std::sync::Arc::clone(&udf));
            }
            udf::register_single_table_udf(&self.context, std::sync::Arc::clone(&udf))
                .map_err(SqlError::from)?;
            // `OR REPLACE` redefines an existing function: drop any cached plan
            // that resolved the previous definition.
            self.bump_udf_version();
            let empty = self.context.sql("SELECT 1 WHERE FALSE").await?;
            return Ok(
                self.attach_query_metadata(self.make_sql_df("create-function", empty), query)
            );
        }

        if query
            .trim_start()
            .to_ascii_uppercase()
            .starts_with("MERGE INTO")
        {
            let batches = lakehouse::execute_merge_sql(&self.context, query).await?;
            let merge_table = next_ephemeral_name("merge_result");
            lakehouse::register_scan_batches(&self.context, &merge_table, batches).await?;
            let dataframe = self
                .context
                .sql(&format!("SELECT * FROM {merge_table}"))
                .await?;
            return Ok(self.attach_query_metadata(self.make_sql_df("merge", dataframe), query));
        }

        // ── Intercept CREATE [OR REPLACE] TABLE <iceberg-table> AS <query> ───
        // Durable CTAS (gap G17): when the target resolves to a registered
        // Iceberg catalog, execute the inner query on this engine and land
        // the result stream directly in Iceberg (rolling Parquet parts +
        // snapshot commit) instead of materializing it as a session table.
        // The result is a single row of landing counts — the full result set
        // never crosses a wire. Targets that do not resolve to an Iceberg
        // catalog fall through to DataFusion's session-local CTAS.
        #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
        if trimmed.to_ascii_uppercase().starts_with("CREATE ")
            && let Some(parsed_ctas) = parse_ctas(trimmed)
        {
            let resolved = self.resolve_iceberg_table(&parsed_ctas.table_ref);
            // PARTITIONED BY only has meaning for Iceberg targets; erroring
            // beats silently creating an unpartitioned session table.
            if resolved.is_none() && !parsed_ctas.partition_by.is_empty() {
                return Err(SqlError::DataFusion {
                    message: format!(
                        "PARTITIONED BY requires an Iceberg catalog table; `{}` does not \
                         resolve to a registered Iceberg catalog",
                        parsed_ctas.table_ref
                    ),
                });
            }
            if let Some((iceberg_catalog, table_ident)) = resolved {
                return self
                    .execute_iceberg_ctas(iceberg_catalog, table_ident, parsed_ctas, query)
                    .await;
            }
        }

        // ── Intercept CALL system.<proc> ──────────────────────────────────────
        // Route Iceberg maintenance procedures to registered KrishivCatalogs.
        #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
        if trimmed.to_ascii_uppercase().starts_with("CALL SYSTEM.") {
            let result = self.dispatch_call_system(trimmed).await?;
            let call_table = next_ephemeral_name("call_result");
            lakehouse::register_scan_batches(&self.context, &call_table, vec![result]).await?;
            let dataframe = self
                .context
                .sql(&format!("SELECT * FROM {call_table}"))
                .await?;
            return Ok(self.attach_query_metadata(self.make_sql_df("call", dataframe), query));
        }

        // ── Intercept ANALYZE TABLE <ref> [FOR COLUMNS (c1, …)] ──────────────
        // Phase 54 statistics collection: one scan (COUNT(*) plus optional
        // per-column approx_distinct/min/max/null-count) feeding the engine's
        // row-count registry (BroadcastAutoRule) and the process-global
        // TableStatsRegistry (CBO / AQE cost model).
        if trimmed
            .get(..14)
            .is_some_and(|p| p.eq_ignore_ascii_case("ANALYZE TABLE "))
        {
            let result = self.dispatch_analyze_table(trimmed).await?;
            let res_table = next_ephemeral_name("analyze_result");
            lakehouse::register_scan_batches(&self.context, &res_table, vec![result]).await?;
            let dataframe = self
                .context
                .sql(&format!("SELECT * FROM {res_table}"))
                .await?;
            return Ok(self.attach_query_metadata(self.make_sql_df("analyze", dataframe), query));
        }

        // ── Intercept DELETE FROM <iceberg-table> [WHERE …] ──────────────────
        // Route to copy-on-write iceberg_delete_where when the table is tracked
        // by a registered KrishivCatalog. Falls through to DataFusion otherwise.
        #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
        if trimmed.to_ascii_uppercase().starts_with("DELETE FROM ")
            && let Some((table_ref, predicate)) = parse_dml_delete(trimmed)
            && let Some((iceberg_catalog, table_ident)) = self.resolve_iceberg_table(&table_ref)
        {
            use arrow::array::{ArrayRef, Int64Array};
            use arrow::datatypes::{DataType, Field, Schema};
            let (deleted, _) = krishiv_connectors::lakehouse::dml::iceberg_delete_where(
                iceberg_catalog,
                &table_ident,
                &predicate,
                &self.context,
            )
            .await
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;
            // Phase 54 auto-stats: keep any known row count in step.
            self.adjust_table_row_count_stat(&table_ref, -(deleted as i64));
            let schema = Arc::new(Schema::new(vec![Field::new(
                "deleted_rows",
                DataType::Int64,
                false,
            )]));
            let array: ArrayRef = Arc::new(Int64Array::from(vec![deleted as i64]));
            let batch =
                RecordBatch::try_new(schema, vec![array]).map_err(|e| SqlError::DataFusion {
                    message: e.to_string(),
                })?;
            let res_table = next_ephemeral_name("delete_result");
            lakehouse::register_scan_batches(&self.context, &res_table, vec![batch]).await?;
            let dataframe = self
                .context
                .sql(&format!("SELECT * FROM {res_table}"))
                .await?;
            return Ok(self.attach_query_metadata(self.make_sql_df("delete", dataframe), query));
        }

        // ── Intercept UPDATE <iceberg-table> SET … [WHERE …] ─────────────────
        #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
        if trimmed.to_ascii_uppercase().starts_with("UPDATE ")
            && let Some(parsed) = parse_dml_update(trimmed)
            && let Some((iceberg_catalog, table_ident)) =
                self.resolve_iceberg_table(&parsed.table_ref)
        {
            use arrow::array::{ArrayRef, Int64Array};
            use arrow::datatypes::{DataType, Field, Schema};
            let borrowed: Vec<(&str, &str)> = parsed
                .assignments
                .iter()
                .map(|(c, e)| (c.as_str(), e.as_str()))
                .collect();
            let pred = parsed.predicate.as_deref();
            let (updated, _) = krishiv_connectors::lakehouse::dml::iceberg_update_where(
                iceberg_catalog,
                &table_ident,
                &borrowed,
                pred,
                &self.context,
            )
            .await
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;
            let schema = Arc::new(Schema::new(vec![Field::new(
                "updated_rows",
                DataType::Int64,
                false,
            )]));
            let array: ArrayRef = Arc::new(Int64Array::from(vec![updated as i64]));
            let batch =
                RecordBatch::try_new(schema, vec![array]).map_err(|e| SqlError::DataFusion {
                    message: e.to_string(),
                })?;
            let res_table = next_ephemeral_name("update_result");
            lakehouse::register_scan_batches(&self.context, &res_table, vec![batch]).await?;
            let dataframe = self
                .context
                .sql(&format!("SELECT * FROM {res_table}"))
                .await?;
            return Ok(self.attach_query_metadata(self.make_sql_df("update", dataframe), query));
        }

        // ── Intercept INSERT INTO <iceberg-table> [SELECT|VALUES ...] ────────
        // #219: DataFusion's own ListingTable::insert_into rejects every
        // catalog table deterministically — each data file parses to an
        // exact object path, never a URL ending in `/`, so its "backed by a
        // single file" collection check always fails, table-exists-or-not.
        // Route the common form (no explicit column list, i.e. all columns
        // in table order) to a durable append landing instead. An explicit
        // column list falls through to DataFusion's existing (rejecting)
        // path — narrower form, not yet supported here (residual).
        #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
        if trimmed.to_ascii_uppercase().starts_with("INSERT ")
            && let Some(parsed) = parse_dml_insert(trimmed)
            && parsed.columns.is_empty()
            && let Some((iceberg_catalog, table_ident)) =
                self.resolve_iceberg_table(&parsed.table_ref)
        {
            use arrow::array::{ArrayRef, Int64Array};
            use arrow::datatypes::{DataType, Field, Schema};
            let source_df = self.context.sql(&parsed.inner_query).await?;
            let stream = source_df
                .execute_stream()
                .await
                .map_err(|e| SqlError::DataFusion {
                    message: e.to_string(),
                })?;
            let report = krishiv_connectors::lakehouse::dml::iceberg_append_into(
                iceberg_catalog,
                &table_ident,
                stream,
            )
            .await
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;
            // Phase 54 auto-stats: keep any known row count in step.
            self.adjust_table_row_count_stat(&parsed.table_ref, report.rows as i64);
            let schema = Arc::new(Schema::new(vec![Field::new(
                "inserted_rows",
                DataType::Int64,
                false,
            )]));
            let array: ArrayRef = Arc::new(Int64Array::from(vec![report.rows as i64]));
            let batch =
                RecordBatch::try_new(schema, vec![array]).map_err(|e| SqlError::DataFusion {
                    message: e.to_string(),
                })?;
            let res_table = next_ephemeral_name("insert_result");
            lakehouse::register_scan_batches(&self.context, &res_table, vec![batch]).await?;
            let dataframe = self
                .context
                .sql(&format!("SELECT * FROM {res_table}"))
                .await?;
            return Ok(self.attach_query_metadata(self.make_sql_df("insert", dataframe), query));
        }

        // ── Intercept MATCH_RECOGNIZE ─────────────────────────────────────────
        // DataFusion does not parse MATCH_RECOGNIZE. Route it through the CEP
        // path: parse → run PatternMatcher on the source table → return results.
        if query.to_ascii_uppercase().contains(" MATCH_RECOGNIZE ")
            && let Some(stmt) = cep_sql::parse_match_recognize(query)?
        {
            let is_streaming = self.is_streaming_source(&stmt.source_table);
            // For streaming sources collect a bounded window of recent events
            // (capped at the configured limit) so the query terminates. The
            // cap is configurable through `KRISHIV_MATCH_RECOGNIZE_STREAMING_LIMIT`
            // (default 100_000) so users can raise it for high-rate streams
            // or lower it to bound memory on small executors. The truncation
            // is logged at warn level because the result is no longer a
            // complete match over the unbounded stream.
            let streaming_limit = streaming_match_recognize_limit_from_env();
            let source_sql = if is_streaming {
                format!(
                    "SELECT * FROM {} LIMIT {}",
                    stmt.source_table, streaming_limit
                )
            } else {
                format!("SELECT * FROM {}", stmt.source_table)
            };
            let source_df = self.context.sql(&source_sql).await?;
            let source_batches = source_df.collect().await?;
            if is_streaming {
                tracing::warn!(
                    source = %stmt.source_table,
                    limit = streaming_limit,
                    collected_rows = source_batches.iter().map(|b| b.num_rows()).sum::<usize>(),
                    "MATCH_RECOGNIZE executed against a streaming source under \
                     bounded materialisation; results only cover the first {0} rows \
                     of the source. Set KRISHIV_MATCH_RECOGNIZE_STREAMING_LIMIT to a \
                     larger value if your executor has the memory budget.",
                    streaming_limit
                );
            }
            let results = cep_sql::execute_match_recognize(stmt, &source_batches)?;
            let cep_table = next_ephemeral_name("cep_result");
            lakehouse::register_scan_batches(&self.context, &cep_table, results).await?;
            let dataframe = self
                .context
                .sql(&format!("SELECT * FROM {cep_table}"))
                .await?;
            return Ok(self.attach_query_metadata(self.make_sql_df("cep", dataframe), query));
        }

        // Rewrite PIVOT / UNPIVOT into equivalent CASE WHEN / UNION ALL SQL —
        // DataFusion does not parse either construct natively.
        // Pipe syntax (`FROM t |> WHERE x |> SELECT y`) lowers to standard SQL
        // before every other rewrite, since those all expect standard SQL.
        // Gated on a leading `FROM ` *and* a `|>`, so no ordinary statement
        // reaches it. This module existed but was never declared as a module —
        // it was not compiled, its tests never ran, and the documented feature
        // was silently absent.
        let piped = pipe_syntax::process_pipe_syntax(query).map_err(|error| {
            SqlError::Unsupported {
                feature: error.to_string(),
            }
        })?;
        let query: &str = &piped;

        let query = &pivot_sql::rewrite_pivot_unpivot(query)?;

        // Rewrite TUMBLE/HOP/SESSION TVFs before other preprocessing.
        let query = &streaming_tvf::rewrite_window_tvfs(query);

        let (rewritten, as_ofs) =
            lakehouse::preprocess_as_of_sql(query).unwrap_or_else(|_| (query.to_string(), vec![]));
        lakehouse::apply_as_of_refs(&self.context, &as_ofs).await?;

        // ── Plan cache ────────────────────────────────────────────────────────
        // Check the cache before sending the query through DataFusion's full
        // parse → analyse → optimise pipeline. Only cache simple queries without
        // DDL or AS-OF refs; DDL side effects must not be bypassed.
        // Single-lock design: lookup and insert share the same Mutex<PlanCache>,
        // eliminating the TOCTOU race of the previous DashMap + VecDeque approach.
        let can_cache = as_ofs.is_empty();
        let shuffle_override = self
            .shuffle_partitions
            .read()
            .map(|g| *g)
            .unwrap_or_else(|e| *e.into_inner());
        if can_cache {
            // Scope the guard so it is dropped before any .await point.
            let cached_plan: Option<datafusion::logical_expr::LogicalPlan> = self
                .plan_cache
                .lock()
                .unwrap_or_else(|e| e.into_inner())
                .get(&rewritten)
                .cloned();
            if let Some(plan) = cached_plan {
                let dataframe = self.context.execute_logical_plan(plan).await?;
                return Ok(self.attach_query_metadata(
                    self.make_sql_df("sql-query", dataframe)
                        .with_shuffle_partitions(shuffle_override),
                    &rewritten,
                ));
            }
        }

        // Register the backing S3 object store before executing a
        // `CREATE EXTERNAL TABLE … LOCATION 's3://…'` DDL. DataFusion infers the
        // schema by listing the location during DDL execution, which needs the
        // object store registered on the runtime env first. The `register_parquet`
        // path already does this; DDL-created external tables did not
        // (engine-s3-ddl-gap), so an `s3://` external table previously failed at
        // plan time with "no suitable object store found for s3://…". No-op for
        // file/connector locations, so it is safe for every external-table DDL.
        if let Some(location) = extract_create_external_table_location(&rewritten) {
            self.register_s3_object_store_for_warehouse(&location)
                .map_err(|message| SqlError::DataFusion { message })?;
        }

        let dataframe = self.context.sql(&rewritten).await?;

        // After CREATE EXTERNAL TABLE DDL, try to extract row-count statistics
        // from the newly registered table provider so `BroadcastAutoRule` can
        // fire for small connector-backed tables (e.g. Parquet/S3 via DDL).
        if let Some(table_name) = extract_create_external_table_name(&rewritten)
            && !table_name.is_empty()
            && let Ok(provider) = self.context.table_provider(&table_name).await
        {
            let maybe_rows = provider
                .statistics()
                .and_then(|s| s.num_rows.get_value().copied());
            if let Some(n) = maybe_rows
                && let Ok(mut counts) = self.table_row_counts.write()
            {
                counts.entry(table_name).or_insert(n as u64);
            }
        }

        // Cache the logical plan for future repeated calls.
        if can_cache {
            let plan = dataframe.logical_plan().clone();
            match self.plan_cache.lock() {
                Ok(mut cache) => cache.insert(rewritten.clone(), plan),
                Err(poisoned) => poisoned.into_inner().insert(rewritten.clone(), plan),
            }
        }

        Ok(self.attach_query_metadata(
            self.make_sql_df("sql-query", dataframe)
                .with_shuffle_partitions(shuffle_override),
            &rewritten,
        ))
    }

    /// Execute a SQL query with a timeout.
    ///
    /// Returns [`SqlError::Timeout`] if `timeout_ms` elapses before the query
    /// produces a result.  The underlying DataFusion task is abandoned (not
    /// cancelled at the engine level) when the timeout fires; its resources are
    /// released when the spawned task eventually completes.
    pub async fn execute_with_timeout(
        &self,
        query: impl AsRef<str> + Send,
        timeout_ms: u64,
    ) -> SqlResult<SqlDataFrame> {
        let timeout = std::time::Duration::from_millis(timeout_ms);
        tokio::time::timeout(timeout, self.sql(query))
            .await
            .map_err(|_| SqlError::Timeout { timeout_ms })?
    }

    /// Execute a SQL query tagged with a caller-supplied operation ID.
    ///
    /// The operation ID is recorded in the returned [`TaggedQueryResult`] and
    /// can be used to correlate logs, metrics, and cancellation requests.
    /// If `cancelled_ids` contains `operation_id` before execution begins the
    /// function returns [`SqlError::OperationCancelled`] immediately.
    pub async fn execute_with_operation_id(
        &self,
        operation_id: u64,
        query: impl AsRef<str> + Send,
        cancelled_ids: &OperationRegistry,
    ) -> SqlResult<TaggedQueryResult> {
        if cancelled_ids.is_cancelled(operation_id) {
            return Err(SqlError::OperationCancelled { operation_id });
        }
        let df = self.sql(query).await?;
        Ok(TaggedQueryResult {
            operation_id,
            inner: df,
        })
    }

    /// Resolve a SQL table reference to an `(Arc<dyn Catalog>, TableIdent)` pair
    /// from the registered Iceberg catalogs.
    ///
    /// Accepts 2-part (`ns.tbl`) and 3-part (`cat.ns.tbl`) references.
    /// Returns `None` when no catalog is registered or the reference is ambiguous.
    #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
    fn resolve_iceberg_table(
        &self,
        table_ref: &str,
    ) -> Option<(Arc<dyn iceberg::Catalog + Send + Sync>, iceberg::TableIdent)> {
        let parts: Vec<&str> = table_ref.splitn(3, '.').collect();
        let (catalog_arc, ns_str, table_str) = {
            let guard = self
                .iceberg_catalogs
                .read()
                .unwrap_or_else(|e| e.into_inner());
            if guard.is_empty() {
                return None;
            }
            match parts.len() {
                2 => {
                    let (cat, _) = guard.first()?;
                    (Arc::clone(cat), *parts.first()?, *parts.get(1)?)
                }
                3 => {
                    let cat_name = parts.first().copied()?;
                    let (cat, _) = guard.iter().find(|(_, n)| n == cat_name)?;
                    (Arc::clone(cat), *parts.get(1)?, *parts.get(2)?)
                }
                _ => return None,
            }
        };
        let ns = iceberg::NamespaceIdent::from_vec(vec![ns_str.to_string()]).ok()?;
        let ident = iceberg::TableIdent::new(ns, table_str.to_string());
        Some((catalog_arc.as_iceberg(), ident))
    }

    /// Execute a durable Iceberg CTAS: run the inner query on this engine
    /// and land the result stream directly in the Iceberg table (rolling
    /// Parquet parts fanned out per `PARTITIONED BY` value + one snapshot
    /// commit). The result is a single row of landing counts — the full
    /// result set never crosses a wire (gap G17).
    #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
    async fn execute_iceberg_ctas(
        &self,
        iceberg_catalog: Arc<dyn iceberg::Catalog + Send + Sync>,
        table_ident: iceberg::TableIdent,
        parsed_ctas: ParsedCtas,
        query: &str,
    ) -> SqlResult<SqlDataFrame> {
        use arrow::array::{ArrayRef, Int64Array};
        use arrow::datatypes::{DataType, Field, Schema};
        use krishiv_connectors::lakehouse::partitioned_write::parse_partition_transform;

        let partition_by = parsed_ctas
            .partition_by
            .iter()
            .map(|item| parse_partition_transform(item))
            .collect::<Result<Vec<_>, _>>()
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;

        let dataframe = self.context.sql(&parsed_ctas.inner_query).await?;
        let stream = dataframe
            .execute_stream()
            .await
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;
        let report = krishiv_connectors::lakehouse::dml::land_ctas(
            iceberg_catalog,
            &table_ident,
            parsed_ctas.or_replace,
            &partition_by,
            stream,
        )
        .await
        .map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })?;
        // The target table (or its schema) changed under any cached plan.
        self.invalidate_plan_cache();
        // Phase 54 auto-stats: the landing report gives an exact row count.
        self.record_table_row_count_stat(&parsed_ctas.table_ref, report.rows as u64);

        let schema = Arc::new(Schema::new(vec![
            Field::new("rows_written", DataType::Int64, false),
            Field::new("bytes_written", DataType::Int64, false),
            Field::new("data_files", DataType::Int64, false),
            Field::new("snapshot_id", DataType::Int64, false),
        ]));
        let columns: Vec<ArrayRef> = vec![
            Arc::new(Int64Array::from(vec![report.rows as i64])),
            Arc::new(Int64Array::from(vec![report.bytes as i64])),
            Arc::new(Int64Array::from(vec![report.data_files as i64])),
            Arc::new(Int64Array::from(vec![report.snapshot_id])),
        ];
        let batch = RecordBatch::try_new(schema, columns).map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })?;
        let res_table = next_ephemeral_name("ctas_result");
        lakehouse::register_scan_batches(&self.context, &res_table, vec![batch]).await?;
        let dataframe = self
            .context
            .sql(&format!("SELECT * FROM {res_table}"))
            .await?;
        Ok(self.attach_query_metadata(self.make_sql_df("ctas", dataframe), query))
    }

    /// Phase 54 auto-stats: record an absolute `row_count` for `table_ref`
    /// in both the engine row-count registry and the process-global stats
    /// registry. Called from write paths (Iceberg CTAS) so planner
    /// statistics stay warm without an explicit `ANALYZE TABLE` run.
    #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
    fn record_table_row_count_stat(&self, table_ref: &str, row_count: u64) {
        let registry = krishiv_plan::optimizer::global_table_stats();
        let mut names = vec![table_ref];
        let bare = table_ref.rsplit('.').next().unwrap_or(table_ref);
        if bare != table_ref {
            names.push(bare);
        }
        for name in &names {
            let mut stats = registry
                .get(name)
                .unwrap_or_else(|| krishiv_plan::optimizer::TableCboStats::new(*name));
            stats.row_count = Some(row_count);
            registry.put(stats);
        }
        if let Ok(mut counts) = self.table_row_counts.write() {
            for name in &names {
                counts.insert((*name).to_owned(), row_count);
            }
        }
    }

    /// Phase 54 auto-stats: apply a signed row-count delta (Iceberg DELETE)
    /// to any existing statistic for `table_ref`. Tables never analyzed or
    /// written through this engine are left alone — a delta without a base
    /// count would fabricate data.
    #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
    fn adjust_table_row_count_stat(&self, table_ref: &str, delta: i64) {
        let registry = krishiv_plan::optimizer::global_table_stats();
        let mut names = vec![table_ref];
        let bare = table_ref.rsplit('.').next().unwrap_or(table_ref);
        if bare != table_ref {
            names.push(bare);
        }
        for name in &names {
            if let Some(mut stats) = registry.get(name)
                && let Some(current) = stats.row_count
            {
                stats.row_count = Some(current.saturating_add_signed(delta));
                registry.put(stats);
            }
        }
        if let Ok(mut counts) = self.table_row_counts.write() {
            for name in &names {
                if let Some(current) = counts.get(*name).copied() {
                    counts.insert((*name).to_owned(), current.saturating_add_signed(delta));
                }
            }
        }
    }

    /// Execute `ANALYZE TABLE <ref> [COMPUTE STATISTICS] [FOR COLUMNS (c1, …)]`
    /// (Phase 54).
    ///
    /// Runs one aggregation scan over the table: `COUNT(*)` always, plus
    /// `approx_distinct` / `min` / `max` / non-null count per requested
    /// column. Results land in the engine's `table_row_counts` registry
    /// (the `BroadcastAutoRule` feed) and the process-global
    /// [`krishiv_plan::optimizer::TableStatsRegistry`]; the returned batch
    /// summarizes what was collected. `avg_row_bytes` is taken from the
    /// provider's own statistics when it reports byte sizes (Parquet does).
    async fn dispatch_analyze_table(&self, stmt: &str) -> SqlResult<RecordBatch> {
        use arrow::array::{ArrayRef, Int64Array, StringArray};
        use arrow::datatypes::{DataType, Field, Schema};

        let rest = stmt
            .get(14..)
            .unwrap_or("")
            .trim()
            .trim_end_matches(';')
            .trim();
        let (table_ref, tail) = match rest.split_once(char::is_whitespace) {
            Some((t, tail)) => (t.trim(), tail.trim()),
            None => (rest, ""),
        };
        if table_ref.is_empty() {
            return Err(SqlError::DataFusion {
                message: String::from("ANALYZE TABLE: table reference is required"),
            });
        }
        // Optional noise word (Spark compatibility), then optional column list.
        let mut tail = tail;
        if tail
            .get(..18)
            .is_some_and(|p| p.eq_ignore_ascii_case("COMPUTE STATISTICS"))
        {
            tail = tail.get(18..).unwrap_or("").trim();
        }
        let columns: Vec<String> = if tail
            .get(..11)
            .is_some_and(|p| p.eq_ignore_ascii_case("FOR COLUMNS"))
        {
            tail.get(11..)
                .unwrap_or("")
                .trim()
                .trim_start_matches('(')
                .trim_end_matches(')')
                .split(',')
                .map(|c| c.trim().trim_matches('"').to_owned())
                .filter(|c| !c.is_empty())
                .collect()
        } else if tail.is_empty() {
            Vec::new()
        } else {
            return Err(SqlError::DataFusion {
                message: format!("ANALYZE TABLE: unexpected trailing clause: {tail}"),
            });
        };

        // One scan: COUNT(*) plus four aggregates per analyzed column.
        let mut projections = vec![String::from("count(*)")];
        for c in &columns {
            projections.push(format!("approx_distinct(\"{c}\")"));
            projections.push(format!("min(\"{c}\")"));
            projections.push(format!("max(\"{c}\")"));
            projections.push(format!("count(\"{c}\")"));
        }
        let scan_sql = format!("SELECT {} FROM {table_ref}", projections.join(", "));
        let batches = self.context.sql(&scan_sql).await?.collect().await?;
        let row =
            batches
                .iter()
                .find(|b| b.num_rows() > 0)
                .ok_or_else(|| SqlError::DataFusion {
                    message: format!("ANALYZE TABLE {table_ref}: aggregation returned no rows"),
                })?;
        let cell_string = |col: usize| -> Option<String> {
            let column = row.columns().get(col)?;
            if column.is_null(0) {
                return None;
            }
            arrow::util::display::array_value_to_string(column, 0).ok()
        };
        let cell_u64 = |col: usize| -> Option<u64> { cell_string(col)?.parse().ok() };
        let row_count = cell_u64(0).ok_or_else(|| SqlError::DataFusion {
            message: format!("ANALYZE TABLE {table_ref}: COUNT(*) unreadable"),
        })?;

        let mut column_stats = Vec::with_capacity(columns.len());
        for (i, name) in columns.iter().enumerate() {
            let base = 1 + i * 4;
            let non_null = cell_u64(base + 3);
            column_stats.push(krishiv_plan::optimizer::ColumnCboStats {
                name: name.clone(),
                ndv: cell_u64(base),
                min: cell_string(base + 1),
                max: cell_string(base + 2),
                null_count: non_null.map(|n| row_count.saturating_sub(n)),
            });
        }

        // Provider-reported byte sizes give avg_row_bytes when available.
        let avg_row_bytes = match self.context.table_provider(table_ref).await {
            Ok(provider) => provider.statistics().and_then(|s| {
                let rows = s.num_rows.get_value().copied()?;
                let bytes = s.total_byte_size.get_value().copied()?;
                (rows > 0).then(|| (bytes / rows) as u64)
            }),
            Err(_) => None,
        };

        let mut stats =
            krishiv_plan::optimizer::TableCboStats::new(table_ref).with_row_count(row_count);
        if let Some(bytes) = avg_row_bytes {
            stats = stats.with_avg_row_bytes(bytes);
        }
        if let Some(max_ndv) = column_stats.iter().filter_map(|c| c.ndv).max() {
            // Table-level NDV proxy: the widest column NDV (join-key upper bound).
            stats = stats.with_ndv(max_ndv);
        }
        stats.columns = column_stats;
        let registry = krishiv_plan::optimizer::global_table_stats();
        // Register under the full reference AND the bare table name — scan
        // nodes may carry either depending on how the table was registered.
        let bare = table_ref.rsplit('.').next().unwrap_or(table_ref);
        if bare != table_ref {
            let mut bare_stats = stats.clone();
            bare_stats.table = bare.to_owned();
            registry.put(bare_stats);
        }
        let analyzed_columns = stats.columns.len();
        registry.put(stats);
        if let Ok(mut counts) = self.table_row_counts.write() {
            counts.insert(table_ref.to_owned(), row_count);
            if bare != table_ref {
                counts.insert(bare.to_owned(), row_count);
            }
        }
        self.invalidate_plan_cache();

        let schema = Arc::new(Schema::new(vec![
            Field::new("table_name", DataType::Utf8, false),
            Field::new("row_count", DataType::Int64, false),
            Field::new("avg_row_bytes", DataType::Int64, true),
            Field::new("columns_analyzed", DataType::Int64, false),
        ]));
        let columns_out: Vec<ArrayRef> = vec![
            Arc::new(StringArray::from(vec![table_ref.to_owned()])),
            Arc::new(Int64Array::from(vec![row_count as i64])),
            Arc::new(Int64Array::from(vec![avg_row_bytes.map(|b| b as i64)])),
            Arc::new(Int64Array::from(vec![analyzed_columns as i64])),
        ];
        RecordBatch::try_new(schema, columns_out).map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })
    }

    /// Dispatch a `CALL system.<proc>(...)` statement to the appropriate
    /// Iceberg maintenance function on the first registered KrishivCatalog.
    #[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
    async fn dispatch_call_system(&self, stmt: &str) -> SqlResult<RecordBatch> {
        use arrow::array::{ArrayRef, Int64Array};
        use arrow::datatypes::{DataType, Field, Schema};

        let upper = stmt.to_ascii_uppercase();
        const PREFIX: &str = "CALL SYSTEM.";
        let upper_after = &upper[PREFIX.len()..];
        let orig_after = &stmt[PREFIX.len()..];

        let paren = upper_after.find('(').ok_or_else(|| SqlError::DataFusion {
            message: format!("CALL: missing '(' in: {stmt}"),
        })?;
        let proc_name = upper_after[..paren].trim();

        let args_raw = orig_after[paren + 1..]
            .trim_end_matches(';')
            .trim()
            .trim_end_matches(')')
            .trim();
        let args = call_args_from_str(args_raw);

        let iceberg_catalog = {
            let guard = self
                .iceberg_catalogs
                .read()
                .unwrap_or_else(|e| e.into_inner());
            guard
                .first()
                .ok_or_else(|| SqlError::DataFusion {
                    message: "CALL system: no Iceberg catalog registered".to_string(),
                })?
                .0
                .as_iceberg()
        };

        let table_ref = args.first().ok_or_else(|| SqlError::DataFusion {
            message: format!("CALL {proc_name}: table reference argument is required"),
        })?;
        let table_ident = iceberg_table_ident(table_ref)?;

        // maintain_table returns a three-column report, unlike the single
        // counters below: CALL system.maintain_table('ns.tbl'[, '7 days'
        // [, target_file_bytes [, retain_last]]]).
        if proc_name == "MAINTAIN_TABLE" {
            let older_than = parse_call_duration(args.get(1).map_or("7 days", |s| s.as_str()))?;
            let target_bytes = args
                .get(2)
                .and_then(|s| s.parse::<u64>().ok())
                .unwrap_or(128 * 1024 * 1024);
            let retain_last = args
                .get(3)
                .and_then(|s| s.parse::<usize>().ok())
                .unwrap_or(1);
            let report = krishiv_connectors::lakehouse::maintenance::maintain_table(
                iceberg_catalog,
                &table_ident,
                target_bytes,
                older_than,
                retain_last,
            )
            .await
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;
            let schema = Arc::new(Schema::new(vec![
                Field::new("compacted_files", DataType::Int64, false),
                Field::new("expired_snapshots", DataType::Int64, false),
                Field::new("removed_orphans", DataType::Int64, false),
            ]));
            let columns: Vec<ArrayRef> = vec![
                Arc::new(Int64Array::from(vec![report.compacted_files as i64])),
                Arc::new(Int64Array::from(vec![report.expired_snapshots as i64])),
                Arc::new(Int64Array::from(vec![report.removed_orphans as i64])),
            ];
            return RecordBatch::try_new(schema, columns).map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            });
        }

        let count: i64 = match proc_name {
            "EXPIRE_SNAPSHOTS" => {
                let dur_s = args.get(1).ok_or_else(|| SqlError::DataFusion {
                    message: "CALL expire_snapshots: duration argument is required".to_string(),
                })?;
                let older_than = parse_call_duration(dur_s)?;
                let retain_last = args
                    .get(2)
                    .and_then(|s| s.parse::<usize>().ok())
                    .unwrap_or(1);
                krishiv_connectors::lakehouse::maintenance::expire_snapshots(
                    iceberg_catalog,
                    &table_ident,
                    older_than,
                    retain_last,
                )
                .await
                .map_err(|e| SqlError::DataFusion {
                    message: e.to_string(),
                })? as i64
            }
            "REMOVE_ORPHAN_FILES" => {
                let dur_s = args.get(1).ok_or_else(|| SqlError::DataFusion {
                    message: "CALL remove_orphan_files: duration argument is required".to_string(),
                })?;
                let older_than = parse_call_duration(dur_s)?;
                krishiv_connectors::lakehouse::maintenance::remove_orphan_files(
                    iceberg_catalog,
                    &table_ident,
                    older_than,
                )
                .await
                .map_err(|e| SqlError::DataFusion {
                    message: e.to_string(),
                })? as i64
            }
            "COMPACT_DATA_FILES" => {
                let target_bytes = args
                    .get(1)
                    .and_then(|s| s.parse::<u64>().ok())
                    .unwrap_or(128 * 1024 * 1024);
                krishiv_connectors::lakehouse::maintenance::compact_data_files(
                    iceberg_catalog,
                    &table_ident,
                    target_bytes,
                )
                .await
                .map_err(|e| SqlError::DataFusion {
                    message: e.to_string(),
                })? as i64
            }
            other => {
                return Err(SqlError::Unsupported {
                    feature: format!("CALL system.{other}: unknown procedure"),
                });
            }
        };

        let col = match proc_name {
            "EXPIRE_SNAPSHOTS" => "expired_snapshots",
            "REMOVE_ORPHAN_FILES" => "removed_files",
            "COMPACT_DATA_FILES" => "rewritten_files",
            _ => "result",
        };
        let schema = Arc::new(Schema::new(vec![Field::new(col, DataType::Int64, false)]));
        let array: ArrayRef = Arc::new(Int64Array::from(vec![count]));
        RecordBatch::try_new(schema, vec![array]).map_err(|e| SqlError::DataFusion {
            message: e.to_string(),
        })
    }
}

/// A query result annotated with the operation ID that produced it.
pub struct TaggedQueryResult {
    /// The caller-supplied operation ID.
    pub operation_id: u64,
    /// The underlying SQL DataFrame.
    pub inner: SqlDataFrame,
}

/// Registry of cancelled operation IDs and optional progress snapshots.
///
/// Callers can cancel an in-flight operation by registering its ID here before
/// or during execution.  [`SqlEngine::execute_with_operation_id`] checks this
/// registry at the start of execution.
#[derive(Clone, Default)]
pub struct OperationRegistry {
    cancelled: Arc<std::sync::RwLock<std::collections::HashSet<u64>>>,
    progress: Arc<std::sync::RwLock<std::collections::HashMap<u64, (u64, u64)>>>,
}

impl OperationRegistry {
    /// Create a new, empty operation registry.
    pub fn new() -> Self {
        Self::default()
    }

    /// Cancel an operation by ID.  Subsequent
    /// [`execute_with_operation_id`][SqlEngine::execute_with_operation_id] calls
    /// with this ID will return [`SqlError::OperationCancelled`].
    pub fn cancel(&self, operation_id: u64) {
        if let Ok(mut ids) = self.cancelled.write() {
            ids.insert(operation_id);
        }
    }

    /// Return `true` if `operation_id` has been cancelled.
    pub fn is_cancelled(&self, operation_id: u64) -> bool {
        self.cancelled
            .read()
            .map(|ids| ids.contains(&operation_id))
            .unwrap_or(false)
    }

    /// Remove a cancelled ID (e.g. once the operation has been cleaned up).
    pub fn remove(&self, operation_id: u64) {
        if let Ok(mut ids) = self.cancelled.write() {
            ids.remove(&operation_id);
        }
        if let Ok(mut progress) = self.progress.write() {
            progress.remove(&operation_id);
        }
    }

    /// Record row-level progress for an operation.
    pub fn update_progress(&self, operation_id: u64, rows_scanned: u64, rows_emitted: u64) {
        if let Ok(mut progress) = self.progress.write() {
            progress.insert(operation_id, (rows_scanned, rows_emitted));
        }
    }

    /// Return the latest `(rows_scanned, rows_emitted)` snapshot, if any.
    pub fn progress(&self, operation_id: u64) -> Option<(u64, u64)> {
        self.progress
            .read()
            .ok()
            .and_then(|progress| progress.get(&operation_id).copied())
    }

    /// Return all currently cancelled operation IDs.
    pub fn cancelled_ids(&self) -> Vec<u64> {
        self.cancelled
            .read()
            .map(|ids| ids.iter().copied().collect())
            .unwrap_or_default()
    }
}

/// Extract the table name from a `CREATE EXTERNAL TABLE <name> ...` DDL statement.
///
/// Returns `None` for any other SQL statement. Used to populate `table_row_counts`
/// after DDL so that `BroadcastAutoRule` can fire for connector-backed tables.
pub(crate) fn extract_create_external_table_name(query: &str) -> Option<String> {
    use datafusion::sql::parser::{DFParser, Statement as DFStatement};
    let mut stmts = DFParser::parse_sql(query).ok()?;
    match stmts.pop_front()? {
        DFStatement::CreateExternalTable(create) => Some(create.name.to_string()),
        _ => None,
    }
}

/// Extract the `LOCATION` URI of a `CREATE EXTERNAL TABLE … LOCATION '<uri>'`
/// statement, or `None` for any other SQL.
///
/// Used to register the backing S3 object store before the DDL executes, so an
/// `s3://`/`s3a://` location can be schema-inferred and scanned. Mirrors
/// [`extract_create_external_table_name`] (same single-parse, first-statement
/// contract).
pub(crate) fn extract_create_external_table_location(query: &str) -> Option<String> {
    use datafusion::sql::parser::{DFParser, Statement as DFStatement};
    let mut stmts = DFParser::parse_sql(query).ok()?;
    match stmts.pop_front()? {
        DFStatement::CreateExternalTable(create) => Some(create.location),
        _ => None,
    }
}

/// Engine-agnostic interface over a prepared query result.
///
/// Hides the concrete [`SqlDataFrame`] (which holds a DataFusion `DataFrame`)
/// behind a stable trait so that `krishiv-api` and other callers are not
/// forced to depend on DataFusion types.  `datafusion` stays an implementation
/// detail inside `krishiv-sql`; a future engine swap only requires a new impl.
/// Engine-neutral grouping-set mode for canonical DataFrame aggregation.
pub enum GroupingMode<'a> {
    Sets(Vec<Vec<&'a krishiv_plan::expression::Expr>>),
    Cube(Vec<&'a krishiv_plan::expression::Expr>),
    Rollup(Vec<&'a krishiv_plan::expression::Expr>),
}

#[async_trait::async_trait]
pub trait KrishivDataFrameOps: Send + Sync {
    /// Execute and collect all result batches.
    async fn collect(&self) -> SqlResult<Vec<RecordBatch>>;
    /// Execute, collect results, and return lightweight runtime statistics.
    async fn collect_with_stats(&self) -> SqlResult<(Vec<RecordBatch>, SqlExecutionStats)>;
    /// Explain the physical and logical plan text (does not execute).
    async fn explain(&self) -> SqlResult<String>;

    /// Execute and report per-operator runtime metrics.
    ///
    /// Defaults to an error rather than silently falling back to a plain
    /// EXPLAIN: a caller asking for measurements must not be handed a plan
    /// that looks like one.
    async fn explain_analyze(&self) -> SqlResult<String> {
        Err(SqlError::DataFusion {
            message: String::from("EXPLAIN ANALYZE is not supported for this dataframe backend"),
        })
    }
    /// Explain the logical plan text without executing.
    fn explain_logical(&self) -> String;
    /// Build a Krishiv [`LogicalPlan`] wrapper for this DataFrame.
    fn krishiv_logical_plan(&self) -> LogicalPlan;
    /// The original SQL query string, if any.
    fn query(&self) -> Option<&str>;
    /// SQL text for the **current** logical plan (reflecting every applied
    /// transform), suitable for defining an incremental view. Implemented via the
    /// DataFusion unparser; the default errors for ops that cannot be unparsed.
    fn to_sql(&self) -> SqlResult<String> {
        Err(SqlError::Unsupported {
            feature: "to_sql (plan unparsing) is not supported for this DataFrame".into(),
        })
    }
    /// Execute and return a record batch stream.
    async fn execute_stream(&self) -> SqlResult<SqlStream>;

    // ── DataFrame transforms (lazy) ─────────────────────────────────────────

    /// Return the Arrow schema of this DataFrame.
    fn schema(&self) -> SchemaRef;

    /// Select columns by name.
    async fn select(&self, columns: &[&str]) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Select arbitrary SQL expressions.
    async fn select_exprs(
        &self,
        expressions: &[&krishiv_plan::expression::Expr],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Unnest (explode) one or more array/list columns, producing one output
    /// row per element. Multiple equal-length columns are zipped element-wise
    /// (DataFusion `DataFrame::unnest_columns`).
    async fn unnest_columns(&self, columns: &[&str]) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Group by expressions and compute aggregate expressions.
    async fn aggregate(
        &self,
        group_exprs: &[&krishiv_plan::expression::Expr],
        aggregate_exprs: &[&krishiv_plan::expression::Expr],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Aggregate using GROUPING SETS, CUBE, or ROLLUP.
    async fn aggregate_grouping(
        &self,
        grouping: GroupingMode<'_>,
        aggregate_exprs: &[&krishiv_plan::expression::Expr],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Pivot known values into aggregate columns.
    async fn pivot(
        &self,
        group_exprs: &[&krishiv_plan::expression::Expr],
        pivot_column: &krishiv_plan::expression::Expr,
        aggregate_expr: &krishiv_plan::expression::Expr,
        values: &[(krishiv_plan::expression::ScalarValue, String)],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Unpivot columns into name/value rows while preserving other columns.
    async fn unpivot(
        &self,
        columns: &[&str],
        name_column: &str,
        value_column: &str,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Filter rows by a SQL predicate expression.
    async fn filter(&self, predicate: &str) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Filter rows using the engine-owned typed expression AST.
    async fn filter_expr(
        &self,
        predicate: &krishiv_plan::expression::Expr,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Limit the number of rows.
    async fn limit(&self, n: usize) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Remove duplicate rows.
    async fn distinct(&self) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Drop rows with nulls in selected columns; an empty list checks all columns.
    async fn drop_nulls(&self, columns: &[&str]) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Bernoulli-sample rows.
    async fn sample(&self, fraction: f64) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Sort by columns with optional descending flags.
    async fn sort(
        &self,
        columns: &[&str],
        descending: &[bool],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Assign an alias (table name) to this DataFrame.
    async fn alias(&self, alias: &str) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Drop columns by name.
    async fn drop_columns(&self, columns: &[&str]) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Rename a column from `old` to `new`.
    async fn rename_column(&self, old: &str, new: &str) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Add or replace a column with a computed expression.
    async fn with_column(&self, name: &str, expr: &str) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Return the underlying concrete type for downcasting.
    fn as_any(&self) -> &dyn std::any::Any;

    /// Compute summary statistics (delegates to DataFusion's `describe`).
    async fn describe(&self) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Fill null values in `column` with the literal SQL `value`.
    async fn fill_null(&self, column: &str, value: &str)
    -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Join with another DataFrame using a join type and equi-join keys.
    async fn join(
        &self,
        right: &dyn KrishivDataFrameOps,
        how: &str,
        left_on: &[&str],
        right_on: &[&str],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Union this DataFrame with another (UNION ALL semantics).
    async fn union(
        &self,
        right: &dyn KrishivDataFrameOps,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    async fn union_distinct(
        &self,
        right: &dyn KrishivDataFrameOps,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    async fn intersect(
        &self,
        right: &dyn KrishivDataFrameOps,
        distinct: bool,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    async fn except(
        &self,
        right: &dyn KrishivDataFrameOps,
        distinct: bool,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>>;

    /// Register a list of record batches as a named in-memory table in the
    /// same session context that backs this DataFrame.  Used by `cache()`.
    async fn register_batches(&self, name: &str, batches: Vec<RecordBatch>) -> SqlResult<()>;

    /// Deregister a named table from the session context.  Used by `unpersist()`.
    async fn deregister_table(&self, name: &str) -> SqlResult<()>;

    /// Create (or replace) a SQL view named `name` backed by this DataFrame's
    /// query.  Used by `create_or_replace_temp_view()`.
    async fn create_view(&self, name: &str, replace: bool) -> SqlResult<()>;
}

/// Recursively walk a DataFusion `LogicalPlan` and produce Krishiv `PlanNode`
/// entries.  Returns `(nodes, root_id)` where `root_id` is the ID of the
/// top-level Krishiv node representing `plan`.
///
/// Table-scan nodes carry `estimated_rows` when the table name is found in
/// `table_row_counts`.  Unhandled node types fall back to a single opaque
/// `NodeOp::Other` node.
fn df_plan_to_krishiv_nodes(
    plan: &datafusion::logical_expr::LogicalPlan,
    table_row_counts: &std::collections::HashMap<String, u64>,
    counter: &mut usize,
) -> (Vec<krishiv_plan::PlanNode>, String) {
    use datafusion::logical_expr::LogicalPlan as DfPlan;
    use krishiv_plan::{ExecutionKind, NodeOp, PlanNode};

    *counter += 1;
    let idx = *counter;

    match plan {
        DfPlan::TableScan(ts) => {
            let table_name = ts.table_name.table().to_string();
            let row_count = table_row_counts.get(&table_name).copied();
            let filters: Vec<String> = ts.filters.iter().map(|e| e.to_string()).collect();
            let id = format!("scan-{idx}");
            let node = PlanNode::new(&id, format!("Scan {table_name}"), ExecutionKind::Batch)
                .with_op(NodeOp::Scan {
                    table: table_name,
                    filters,
                })
                .with_estimated_rows(row_count);
            (vec![node], id)
        }

        DfPlan::Projection(proj) => {
            let (mut nodes, input_id) =
                df_plan_to_krishiv_nodes(&proj.input, table_row_counts, counter);
            let id = format!("proj-{idx}");
            let columns: Vec<String> = proj.expr.iter().map(|e| e.to_string()).collect();
            nodes.push(
                PlanNode::new(&id, "Projection", ExecutionKind::Batch)
                    .with_op(NodeOp::Project { columns })
                    .with_inputs([input_id]),
            );
            (nodes, id)
        }

        DfPlan::Filter(filter) => {
            let (mut nodes, input_id) =
                df_plan_to_krishiv_nodes(&filter.input, table_row_counts, counter);
            let id = format!("filter-{idx}");
            let predicate = filter.predicate.to_string();
            nodes.push(
                PlanNode::new(&id, "Filter", ExecutionKind::Batch)
                    .with_op(NodeOp::Filter { predicate })
                    .with_inputs([input_id]),
            );
            (nodes, id)
        }

        DfPlan::Aggregate(agg) => {
            let (mut nodes, input_id) =
                df_plan_to_krishiv_nodes(&agg.input, table_row_counts, counter);
            let id = format!("agg-{idx}");
            let group_keys: Vec<String> = agg.group_expr.iter().map(|e| e.to_string()).collect();
            nodes.push(
                PlanNode::new(&id, "Aggregate", ExecutionKind::Batch)
                    .with_op(NodeOp::Aggregate { group_keys })
                    .with_inputs([input_id]),
            );
            (nodes, id)
        }

        DfPlan::Join(join) => {
            let (mut nodes, left_id) =
                df_plan_to_krishiv_nodes(&join.left, table_row_counts, counter);
            let (right_nodes, right_id) =
                df_plan_to_krishiv_nodes(&join.right, table_row_counts, counter);
            nodes.extend(right_nodes);
            let id = format!("join-{idx}");
            // T2: map every DataFusion join variant to its first-class plan
            // counterpart instead of silently downgrading unknowns to `Inner`.
            // `LeftSemi`/`RightSemi`/`LeftAnti`/`RightAnti` are the variants
            // that were previously collapsed.
            let krishiv_join_type = match join.join_type {
                datafusion::common::JoinType::Inner => krishiv_plan::JoinType::Inner,
                datafusion::common::JoinType::Left => krishiv_plan::JoinType::Left,
                datafusion::common::JoinType::Right => krishiv_plan::JoinType::Right,
                datafusion::common::JoinType::Full => krishiv_plan::JoinType::Full,
                datafusion::common::JoinType::LeftSemi => krishiv_plan::JoinType::LeftSemi,
                datafusion::common::JoinType::RightSemi => krishiv_plan::JoinType::RightSemi,
                datafusion::common::JoinType::LeftAnti => krishiv_plan::JoinType::LeftAnti,
                datafusion::common::JoinType::RightAnti => krishiv_plan::JoinType::RightAnti,
                // DataFusion also exposes `LeftMark`/`RightMark` for some
                // subquery-rewritten plans; treat them as Semi for now to
                // preserve the prior behaviour. Future work can split them.
                datafusion::common::JoinType::LeftMark => krishiv_plan::JoinType::LeftSemi,
                datafusion::common::JoinType::RightMark => krishiv_plan::JoinType::RightSemi,
            };
            nodes.push(
                PlanNode::new(&id, "Join", ExecutionKind::Batch)
                    .with_op(NodeOp::Join {
                        join_type: krishiv_join_type,
                    })
                    .with_inputs([left_id, right_id]),
            );
            (nodes, id)
        }

        DfPlan::Sort(sort) => {
            let (mut nodes, input_id) =
                df_plan_to_krishiv_nodes(&sort.input, table_row_counts, counter);
            let id = format!("sort-{idx}");
            nodes.push(
                PlanNode::new(&id, "Sort", ExecutionKind::Batch)
                    .with_op(NodeOp::Other {
                        description: format!(
                            "Sort({})",
                            sort.expr
                                .iter()
                                .map(|e| e.to_string())
                                .collect::<Vec<_>>()
                                .join(", ")
                        ),
                    })
                    .with_inputs([input_id]),
            );
            (nodes, id)
        }

        DfPlan::Repartition(repart) => {
            let (mut nodes, input_id) =
                df_plan_to_krishiv_nodes(&repart.input, table_row_counts, counter);
            let id = format!("exchange-{idx}");
            let partitioning = krishiv_plan::Partitioning::Unpartitioned;
            nodes.push(
                PlanNode::new(&id, "Exchange", ExecutionKind::Batch)
                    .with_op(NodeOp::Exchange { partitioning })
                    .with_inputs([input_id]),
            );
            (nodes, id)
        }

        DfPlan::Limit(limit) => {
            let (mut nodes, input_id) =
                df_plan_to_krishiv_nodes(&limit.input, table_row_counts, counter);
            let id = format!("limit-{idx}");
            nodes.push(
                PlanNode::new(&id, "Limit", ExecutionKind::Batch)
                    .with_op(NodeOp::Other {
                        description: format!(
                            "Limit(skip={:?}, fetch={:?})",
                            limit.skip.as_ref().map(|e| e.to_string()),
                            limit.fetch.as_ref().map(|e| e.to_string()),
                        ),
                    })
                    .with_inputs([input_id]),
            );
            (nodes, id)
        }

        DfPlan::Union(union) if union.inputs.len() == 1 => {
            if let Some(input) = union.inputs.first() {
                df_plan_to_krishiv_nodes(input, table_row_counts, counter)
            } else {
                (Vec::new(), String::new())
            }
        }
        DfPlan::Union(union) => {
            let mut all_nodes = Vec::new();
            let mut input_ids = Vec::new();
            for input in &union.inputs {
                let (sub_nodes, sub_id) =
                    df_plan_to_krishiv_nodes(input, table_row_counts, counter);
                all_nodes.extend(sub_nodes);
                input_ids.push(sub_id);
            }
            let id = format!("union-{idx}");
            all_nodes.push(
                PlanNode::new(&id, "Union", ExecutionKind::Batch)
                    .with_op(NodeOp::Other {
                        description: "Union".to_string(),
                    })
                    .with_inputs(input_ids),
            );
            (all_nodes, id)
        }

        DfPlan::SubqueryAlias(alias) => {
            // SubqueryAlias is transparent; peel it and continue.
            df_plan_to_krishiv_nodes(&alias.input, table_row_counts, counter)
        }

        DfPlan::Values(_) => {
            let id = format!("values-{idx}");
            let node = PlanNode::new(&id, "Values", ExecutionKind::Batch).with_op(NodeOp::Other {
                description: "Values".to_string(),
            });
            (vec![node], id)
        }

        DfPlan::Extension(_) => {
            let id = format!("ext-{idx}");
            let label = plan.to_string();
            let node = PlanNode::new(&id, label.clone(), ExecutionKind::Batch)
                .with_op(NodeOp::Other { description: label });
            (vec![node], id)
        }

        DfPlan::EmptyRelation(_) => {
            let id = format!("empty-{idx}");
            let node =
                PlanNode::new(&id, "EmptyRelation", ExecutionKind::Batch).with_op(NodeOp::Other {
                    description: "EmptyRelation".to_string(),
                });
            (vec![node], id)
        }

        // Fallback: wrap the entire subplan as an opaque node.
        _ => {
            let id = format!("df-{idx}");
            let label = plan.to_string();
            let node = PlanNode::new(&id, label.clone(), ExecutionKind::Batch)
                .with_op(NodeOp::Other { description: label });
            (vec![node], id)
        }
    }
}

/// Krishiv-owned wrapper around a DataFusion DataFrame.
#[derive(Clone)]
pub struct SqlDataFrame {
    name: String,
    query: Option<String>,
    /// Alias for `query` used by `create_view` — same value.
    query_text: Option<String>,
    execution_kind: ExecutionKind,
    dataframe: DataFusionDataFrame,
    shuffle_partitions: Option<u32>,
    /// Shared session context for table registration (cache/view operations).
    context: SessionContext,
    /// Estimated row counts for registered tables, keyed by table name.
    /// Used by `krishiv_logical_plan` to annotate scan nodes with
    /// `estimated_rows` so `BroadcastAutoRule` can fire.
    table_row_counts: Arc<std::sync::RwLock<HashMap<String, u64>>>,
}

impl fmt::Debug for SqlDataFrame {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SqlDataFrame")
            .field("name", &self.name)
            .field("query", &self.query)
            .field("shuffle_partitions", &self.shuffle_partitions)
            .finish_non_exhaustive()
    }
}

impl SqlDataFrame {
    fn new(
        name: impl Into<String>,
        dataframe: DataFusionDataFrame,
        table_row_counts: Arc<std::sync::RwLock<HashMap<String, u64>>>,
    ) -> Self {
        Self {
            name: name.into(),
            query: None,
            query_text: None,
            execution_kind: ExecutionKind::Batch,
            dataframe,
            shuffle_partitions: None,
            context: SessionContext::default(),
            table_row_counts,
        }
    }

    /// Attach the session context so cache/view operations share the live session.
    pub(crate) fn with_context(mut self, context: SessionContext) -> Self {
        self.context = context;
        self
    }

    fn with_query(mut self, query: impl Into<String>) -> Self {
        let q = query.into();
        self.query_text = Some(q.clone());
        self.query = Some(q);
        self
    }

    fn with_execution_kind(mut self, kind: ExecutionKind) -> Self {
        self.execution_kind = kind;
        self
    }

    fn with_shuffle_partitions(mut self, n: Option<u32>) -> Self {
        self.shuffle_partitions = n;
        self
    }

    /// Original SQL query when created from [`SqlEngine::sql`].
    pub fn query(&self) -> Option<&str> {
        self.query.as_deref()
    }

    /// The Arrow schema of this DataFrame's output.
    ///
    /// Available immediately after planning — no execution happens. Used by
    /// the Flight SQL server to populate `dataset_schema` on prepared
    /// statements so JDBC clients can route query-vs-update correctly.
    pub fn arrow_schema(&self) -> arrow::datatypes::SchemaRef {
        std::sync::Arc::new(self.dataframe.schema().as_arrow().clone())
    }

    /// Return a new `SqlDataFrame` with the given DataFusion DataFrame,
    /// preserving the rest of this instance's state.  The new name suffix
    /// helps distinguish transform steps in logical-plan descriptions.
    fn with_new_dataframe(&self, df: DataFusionDataFrame, tag: &str) -> Self {
        Self {
            name: format!("{}-{}", self.name, tag),
            query: None,
            query_text: None,
            execution_kind: self.execution_kind,
            dataframe: df,
            shuffle_partitions: self.shuffle_partitions,
            context: self.context.clone(),
            table_row_counts: self.table_row_counts.clone(),
        }
    }

    /// Create a Krishiv logical plan wrapper for this DataFrame.
    ///
    /// Walks the DataFusion logical plan tree, creating Krishiv `PlanNode`
    /// entries for each operator. Table-scan nodes are annotated with
    /// `estimated_rows` from the engine's table-row-count registry, allowing
    /// `BroadcastAutoRule` to identify small tables for broadcast join
    /// promotion. The plan is then run through the logical optimizer before
    /// being returned.
    pub fn krishiv_logical_plan(&self) -> LogicalPlan {
        let df_plan = self.dataframe.logical_plan();
        let counts = self
            .table_row_counts
            .read()
            .unwrap_or_else(|e| e.into_inner());
        let mut counter = 0usize;
        let (nodes, _root_id) = df_plan_to_krishiv_nodes(df_plan, &counts, &mut counter);

        let mut plan = LogicalPlan::new(self.name.clone(), self.execution_kind);
        for node in nodes {
            plan = plan.with_node(node);
        }

        // Run the logical optimizer so BroadcastAutoRule fires on eligible scans.
        // An optimizer failure falls back to the unoptimized (still valid) plan;
        // execution correctness does not depend on optimization, but the failure
        // must be observable rather than silent.
        let optimizer = krishiv_plan::optimizer::default_logical_optimizer();
        let fallback = plan.clone();
        match optimizer.optimize(plan) {
            Ok(result) => result.plan,
            Err(error) => {
                tracing::warn!(
                    plan = %self.name,
                    %error,
                    "logical optimizer failed; using unoptimized plan"
                );
                fallback
            }
        }
    }

    /// Explain the logical plan without executing it.
    pub fn explain_logical(&self) -> String {
        self.dataframe.logical_plan().to_string()
    }

    /// Explain logical and physical plan details through DataFusion.
    pub async fn explain(&self) -> SqlResult<String> {
        let batches = self
            .dataframe
            .clone()
            .explain(false, false)?
            .collect()
            .await?;
        pretty_batches(&batches)
    }

    /// Execute the query and report per-operator runtime metrics.
    ///
    /// `explain` alone shows the plan the optimizer produced, which is enough
    /// to confirm a query is *planned* well and not enough to explain why it
    /// is slow. TPC-H q17 plans correctly here — decorrelated into a grouped
    /// aggregate plus two partitioned hash joins, with projection and
    /// predicate pushdown both active — and still runs ~6x DuckDB. The plan
    /// shows `DynamicFilter [ empty ]` on the lineitem scans because dynamic
    /// filters are populated at run time, so a static EXPLAIN can never say
    /// whether the join's build side actually pruned the probe scan.
    ///
    /// Answering that needs the counters: rows emitted per operator, row
    /// groups pruned, bytes scanned, time per partition. Those only exist
    /// after execution, which is what `analyze` turns on.
    pub async fn explain_analyze(&self) -> SqlResult<String> {
        let batches = self
            .dataframe
            .clone()
            .explain(false, true)?
            .collect()
            .await?;
        pretty_batches(&batches)
    }

    /// Execute and collect this DataFrame.
    ///
    /// Boxed at the definition — see the compile-time note on
    /// [`SqlEngine::sql`].
    pub fn collect(
        &self,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = SqlResult<Vec<RecordBatch>>> + Send + '_>>
    {
        Box::pin(async move { Ok(self.dataframe.clone().collect().await?) })
    }

    /// Execute and return a record batch stream.
    ///
    /// Boxed at the definition (like [`SqlEngine::sql`]) so the DataFusion
    /// planning future inside never leaks as an opaque type to consumer
    /// crates — see the compile-time note on `sql`.
    pub fn execute_stream(
        &self,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = SqlResult<SqlStream>> + Send + '_>>
    {
        Box::pin(self.execute_stream_boxed_body())
    }

    async fn execute_stream_boxed_body(&self) -> SqlResult<SqlStream> {
        Ok(self.execute_stream_with_schema_boxed_body().await?.1)
    }

    /// Execute, returning the stream **and the schema its batches actually
    /// carry**.
    ///
    /// [`SqlStream`] is a bare `Pin<Box<dyn Stream>>`, so wrapping
    /// DataFusion's stream throws away the one thing only it knows: the
    /// physical output schema. `KrishivDataFrameOps::schema` is not a
    /// substitute — that is the *logical* schema, and physical planning
    /// re-types expressions. TPC-H q17's `avg(l_quantity)` is
    /// `Decimal128(15, 2)` logically and `Decimal128(30, 15)` in the batches,
    /// and a caller that labelled those batches with the logical schema
    /// produced shuffle partitions whose rows violated their own schema.
    ///
    /// Callers that need to declare a schema before the first batch arrives —
    /// or when no batch ever arrives — want this, not `schema()`.
    pub fn execute_stream_with_schema(
        &self,
    ) -> futures::future::BoxFuture<'_, SqlResult<(SchemaRef, SqlStream)>> {
        Box::pin(self.execute_stream_with_schema_boxed_body())
    }

    async fn execute_stream_with_schema_boxed_body(&self) -> SqlResult<(SchemaRef, SqlStream)> {
        let df_stream = self.dataframe.clone().execute_stream().await?;
        let schema = df_stream.schema();
        use futures::StreamExt;
        let mapped = df_stream.map(|res| {
            res.map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })
        });
        Ok((schema, Box::pin(mapped)))
    }

    /// Execute and collect this DataFrame, also returning lightweight runtime statistics.
    ///
    /// Collects `output_rows` from DataFusion's execution metrics. `cpu_nanos`
    /// is approximated from `elapsed_compute` when available. `spill_bytes`
    /// and `spill_count` are aggregated across every operator in the physical
    /// plan tree (sorts, hash joins, and aggregations report spills when the
    /// memory pool forces them to disk); other fields default to 0.
    pub fn collect_with_stats(
        &self,
    ) -> futures::future::BoxFuture<'_, SqlResult<(Vec<RecordBatch>, SqlExecutionStats)>> {
        Box::pin(self.collect_with_stats_boxed_body())
    }

    async fn collect_with_stats_boxed_body(
        &self,
    ) -> SqlResult<(Vec<RecordBatch>, SqlExecutionStats)> {
        use datafusion::physical_plan::collect as df_collect;

        let df = self.dataframe.clone();
        let task_ctx = df.task_ctx();
        let physical_plan = df.create_physical_plan().await?;

        let batches = df_collect(physical_plan.clone(), task_ctx.into()).await?;

        let mut output_rows: u64 = batches.iter().map(|b| b.num_rows() as u64).sum();
        let mut cpu_nanos: u64 = 0;

        if let Some(metrics) = physical_plan.metrics() {
            if let Some(v) = metrics.output_rows() {
                output_rows = v as u64;
            }
            if let Some(t) = metrics.elapsed_compute() {
                cpu_nanos = t as u64;
            }
        }

        let (spill_bytes, spill_count) = aggregate_spill_metrics(physical_plan.as_ref());

        Ok((
            batches,
            SqlExecutionStats {
                output_rows,
                cpu_nanos,
                spill_bytes,
                spill_count,
            },
        ))
    }

    /// Execute this DataFrame as a record-batch stream, returning a stats
    /// handle that reads the same runtime metrics as [`collect_with_stats`]
    /// once the stream has been fully drained.
    ///
    /// Unlike `collect_with_stats`, the caller never holds more than one
    /// batch in memory — the intended path for large results that are
    /// spooled to disk or written straight into a sink.
    ///
    /// [`collect_with_stats`]: Self::collect_with_stats
    pub fn execute_stream_with_stats(
        &self,
    ) -> futures::future::BoxFuture<'_, SqlResult<(SqlStream, SqlStatsHandle)>> {
        Box::pin(self.execute_stream_with_stats_boxed_body())
    }

    async fn execute_stream_with_stats_boxed_body(&self) -> SqlResult<(SqlStream, SqlStatsHandle)> {
        use futures::StreamExt;

        let df = self.dataframe.clone();
        let task_ctx = df.task_ctx();
        let physical_plan = df.create_physical_plan().await?;
        let df_stream = datafusion::physical_plan::execute_stream(
            physical_plan.clone(),
            std::sync::Arc::new(task_ctx),
        )?;
        let mapped = df_stream.map(|res| {
            res.map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })
        });
        Ok((
            Box::pin(mapped),
            SqlStatsHandle {
                plan: physical_plan,
            },
        ))
    }
}

/// Handle onto a streamed execution's physical plan; reads runtime metrics
/// (output rows, CPU time, spill totals) after the stream is drained.
pub struct SqlStatsHandle {
    plan: std::sync::Arc<dyn datafusion::physical_plan::ExecutionPlan>,
}

impl SqlStatsHandle {
    /// Aggregate execution statistics from the plan's runtime metrics.
    ///
    /// Only meaningful once the associated stream has been fully consumed;
    /// calling earlier reports the metrics accumulated so far.
    pub fn stats(&self) -> SqlExecutionStats {
        let mut output_rows: u64 = 0;
        let mut cpu_nanos: u64 = 0;
        if let Some(metrics) = self.plan.metrics() {
            if let Some(v) = metrics.output_rows() {
                output_rows = v as u64;
            }
            if let Some(t) = metrics.elapsed_compute() {
                cpu_nanos = t as u64;
            }
        }
        let (spill_bytes, spill_count) = aggregate_spill_metrics(self.plan.as_ref());
        SqlExecutionStats {
            output_rows,
            cpu_nanos,
            spill_bytes,
            spill_count,
        }
    }
}

/// Recursively sum `spilled_bytes` and `spill_count` metrics across every
/// operator in a physical plan tree.
///
/// The root node's `metrics()` only reflects the root operator; spilling
/// happens in interior sort/join/aggregate nodes, so the whole tree must be
/// walked to account for all disk spill activity.
fn aggregate_spill_metrics(plan: &dyn datafusion::physical_plan::ExecutionPlan) -> (u64, u64) {
    let mut spill_bytes: u64 = 0;
    let mut spill_count: u64 = 0;
    if let Some(metrics) = plan.metrics() {
        if let Some(bytes) = metrics.spilled_bytes() {
            spill_bytes = spill_bytes.saturating_add(bytes as u64);
        }
        if let Some(count) = metrics.spill_count() {
            spill_count = spill_count.saturating_add(count as u64);
        }
    }
    for child in plan.children() {
        let (child_bytes, child_count) = aggregate_spill_metrics(child.as_ref());
        spill_bytes = spill_bytes.saturating_add(child_bytes);
        spill_count = spill_count.saturating_add(child_count);
    }
    (spill_bytes, spill_count)
}

/// Lightweight execution statistics collected from a DataFusion physical plan.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct SqlExecutionStats {
    pub output_rows: u64,
    pub cpu_nanos: u64,
    /// Total bytes spilled to disk across all operators in the plan.
    pub spill_bytes: u64,
    /// Number of spill events (roughly: spill files written) across all operators.
    pub spill_count: u64,
}

fn top_level_alias_index(expression: &str) -> Option<usize> {
    let bytes = expression.as_bytes();
    let mut depth = 0usize;
    let mut single_quoted = false;
    let mut double_quoted = false;
    let mut candidate = None;
    let mut index = 0usize;
    while index < bytes.len() {
        let Some(&byte) = bytes.get(index) else {
            break;
        };
        match byte {
            b'\'' if !double_quoted => {
                if single_quoted && bytes.get(index + 1) == Some(&b'\'') {
                    index += 2;
                    continue;
                }
                single_quoted = !single_quoted;
            }
            b'"' if !single_quoted => {
                if double_quoted && bytes.get(index + 1) == Some(&b'"') {
                    index += 2;
                    continue;
                }
                double_quoted = !double_quoted;
            }
            b'(' if !single_quoted && !double_quoted => depth += 1,
            b')' if !single_quoted && !double_quoted => depth = depth.saturating_sub(1),
            b' ' if depth == 0
                && !single_quoted
                && !double_quoted
                && bytes
                    .get(index..index + 4)
                    .is_some_and(|slice| slice.eq_ignore_ascii_case(b" AS ")) =>
            {
                candidate = Some(index);
                index += 3;
            }
            _ => {}
        }
        index += 1;
    }
    candidate
}

fn parse_dataframe_expression(
    dataframe: &datafusion::dataframe::DataFrame,
    expression: &str,
) -> SqlResult<datafusion::logical_expr::Expr> {
    if let Some(index) = top_level_alias_index(expression) {
        let (body, alias) = expression.split_at(index);
        let alias = alias[4..].trim();
        if !alias.is_empty() {
            let alias = alias
                .strip_prefix('"')
                .and_then(|value| value.strip_suffix('"'))
                .unwrap_or(alias)
                .replace("\"\"", "\"");
            return Ok(dataframe.parse_sql_expr(body.trim())?.alias(alias));
        }
    }
    dataframe.parse_sql_expr(expression).map_err(Into::into)
}

/// Parse the stable SQL-expression subset into the same engine-owned AST used by Rust and Python.
pub fn parse_public_expression(sql: &str) -> SqlResult<krishiv_plan::expression::Expr> {
    let dialect = GenericDialect {};
    let mut parser =
        Parser::new(&dialect)
            .try_with_sql(sql)
            .map_err(|error| SqlError::Unsupported {
                feature: format!("public expression parse: {error}"),
            })?;
    let expression = parser.parse_expr().map_err(|error| SqlError::Unsupported {
        feature: format!("public expression parse: {error}"),
    })?;
    sqlparser_expression_to_public(&expression)
}

fn sqlparser_expression_to_public(
    expression: &datafusion::sql::sqlparser::ast::Expr,
) -> SqlResult<krishiv_plan::expression::Expr> {
    use datafusion::sql::sqlparser::ast::{BinaryOperator as SqlOperator, Expr as SqlExpr, Value};
    use krishiv_plan::expression::{BinaryOperator, Expr, ScalarValue};

    Ok(match expression {
        SqlExpr::Identifier(identifier) => Expr::Column {
            path: vec![identifier.value.clone()],
        },
        SqlExpr::CompoundIdentifier(identifiers) => Expr::Column {
            path: identifiers
                .iter()
                .map(|identifier| identifier.value.clone())
                .collect(),
        },
        SqlExpr::Nested(expression) => sqlparser_expression_to_public(expression)?,
        SqlExpr::IsNull(expression) => Expr::IsNull {
            expression: Box::new(sqlparser_expression_to_public(expression)?),
            negated: false,
        },
        SqlExpr::IsNotNull(expression) => Expr::IsNull {
            expression: Box::new(sqlparser_expression_to_public(expression)?),
            negated: true,
        },
        SqlExpr::BinaryOp { left, op, right } => Expr::Binary {
            left: Box::new(sqlparser_expression_to_public(left)?),
            op: match op {
                SqlOperator::Eq => BinaryOperator::Eq,
                SqlOperator::NotEq => BinaryOperator::NotEq,
                SqlOperator::Gt => BinaryOperator::Gt,
                SqlOperator::GtEq => BinaryOperator::GtEq,
                SqlOperator::Lt => BinaryOperator::Lt,
                SqlOperator::LtEq => BinaryOperator::LtEq,
                SqlOperator::And => BinaryOperator::And,
                SqlOperator::Or => BinaryOperator::Or,
                SqlOperator::Plus => BinaryOperator::Plus,
                SqlOperator::Minus => BinaryOperator::Minus,
                SqlOperator::Multiply => BinaryOperator::Multiply,
                SqlOperator::Divide => BinaryOperator::Divide,
                other => {
                    return Err(SqlError::Unsupported {
                        feature: format!("public expression operator {other}"),
                    });
                }
            },
            right: Box::new(sqlparser_expression_to_public(right)?),
        },
        SqlExpr::Value(value) => Expr::Literal {
            value: match &value.value {
                Value::Null => ScalarValue::Null,
                Value::Boolean(value) => ScalarValue::Boolean(*value),
                Value::SingleQuotedString(value) => ScalarValue::Utf8(value.clone()),
                Value::Number(value, _)
                    if value.contains('.') || value.contains('e') || value.contains('E') =>
                {
                    ScalarValue::float64(value.parse::<f64>().map_err(|error| {
                        SqlError::Unsupported {
                            feature: format!("numeric expression literal: {error}"),
                        }
                    })?)
                }
                Value::Number(value, _) => {
                    ScalarValue::Int64(value.parse::<i64>().map_err(|error| {
                        SqlError::Unsupported {
                            feature: format!("integer expression literal: {error}"),
                        }
                    })?)
                }
                other => {
                    return Err(SqlError::Unsupported {
                        feature: format!("public expression literal {other}"),
                    });
                }
            },
        },
        other => {
            return Err(SqlError::Unsupported {
                feature: format!("public expression node {other}"),
            });
        }
    })
}

fn public_data_type_to_arrow(
    data_type: &krishiv_plan::expression::ExprDataType,
) -> arrow::datatypes::DataType {
    use arrow::datatypes::{DataType, Field, IntervalUnit, TimeUnit};
    use krishiv_plan::expression::{ExprDataType, IntervalUnit as PublicIntervalUnit};

    match data_type {
        ExprDataType::Null => DataType::Null,
        ExprDataType::Boolean => DataType::Boolean,
        ExprDataType::Int64 => DataType::Int64,
        ExprDataType::UInt64 => DataType::UInt64,
        ExprDataType::Float64 => DataType::Float64,
        ExprDataType::Utf8 => DataType::Utf8,
        ExprDataType::Binary => DataType::Binary,
        ExprDataType::Decimal128 { precision, scale } => DataType::Decimal128(*precision, *scale),
        ExprDataType::Date32 => DataType::Date32,
        ExprDataType::Timestamp { unit, timezone } => DataType::Timestamp(
            match unit {
                krishiv_plan::expression::TimeUnit::Second => TimeUnit::Second,
                krishiv_plan::expression::TimeUnit::Millisecond => TimeUnit::Millisecond,
                krishiv_plan::expression::TimeUnit::Microsecond => TimeUnit::Microsecond,
                krishiv_plan::expression::TimeUnit::Nanosecond => TimeUnit::Nanosecond,
            },
            timezone.clone().map(Into::into),
        ),
        ExprDataType::Interval { unit } => DataType::Interval(match unit {
            PublicIntervalUnit::YearMonth => IntervalUnit::YearMonth,
            PublicIntervalUnit::DayTime => IntervalUnit::DayTime,
            PublicIntervalUnit::MonthDayNano => IntervalUnit::MonthDayNano,
        }),
        ExprDataType::List(element) => DataType::List(Arc::new(Field::new(
            "item",
            public_data_type_to_arrow(element),
            true,
        ))),
        ExprDataType::Map { key, value } => DataType::Map(
            Arc::new(Field::new(
                "entries",
                DataType::Struct(
                    vec![
                        Arc::new(Field::new("key", public_data_type_to_arrow(key), false)),
                        Arc::new(Field::new("value", public_data_type_to_arrow(value), true)),
                    ]
                    .into(),
                ),
                false,
            )),
            false,
        ),
        ExprDataType::Struct(fields) => DataType::Struct(
            fields
                .iter()
                .map(|field| {
                    Arc::new(Field::new(
                        &field.name,
                        public_data_type_to_arrow(&field.data_type),
                        field.nullable,
                    ))
                })
                .collect::<Vec<_>>()
                .into(),
        ),
        // Variant: stored as JSON-encoded UTF-8 until Arrow gains a
        // native variant logical type. Read/write paths use Utf8
        // columns and the datafusion engine treats the values as
        // opaque strings.
        ExprDataType::Variant => DataType::Utf8,
    }
}

fn public_scalar_to_datafusion(
    value: &krishiv_plan::expression::ScalarValue,
) -> Option<datafusion::common::ScalarValue> {
    use datafusion::common::ScalarValue;
    use krishiv_plan::expression::{ScalarValue as PublicScalar, TimeUnit};

    Some(match value {
        PublicScalar::Null => ScalarValue::Null,
        PublicScalar::Boolean(value) => ScalarValue::Boolean(Some(*value)),
        PublicScalar::Int64(value) => ScalarValue::Int64(Some(*value)),
        PublicScalar::UInt64(value) => ScalarValue::UInt64(Some(*value)),
        PublicScalar::Float64(bits) => ScalarValue::Float64(Some(f64::from_bits(*bits))),
        PublicScalar::Utf8(value) => ScalarValue::Utf8(Some(value.clone())),
        PublicScalar::Binary(value) => ScalarValue::Binary(Some(value.clone())),
        PublicScalar::Decimal128 {
            value,
            precision,
            scale,
        } => ScalarValue::Decimal128(Some(*value), *precision, *scale),
        PublicScalar::Date32(value) => ScalarValue::Date32(Some(*value)),
        PublicScalar::Timestamp {
            value,
            unit,
            timezone,
        } => {
            let timezone = timezone.clone().map(Into::into);
            match unit {
                TimeUnit::Second => ScalarValue::TimestampSecond(Some(*value), timezone),
                TimeUnit::Millisecond => ScalarValue::TimestampMillisecond(Some(*value), timezone),
                TimeUnit::Microsecond => ScalarValue::TimestampMicrosecond(Some(*value), timezone),
                TimeUnit::Nanosecond => ScalarValue::TimestampNanosecond(Some(*value), timezone),
            }
        }
        PublicScalar::Interval { .. } => return None,
    })
}

/// Lower the versioned engine-owned expression contract into a DataFusion expression.
///
/// Ordinary nodes are lowered structurally. `RawSql`, generic function calls, aggregate
/// calls, and interval literals intentionally use DataFusion's SQL analyzer as the
/// compatibility/preview path until those families receive dedicated typed nodes.
fn lower_public_expression(
    dataframe: &datafusion::dataframe::DataFrame,
    expression: &krishiv_plan::expression::Expr,
) -> SqlResult<datafusion::logical_expr::Expr> {
    expression
        .validate()
        .map_err(|error| SqlError::Unsupported {
            feature: format!("invalid public expression: {error}"),
        })?;
    use datafusion::logical_expr::{Expr as DataFusionExpr, Operator, binary_expr, cast, try_cast};
    use krishiv_plan::expression::{BinaryOperator, Expr};

    Ok(match expression {
        Expr::Column { path } if path.len() == 1 => {
            datafusion::prelude::col(path.first().map(String::as_str).unwrap_or(""))
        }
        Expr::Column { .. } => parse_dataframe_expression(dataframe, &expression.to_sql())?,
        Expr::Literal { value } => match public_scalar_to_datafusion(value) {
            Some(value) => DataFusionExpr::Literal(value, None),
            None => parse_dataframe_expression(dataframe, &expression.to_sql())?,
        },
        Expr::Alias { expression, name } => {
            lower_public_expression(dataframe, expression)?.alias(name)
        }
        Expr::Binary { left, op, right } => binary_expr(
            lower_public_expression(dataframe, left)?,
            match op {
                BinaryOperator::Eq => Operator::Eq,
                BinaryOperator::NotEq => Operator::NotEq,
                BinaryOperator::Gt => Operator::Gt,
                BinaryOperator::GtEq => Operator::GtEq,
                BinaryOperator::Lt => Operator::Lt,
                BinaryOperator::LtEq => Operator::LtEq,
                BinaryOperator::And => Operator::And,
                BinaryOperator::Or => Operator::Or,
                BinaryOperator::Plus => Operator::Plus,
                BinaryOperator::Minus => Operator::Minus,
                BinaryOperator::Multiply => Operator::Multiply,
                BinaryOperator::Divide => Operator::Divide,
            },
            lower_public_expression(dataframe, right)?,
        ),
        Expr::IsNull {
            expression,
            negated,
        } => {
            let expression = lower_public_expression(dataframe, expression)?;
            if *negated {
                expression.is_not_null()
            } else {
                expression.is_null()
            }
        }
        Expr::Cast {
            expression,
            data_type,
            safe,
        } => {
            let expression = lower_public_expression(dataframe, expression)?;
            let data_type = public_data_type_to_arrow(data_type);
            if *safe {
                try_cast(expression, data_type)
            } else {
                cast(expression, data_type)
            }
        }
        Expr::Sort { .. } => {
            return Err(SqlError::Unsupported {
                feature: "standalone sort expressions are only valid inside windows or order_by"
                    .into(),
            });
        }
        Expr::Aggregate { .. }
        | Expr::Function { .. }
        | Expr::Window { .. }
        | Expr::RawSql { .. } => parse_dataframe_expression(dataframe, &expression.to_sql())?,
    })
}

fn sql_dataframe<'a>(
    dataframe: &'a dyn KrishivDataFrameOps,
    operation: &str,
) -> SqlResult<&'a SqlDataFrame> {
    dataframe
        .as_any()
        .downcast_ref::<SqlDataFrame>()
        .ok_or_else(|| SqlError::DataFusion {
            message: format!("right DataFrame must be SqlDataFrame for {operation}"),
        })
}

#[async_trait::async_trait]
impl KrishivDataFrameOps for SqlDataFrame {
    async fn collect(&self) -> SqlResult<Vec<RecordBatch>> {
        SqlDataFrame::collect(self).await
    }
    async fn collect_with_stats(&self) -> SqlResult<(Vec<RecordBatch>, SqlExecutionStats)> {
        SqlDataFrame::collect_with_stats(self).await
    }
    async fn explain_analyze(&self) -> SqlResult<String> {
        SqlDataFrame::explain_analyze(self).await
    }

    async fn explain(&self) -> SqlResult<String> {
        SqlDataFrame::explain(self).await
    }
    fn explain_logical(&self) -> String {
        SqlDataFrame::explain_logical(self)
    }
    fn krishiv_logical_plan(&self) -> LogicalPlan {
        let label = self.dataframe.logical_plan().to_string();
        let mut plan = LogicalPlan::new(self.name.clone(), ExecutionKind::Batch).with_node(
            PlanNode::new("datafusion-logical", label, ExecutionKind::Batch),
        );
        if let Some(n) = self.shuffle_partitions {
            plan = plan.with_shuffle_partitions(Some(n));
        }
        plan
    }
    fn query(&self) -> Option<&str> {
        SqlDataFrame::query(self)
    }
    fn to_sql(&self) -> SqlResult<String> {
        // Unparse the CURRENT DataFusion logical plan (reflects all transforms).
        // Fall back to the original query string if unparsing is unavailable.
        match datafusion::sql::unparser::plan_to_sql(self.dataframe.logical_plan()) {
            Ok(statement) => Ok(statement.to_string()),
            Err(err) => self
                .query()
                .map(str::to_string)
                .ok_or_else(|| SqlError::Unsupported {
                    feature: format!("cannot render DataFrame plan as SQL: {err}"),
                }),
        }
    }
    async fn execute_stream(&self) -> SqlResult<SqlStream> {
        SqlDataFrame::execute_stream(self).await
    }

    // ── DataFrame transforms ────────────────────────────────────────────────

    fn schema(&self) -> SchemaRef {
        SchemaRef::from(self.dataframe.schema().clone())
    }

    async fn select(&self, columns: &[&str]) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let df = self.dataframe.clone().select_columns(columns)?;
        Ok(Box::new(self.with_new_dataframe(df, "select")))
    }

    async fn select_exprs(
        &self,
        expressions: &[&krishiv_plan::expression::Expr],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let expressions = expressions
            .iter()
            .map(|expression| lower_public_expression(&self.dataframe, expression))
            .collect::<Result<Vec<_>, _>>()?;
        let df = self.dataframe.clone().select(expressions)?;
        Ok(Box::new(self.with_new_dataframe(df, "select_exprs")))
    }

    async fn unnest_columns(&self, columns: &[&str]) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let df = self.dataframe.clone().unnest_columns(columns)?;
        Ok(Box::new(self.with_new_dataframe(df, "unnest")))
    }

    async fn aggregate(
        &self,
        group_exprs: &[&krishiv_plan::expression::Expr],
        aggregate_exprs: &[&krishiv_plan::expression::Expr],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        if aggregate_exprs.is_empty() {
            return Err(SqlError::Unsupported {
                feature: "aggregate requires at least one aggregate expression".into(),
            });
        }
        let group_exprs = group_exprs
            .iter()
            .map(|expression| lower_public_expression(&self.dataframe, expression))
            .collect::<Result<Vec<_>, _>>()?;
        let aggregate_exprs = aggregate_exprs
            .iter()
            .map(|expression| lower_public_expression(&self.dataframe, expression))
            .collect::<Result<Vec<_>, _>>()?;
        let df = self
            .dataframe
            .clone()
            .aggregate(group_exprs, aggregate_exprs)?;
        Ok(Box::new(self.with_new_dataframe(df, "aggregate")))
    }

    async fn aggregate_grouping(
        &self,
        grouping: GroupingMode<'_>,
        aggregate_exprs: &[&krishiv_plan::expression::Expr],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        if aggregate_exprs.is_empty() {
            return Err(SqlError::Unsupported {
                feature: "grouping aggregation requires at least one aggregate expression".into(),
            });
        }
        let lower = |expression: &&krishiv_plan::expression::Expr| {
            lower_public_expression(&self.dataframe, expression)
        };
        let group = match grouping {
            GroupingMode::Sets(sets) => datafusion::logical_expr::grouping_set(
                sets.into_iter()
                    .map(|set| set.iter().map(lower).collect::<Result<Vec<_>, _>>())
                    .collect::<Result<Vec<_>, _>>()?,
            ),
            GroupingMode::Cube(expressions) => datafusion::logical_expr::cube(
                expressions
                    .iter()
                    .map(lower)
                    .collect::<Result<Vec<_>, _>>()?,
            ),
            GroupingMode::Rollup(expressions) => datafusion::logical_expr::rollup(
                expressions
                    .iter()
                    .map(lower)
                    .collect::<Result<Vec<_>, _>>()?,
            ),
        };
        let aggregates = aggregate_exprs
            .iter()
            .map(lower)
            .collect::<Result<Vec<_>, _>>()?;
        let df = self.dataframe.clone().aggregate(vec![group], aggregates)?;
        Ok(Box::new(self.with_new_dataframe(df, "aggregate_grouping")))
    }

    async fn pivot(
        &self,
        group_exprs: &[&krishiv_plan::expression::Expr],
        pivot_column: &krishiv_plan::expression::Expr,
        aggregate_expr: &krishiv_plan::expression::Expr,
        values: &[(krishiv_plan::expression::ScalarValue, String)],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        use krishiv_plan::expression::Expr as PublicExpr;
        let (function, input, distinct) = match aggregate_expr {
            PublicExpr::Aggregate {
                function,
                expression: Some(input),
                distinct,
            } => (*function, input.as_ref(), *distinct),
            _ => {
                return Err(SqlError::Unsupported {
                    feature: "pivot requires an aggregate expression with one input".into(),
                });
            }
        };
        if values.is_empty() {
            return Err(SqlError::Unsupported {
                feature: "pivot requires at least one value".into(),
            });
        }
        let group_exprs = group_exprs
            .iter()
            .map(|expression| lower_public_expression(&self.dataframe, expression))
            .collect::<Result<Vec<_>, _>>()?;
        let aggregates = values
            .iter()
            .map(|(value, alias)| {
                let conditional = PublicExpr::raw(format!(
                    "CASE WHEN {} = {} THEN {} END",
                    pivot_column.to_sql(),
                    value.to_sql_literal(),
                    input.to_sql()
                ));
                let aggregate = PublicExpr::Aggregate {
                    function,
                    expression: Some(Box::new(conditional)),
                    distinct,
                }
                .alias(alias);
                lower_public_expression(&self.dataframe, &aggregate)
            })
            .collect::<Result<Vec<_>, _>>()?;
        let dataframe = self.dataframe.clone().aggregate(group_exprs, aggregates)?;
        Ok(Box::new(self.with_new_dataframe(dataframe, "pivot")))
    }

    async fn unpivot(
        &self,
        columns: &[&str],
        name_column: &str,
        value_column: &str,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        if columns.is_empty() {
            return Err(SqlError::Unsupported {
                feature: "unpivot requires at least one column".into(),
            });
        }
        let retained = self
            .dataframe
            .schema()
            .fields()
            .iter()
            .map(|field| field.name().as_str())
            .filter(|name| !columns.contains(name))
            .collect::<Vec<_>>();
        let mut branches = Vec::with_capacity(columns.len());
        for column in columns {
            let mut expressions = retained
                .iter()
                .map(|name| datafusion::logical_expr::col(*name))
                .collect::<Vec<_>>();
            expressions
                .push(datafusion::logical_expr::lit((*column).to_owned()).alias(name_column));
            expressions.push(datafusion::logical_expr::col(*column).alias(value_column));
            branches.push(self.dataframe.clone().select(expressions)?);
        }
        let mut branches = branches.into_iter();
        let Some(mut dataframe) = branches.next() else {
            return Err(SqlError::Unsupported {
                feature: "unpivot requires at least one branch".into(),
            });
        };
        for branch in branches {
            dataframe = dataframe.union(branch)?;
        }
        Ok(Box::new(self.with_new_dataframe(dataframe, "unpivot")))
    }

    async fn filter(&self, predicate: &str) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let expr = self.dataframe.parse_sql_expr(predicate)?;
        let df = self.dataframe.clone().filter(expr)?;
        Ok(Box::new(self.with_new_dataframe(df, "filter")))
    }

    async fn filter_expr(
        &self,
        predicate: &krishiv_plan::expression::Expr,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let expr = lower_public_expression(&self.dataframe, predicate)?;
        let df = self.dataframe.clone().filter(expr)?;
        Ok(Box::new(self.with_new_dataframe(df, "filter_expr")))
    }

    async fn limit(&self, n: usize) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let df = self.dataframe.clone().limit(0, Some(n))?;
        Ok(Box::new(self.with_new_dataframe(df, "limit")))
    }

    async fn distinct(&self) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let df = self.dataframe.clone().distinct()?;
        Ok(Box::new(self.with_new_dataframe(df, "distinct")))
    }

    async fn drop_nulls(&self, columns: &[&str]) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let columns = if columns.is_empty() {
            self.dataframe
                .schema()
                .fields()
                .iter()
                .map(|field| field.name().as_str())
                .collect::<Vec<_>>()
        } else {
            columns.to_vec()
        };
        let mut predicate: Option<datafusion::logical_expr::Expr> = None;
        for column in columns {
            let next = datafusion::logical_expr::col(column).is_not_null();
            predicate = Some(match predicate {
                Some(current) => current.and(next),
                None => next,
            });
        }
        let df = match predicate {
            Some(predicate) => self.dataframe.clone().filter(predicate)?,
            None => self.dataframe.clone(),
        };
        Ok(Box::new(self.with_new_dataframe(df, "drop_nulls")))
    }

    async fn sample(&self, fraction: f64) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        if !(0.0..=1.0).contains(&fraction) {
            return Err(SqlError::Unsupported {
                feature: "sample fraction must be between 0 and 1".into(),
            });
        }
        let predicate = self
            .dataframe
            .parse_sql_expr(&format!("random() < {fraction}"))?;
        let df = self.dataframe.clone().filter(predicate)?;
        Ok(Box::new(self.with_new_dataframe(df, "sample")))
    }

    async fn sort(
        &self,
        columns: &[&str],
        descending: &[bool],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        use datafusion::logical_expr::SortExpr;
        let exprs: Vec<SortExpr> = columns
            .iter()
            .zip(descending.iter())
            .map(|(col_name, desc)| datafusion::logical_expr::col(*col_name).sort(!desc, *desc))
            .collect();
        let df = self.dataframe.clone().sort(exprs)?;
        Ok(Box::new(self.with_new_dataframe(df, "sort")))
    }

    async fn alias(&self, alias: &str) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let df = self.dataframe.clone().alias(alias)?;
        Ok(Box::new(self.with_new_dataframe(df, "alias")))
    }

    async fn drop_columns(&self, columns: &[&str]) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let df = self.dataframe.clone().drop_columns(columns)?;
        Ok(Box::new(self.with_new_dataframe(df, "drop")))
    }

    async fn rename_column(&self, old: &str, new: &str) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let df = self.dataframe.clone().with_column_renamed(old, new)?;
        Ok(Box::new(self.with_new_dataframe(df, "rename")))
    }

    async fn with_column(&self, name: &str, expr: &str) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let parsed = self.dataframe.parse_sql_expr(expr)?;
        let df = self.dataframe.clone().with_column(name, parsed)?;
        Ok(Box::new(self.with_new_dataframe(df, "with_column")))
    }

    fn as_any(&self) -> &dyn std::any::Any {
        self
    }

    async fn describe(&self) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let df = self.dataframe.clone().describe().await?;
        Ok(Box::new(self.with_new_dataframe(df, "describe")))
    }

    async fn fill_null(
        &self,
        column: &str,
        value: &str,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let expr = format!("COALESCE({column}, {value})");
        let parsed = self.dataframe.parse_sql_expr(&expr)?;
        let df = self.dataframe.clone().with_column(column, parsed)?;
        Ok(Box::new(self.with_new_dataframe(df, "fill_null")))
    }

    async fn join(
        &self,
        right: &dyn KrishivDataFrameOps,
        how: &str,
        left_on: &[&str],
        right_on: &[&str],
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let right_sql = right
            .as_any()
            .downcast_ref::<SqlDataFrame>()
            .ok_or_else(|| SqlError::DataFusion {
                message: "right DataFrame must be SqlDataFrame for join".into(),
            })?;
        use datafusion::common::JoinType;
        let join_type = match how.to_lowercase().as_str() {
            "inner" => JoinType::Inner,
            "left" => JoinType::Left,
            "right" => JoinType::Right,
            "full" | "outer" => JoinType::Full,
            "leftsemi" | "left_semi" => JoinType::LeftSemi,
            "rightsemi" | "right_semi" => JoinType::RightSemi,
            "leftanti" | "left_anti" => JoinType::LeftAnti,
            "rightanti" | "right_anti" => JoinType::RightAnti,
            _ => {
                return Err(SqlError::DataFusion {
                    message: format!("unsupported join type: {how}"),
                });
            }
        };
        let df = self.dataframe.clone().join(
            right_sql.dataframe.clone(),
            join_type,
            left_on,
            right_on,
            None,
        )?;
        Ok(Box::new(self.with_new_dataframe(df, "join")))
    }

    async fn union(
        &self,
        right: &dyn KrishivDataFrameOps,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let right_sql = right
            .as_any()
            .downcast_ref::<SqlDataFrame>()
            .ok_or_else(|| SqlError::DataFusion {
                message: "right DataFrame must be SqlDataFrame for union".into(),
            })?;
        let df = self.dataframe.clone().union(right_sql.dataframe.clone())?;
        Ok(Box::new(self.with_new_dataframe(df, "union")))
    }

    async fn union_distinct(
        &self,
        right: &dyn KrishivDataFrameOps,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let right = sql_dataframe(right, "union_distinct")?;
        let df = self
            .dataframe
            .clone()
            .union_distinct(right.dataframe.clone())?;
        Ok(Box::new(self.with_new_dataframe(df, "union_distinct")))
    }

    async fn intersect(
        &self,
        right: &dyn KrishivDataFrameOps,
        distinct: bool,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let right = sql_dataframe(right, "intersect")?;
        let df = if distinct {
            self.dataframe
                .clone()
                .intersect_distinct(right.dataframe.clone())?
        } else {
            self.dataframe.clone().intersect(right.dataframe.clone())?
        };
        Ok(Box::new(self.with_new_dataframe(df, "intersect")))
    }

    async fn except(
        &self,
        right: &dyn KrishivDataFrameOps,
        distinct: bool,
    ) -> SqlResult<Box<dyn KrishivDataFrameOps>> {
        let right = sql_dataframe(right, "except")?;
        let df = if distinct {
            self.dataframe
                .clone()
                .except_distinct(right.dataframe.clone())?
        } else {
            self.dataframe.clone().except(right.dataframe.clone())?
        };
        Ok(Box::new(self.with_new_dataframe(df, "except")))
    }

    async fn register_batches(&self, name: &str, batches: Vec<RecordBatch>) -> SqlResult<()> {
        let schema = batches
            .first()
            .map(|b| b.schema())
            .unwrap_or_else(|| Arc::new(arrow::datatypes::Schema::empty()));
        let mem_table =
            datafusion::datasource::MemTable::try_new(schema, vec![batches]).map_err(|e| {
                SqlError::DataFusion {
                    message: e.to_string(),
                }
            })?;
        self.context
            .register_table(name, Arc::new(mem_table))
            .map_err(SqlError::from)?;
        Ok(())
    }

    async fn deregister_table(&self, name: &str) -> SqlResult<()> {
        let _ = self
            .context
            .deregister_table(name)
            .map_err(SqlError::from)?;
        Ok(())
    }

    async fn create_view(&self, name: &str, replace: bool) -> SqlResult<()> {
        let query = self
            .query_text
            .as_deref()
            .ok_or_else(|| SqlError::DataFusion {
                message: "create_view requires an SQL query string on the DataFrame".into(),
            })?;
        let or_replace = if replace { "OR REPLACE " } else { "" };
        let safe_name = quote_identifier(name);
        let view_sql = format!("CREATE {or_replace}VIEW {safe_name} AS {query}");
        self.context.sql(&view_sql).await?;
        Ok(())
    }
}

use krishiv_common::sql_util::quote_identifier;

// ── CALL-system helpers ───────────────────────────────────────────────────────

/// Extract positional arguments from the body of a `CALL` statement.
///
/// Handles single-quoted string literals and bare integers.
/// `'catalog.ns.table', '7 days', 5` → `["catalog.ns.table", "7 days", "5"]`
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
fn call_args_from_str(s: &str) -> Vec<String> {
    let mut args: Vec<String> = Vec::new();
    let mut cur = String::new();
    let mut in_str = false;
    let mut after_str = false;
    for ch in s.chars() {
        if after_str {
            if ch == ',' {
                after_str = false;
            }
            continue;
        }
        if in_str {
            if ch == '\'' {
                in_str = false;
                after_str = true;
                args.push(std::mem::take(&mut cur));
            } else {
                cur.push(ch);
            }
        } else if ch == '\'' {
            in_str = true;
        } else if ch == ',' {
            let t = cur.trim().to_string();
            if !t.is_empty() {
                args.push(t);
            }
            cur.clear();
        } else {
            cur.push(ch);
        }
    }
    let t = cur.trim().to_string();
    if !t.is_empty() {
        args.push(t);
    }
    args
}

/// Parse an Iceberg `TableIdent` from a dotted string.
///
/// Accepts:
/// - `"namespace.table"` — single-level namespace
/// - `"catalog.namespace.table"` — catalog prefix is ignored (catalog is
///   selected by registration order, not by name, in the CALL dispatch)
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
fn iceberg_table_ident(table_ref: &str) -> SqlResult<iceberg::TableIdent> {
    let parts: Vec<&str> = table_ref.splitn(3, '.').collect();
    match parts.len() {
        2 => {
            let ns = iceberg::NamespaceIdent::from_vec(vec![
                parts.first().copied().unwrap_or("").to_string(),
            ])
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;
            Ok(iceberg::TableIdent::new(
                ns,
                parts.get(1).copied().unwrap_or("").to_string(),
            ))
        }
        3 => {
            let ns = iceberg::NamespaceIdent::from_vec(vec![
                parts.get(1).copied().unwrap_or("").to_string(),
            ])
            .map_err(|e| SqlError::DataFusion {
                message: e.to_string(),
            })?;
            Ok(iceberg::TableIdent::new(
                ns,
                parts.get(2).copied().unwrap_or("").to_string(),
            ))
        }
        _ => Err(SqlError::DataFusion {
            message: format!(
                "invalid table reference '{table_ref}': expected 'ns.table' or 'cat.ns.table'"
            ),
        }),
    }
}

/// Parse a human-readable duration string into a [`chrono::Duration`].
///
/// Accepted formats: `"N days"`, `"N day"`, `"N hours"`, `"N hour"`,
/// `"N weeks"`, `"N week"`, `"N minutes"`, `"N minute"`.
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
fn parse_call_duration(s: &str) -> SqlResult<chrono::Duration> {
    let s = s.trim();
    let mut it = s.splitn(2, ' ');
    let n: i64 = it
        .next()
        .and_then(|v| v.parse().ok())
        .ok_or_else(|| SqlError::DataFusion {
            message: format!("invalid duration value in '{s}'"),
        })?;
    let unit = it.next().unwrap_or("").trim().to_ascii_lowercase();
    match unit.trim_end_matches('s') {
        "day" => Ok(chrono::Duration::days(n)),
        "hour" => Ok(chrono::Duration::hours(n)),
        "week" => Ok(chrono::Duration::weeks(n)),
        "minute" | "min" => Ok(chrono::Duration::minutes(n)),
        _ => Err(SqlError::DataFusion {
            message: format!("unknown duration unit '{unit}' in '{s}'"),
        }),
    }
}

// ── Iceberg DML helpers ───────────────────────────────────────────────────────

/// Parse `DELETE FROM <table> [WHERE <predicate>]` into `(table_ref, predicate)`
/// using the sqlparser AST, which correctly handles quoted identifiers, comments,
/// and subqueries in predicates.  Returns `None` for non-DELETE statements.
///
/// A missing WHERE clause is returned as `"TRUE"` (delete all rows).
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
fn parse_dml_delete(stmt: &str) -> Option<(String, String)> {
    use datafusion::sql::sqlparser::ast::{FromTable, Statement, TableFactor};
    use datafusion::sql::sqlparser::dialect::GenericDialect;
    use datafusion::sql::sqlparser::parser::Parser;

    let mut stmts = Parser::parse_sql(&GenericDialect {}, stmt).ok()?;
    if stmts.len() != 1 {
        return None;
    }
    let Statement::Delete(delete) = stmts.remove(0) else {
        return None;
    };
    // `Delete::from` is a `FromTable` enum (sqlparser ≥0.54); both arms carry the
    // table list. The first FROM table is the deletion target.
    let tables = match delete.from {
        FromTable::WithFromKeyword(tables) | FromTable::WithoutKeyword(tables) => tables,
    };
    let first_from = tables.into_iter().next()?;
    let table_name = match first_from.relation {
        TableFactor::Table { name, .. } => name.to_string(),
        _ => return None,
    };
    let predicate = delete
        .selection
        .map(|e| e.to_string())
        .unwrap_or_else(|| "TRUE".to_string());
    Some((table_name, predicate))
}

/// Parsed `INSERT INTO <table> [(<columns>)] <source>` statement.
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
struct ParsedInsert {
    /// Dotted target reference exactly as written (`cat.ns.tbl` or `ns.tbl`).
    table_ref: String,
    /// Explicit column list, if given (`INSERT INTO t (a, b) ...`). Empty
    /// means "all columns, in table order" — the only form this engine's
    /// Iceberg append landing currently accepts (#219 residual: an explicit
    /// list would need positional remapping + NULL-fill for omitted
    /// columns, not implemented yet).
    columns: Vec<String>,
    /// The source query text (`SELECT ...` or `VALUES (...)`), rendered
    /// back from the AST so it can be executed on its own.
    inner_query: String,
}

/// Parse `INSERT INTO <table> [(<columns>)] <source-query>` using the
/// sqlparser AST. Returns `None` for non-INSERT statements, multi-statement
/// input, or an INSERT with no source query (bare `DEFAULT VALUES` / MySQL
/// `SET`-assignment forms) — this engine only supports the `SELECT`/`VALUES`
/// source forms, both of which parse into `Insert::source`.
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
fn parse_dml_insert(stmt: &str) -> Option<ParsedInsert> {
    use datafusion::sql::sqlparser::ast::{Statement, TableObject};
    use datafusion::sql::sqlparser::dialect::GenericDialect;
    use datafusion::sql::sqlparser::parser::Parser;

    let mut stmts = Parser::parse_sql(&GenericDialect {}, stmt).ok()?;
    if stmts.len() != 1 {
        return None;
    }
    let Statement::Insert(insert) = stmts.remove(0) else {
        return None;
    };
    let TableObject::TableName(name) = insert.table else {
        return None;
    };
    let inner_query = insert.source?.to_string();
    Some(ParsedInsert {
        table_ref: name.to_string(),
        columns: insert.columns.iter().map(|c| c.to_string()).collect(),
        inner_query,
    })
}

/// Parsed `CREATE [OR REPLACE] TABLE … AS <query>` statement.
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
struct ParsedCtas {
    /// Dotted target reference exactly as written (`cat.ns.tbl` or `ns.tbl`).
    table_ref: String,
    or_replace: bool,
    /// The inner query text (sqlparser AST rendering of the AS body).
    inner_query: String,
    /// Raw `PARTITIONED BY` items (`region`, `bucket(4, id)`, `day(ts)`),
    /// empty for unpartitioned tables.
    partition_by: Vec<String>,
}

/// Lift a `PARTITIONED BY (…)` clause out of a CREATE TABLE statement.
///
/// Iceberg partition transforms (`bucket(4, id)`, `day(ts)`) are not valid
/// column definitions in sqlparser's Hive-style `PARTITIONED BY` list, so
/// the clause is extracted textually before the statement is parsed: the
/// keywords are matched case-insensitively outside single-quoted strings and
/// double-quoted identifiers, and the balanced-paren list that follows is
/// split on top-level commas. Returns the statement with the clause removed
/// plus the raw items; `None` when no clause is present.
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
fn extract_partitioned_by(stmt: &str) -> Option<(String, Vec<String>)> {
    let bytes = stmt.as_bytes();
    let upper = stmt.to_ascii_uppercase();
    let upper_bytes = upper.as_bytes();
    const NEEDLE: &[u8] = b"PARTITIONED";

    fn is_ident_byte(b: u8) -> bool {
        b.is_ascii_alphanumeric() || b == b'_'
    }
    // Advance past a quoted region starting at `i` (index of the opening
    // quote); `''` / `""` escapes stay inside the region.
    fn skip_quoted(bytes: &[u8], mut i: usize, quote: u8) -> usize {
        i += 1;
        while let Some(&b) = bytes.get(i) {
            if b == quote {
                if bytes.get(i + 1) == Some(&quote) {
                    i += 2;
                    continue;
                }
                return i + 1;
            }
            i += 1;
        }
        i
    }

    let mut i = 0;
    while let Some(&b) = bytes.get(i) {
        match b {
            b'\'' | b'"' => i = skip_quoted(bytes, i, b),
            _ => {
                let at_needle = upper_bytes
                    .get(i..)
                    .is_some_and(|rest| rest.starts_with(NEEDLE))
                    && (i == 0
                        || !i
                            .checked_sub(1)
                            .and_then(|p| upper_bytes.get(p))
                            .copied()
                            .is_some_and(is_ident_byte));
                if at_needle {
                    let mut j = i + NEEDLE.len();
                    while bytes.get(j).is_some_and(u8::is_ascii_whitespace) {
                        j += 1;
                    }
                    // Require whitespace between the keywords and `BY` to not
                    // be part of a longer identifier.
                    if j > i + NEEDLE.len()
                        && upper_bytes
                            .get(j..)
                            .is_some_and(|rest| rest.starts_with(b"BY"))
                        && !upper_bytes.get(j + 2).copied().is_some_and(is_ident_byte)
                    {
                        let mut k = j + 2;
                        while bytes.get(k).is_some_and(u8::is_ascii_whitespace) {
                            k += 1;
                        }
                        if bytes.get(k) == Some(&b'(') {
                            // Find the balanced close, respecting quotes.
                            let mut depth = 0i32;
                            let mut c = k;
                            let close = loop {
                                match bytes.get(c) {
                                    // unbalanced: let sqlparser reject it
                                    None => return None,
                                    Some(b'(') => depth += 1,
                                    Some(b')') => {
                                        depth -= 1;
                                        if depth == 0 {
                                            break c;
                                        }
                                    }
                                    Some(&(q @ b'\'' | q @ b'"')) => {
                                        c = skip_quoted(bytes, c, q);
                                        continue;
                                    }
                                    Some(_) => {}
                                }
                                c += 1;
                            };
                            let body = stmt.get(k + 1..close)?;
                            let head = stmt.get(..i)?.trim_end();
                            let tail = stmt.get(close + 1..)?.trim_start();
                            let items = split_top_level_commas(body);
                            let mut remainder = String::with_capacity(stmt.len());
                            remainder.push_str(head);
                            remainder.push(' ');
                            remainder.push_str(tail);
                            return Some((remainder, items));
                        }
                    }
                }
                i += 1;
            }
        }
    }
    None
}

/// Split a parenthesized list body on commas at paren depth zero, skipping
/// quoted regions. Empty items are dropped.
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
fn split_top_level_commas(s: &str) -> Vec<String> {
    let bytes = s.as_bytes();
    let mut items = Vec::new();
    let mut depth = 0i32;
    let mut start = 0usize;
    let mut i = 0;
    while let Some(&b) = bytes.get(i) {
        match b {
            b'(' => depth += 1,
            b')' => depth -= 1,
            b'\'' | b'"' => {
                i += 1;
                while bytes.get(i).is_some_and(|&c| c != b) {
                    i += 1;
                }
            }
            b',' if depth == 0 => {
                if let Some(item) = s.get(start..i).map(str::trim)
                    && !item.is_empty()
                {
                    items.push(item.to_string());
                }
                start = i + 1;
            }
            _ => {}
        }
        i += 1;
    }
    if let Some(last) = s.get(start..).map(str::trim)
        && !last.is_empty()
    {
        items.push(last.to_string());
    }
    items
}

/// Split a SQL body into its top-level statements.
///
/// Splits on `;` outside single-quoted literals (`''` escapes), double-quoted
/// identifiers, `--` line comments, and `/* … */` block comments, preserving
/// each statement's original text (no re-rendering, so literal contents are
/// never altered). Empty pieces (trailing semicolons, blank statements) are
/// dropped. A body with no top-level semicolon comes back as a single item.
fn split_sql_statements(sql: &str) -> Vec<String> {
    let mut items = Vec::new();
    let mut start = 0usize;
    let mut chars = sql.char_indices().peekable();
    while let Some((i, c)) = chars.next() {
        match c {
            '\'' => {
                // Single-quoted literal; '' is an escaped quote, not a close.
                while let Some((_, c2)) = chars.next() {
                    if c2 == '\'' {
                        if chars.peek().is_some_and(|&(_, c3)| c3 == '\'') {
                            chars.next();
                            continue;
                        }
                        break;
                    }
                }
            }
            '"' => {
                for (_, c2) in chars.by_ref() {
                    if c2 == '"' {
                        break;
                    }
                }
            }
            '-' if chars.peek().is_some_and(|&(_, c2)| c2 == '-') => {
                for (_, c2) in chars.by_ref() {
                    if c2 == '\n' {
                        break;
                    }
                }
            }
            '/' if chars.peek().is_some_and(|&(_, c2)| c2 == '*') => {
                chars.next();
                let mut star = false;
                for (_, c2) in chars.by_ref() {
                    if star && c2 == '/' {
                        break;
                    }
                    star = c2 == '*';
                }
            }
            ';' => {
                if let Some(piece) = sql.get(start..i).map(str::trim)
                    && !piece.is_empty()
                {
                    items.push(piece.to_string());
                }
                // ';' is ASCII — one byte — so i + 1 is a char boundary.
                start = i + 1;
            }
            _ => {}
        }
    }
    if let Some(last) = sql.get(start..).map(str::trim)
        && !last.is_empty()
    {
        items.push(last.to_string());
    }
    items
}

/// Parse `CREATE [OR REPLACE] TABLE <ref> [PARTITIONED BY (…)] AS <query>`
/// using the sqlparser AST (with the PARTITIONED BY clause lifted out
/// textually first — see [`extract_partitioned_by`]).
///
/// Returns `None` for anything else — plain column-list CREATE TABLE,
/// CREATE EXTERNAL/TEMPORARY TABLE (DataFusion's own DDL), multi-statement
/// input, or unparseable text — so callers fall through to DataFusion.
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
fn parse_ctas(stmt: &str) -> Option<ParsedCtas> {
    use datafusion::sql::sqlparser::ast::Statement;
    use datafusion::sql::sqlparser::dialect::GenericDialect;
    use datafusion::sql::sqlparser::parser::Parser;

    let (stripped, partition_by) = match extract_partitioned_by(stmt) {
        Some((remainder, items)) => (remainder, items),
        None => (stmt.to_string(), Vec::new()),
    };
    let mut stmts = Parser::parse_sql(&GenericDialect {}, &stripped).ok()?;
    if stmts.len() != 1 {
        return None;
    }
    let Statement::CreateTable(create) = stmts.remove(0) else {
        return None;
    };
    if create.external || create.temporary {
        return None;
    }
    let inner_query = create.query?.to_string();
    Some(ParsedCtas {
        table_ref: create.name.to_string(),
        or_replace: create.or_replace,
        inner_query,
        partition_by,
    })
}

/// Parsed UPDATE statement, decomposed into its components for Iceberg DML.
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
struct ParsedUpdate {
    table_ref: String,
    /// Ordered (column_name, value_expression) pairs from the SET clause.
    assignments: Vec<(String, String)>,
    predicate: Option<String>,
}

/// Parse `UPDATE <table> SET col = expr [, …] [WHERE <predicate>]` using the
/// sqlparser AST.  Returns `None` for non-UPDATE statements or unsupported shapes.
///
/// Replaces the former regex implementation which could not handle quoted
/// identifiers, expressions with commas, or subqueries in predicates.
#[cfg(all(feature = "iceberg-datafusion", feature = "local-catalog"))]
fn parse_dml_update(stmt: &str) -> Option<ParsedUpdate> {
    use datafusion::sql::sqlparser::ast::{Statement, TableFactor};
    use datafusion::sql::sqlparser::dialect::GenericDialect;
    use datafusion::sql::sqlparser::parser::Parser;

    let mut stmts = Parser::parse_sql(&GenericDialect {}, stmt).ok()?;
    if stmts.len() != 1 {
        return None;
    }
    // `Statement::Update` wraps an `Update` struct (sqlparser ≥0.55).
    let Statement::Update(update) = stmts.remove(0) else {
        return None;
    };
    let table_name = match update.table.relation {
        TableFactor::Table { name, .. } => name.to_string(),
        _ => return None,
    };
    // Convert AST assignments to (column_name, expression_string) pairs.
    let parsed_assignments: Vec<(String, String)> = update
        .assignments
        .into_iter()
        .map(|a| {
            // `target` is `AssignmentTarget::ColumnName(ObjectName)` in 0.61.
            let col = a.target.to_string();
            let val = a.value.to_string();
            (col, val)
        })
        .collect();
    if parsed_assignments.is_empty() {
        return None;
    }
    Some(ParsedUpdate {
        table_ref: table_name,
        assignments: parsed_assignments,
        predicate: update.selection.map(|e| e.to_string()),
    })
}

/// Create a Krishiv logical plan wrapper for a SQL query without executing it.
pub fn plan_sql(query: impl Into<String>) -> SqlResult<SqlPlan> {
    let query = query.into();
    if query.trim().is_empty() {
        return Err(SqlError::EmptyQuery);
    }

    if let Some(stmt) = cep_sql::parse_match_recognize(&query)? {
        let logical_plan = cep_sql::plan_match_recognize(stmt, &query);
        let optimized = Optimizer::default().optimize(logical_plan)?;
        return Ok(SqlPlan {
            query,
            logical_plan: optimized.plan,
        });
    }

    let logical_plan =
        LogicalPlan::new("sql-query", ExecutionKind::Batch).with_node(PlanNode::new(
            "sql",
            format!("sql: {}", query.trim()),
            ExecutionKind::Batch,
        ));

    let optimized = Optimizer::default().optimize(logical_plan)?;
    Ok(SqlPlan {
        query,
        logical_plan: optimized.plan,
    })
}

/// Create bootstrap `EXPLAIN` text for a SQL query.
pub fn explain_sql(query: impl Into<String>) -> SqlResult<String> {
    let plan = plan_sql(query)?;
    Ok(plan.logical_plan().describe())
}

/// Explain a SQL query including optimizer rule decisions.
///
/// Runs the logical plan through `optimizer` and appends the optimizer
/// summary to the plan description.
pub fn explain_sql_optimized(query: impl Into<String>, optimizer: &Optimizer) -> SqlResult<String> {
    let plan = plan_sql(query)?;
    let result = optimizer.optimize(plan.logical_plan().clone())?;
    let mut output = result.plan.describe();
    let optimizer_line = result.describe();
    output.push('\n');
    output.push_str(&optimizer_line);
    Ok(output)
}

/// Explain a SQL query and append a cost estimate from the provided cost model.
pub fn explain_sql_with_cost(
    query: impl Into<String>,
    cost_model: &dyn CostModel,
) -> SqlResult<String> {
    let plan = plan_sql(query)?;
    let cost = cost_model.estimate(plan.logical_plan());
    let mut output = plan.logical_plan().describe();
    output.push_str(&format!(
        "\ncost: cpu_nanos={}, memory_bytes={}, network_bytes={}",
        cost.cpu_nanos, cost.memory_bytes, cost.network_bytes
    ));
    Ok(output)
}

/// Return all base table/relation names referenced by `query`.
///
/// This uses the same SQL parser family as DataFusion, so policy checks cover
/// joins, subqueries, CTE bodies, and other nested relation references instead
/// of relying on a single best-effort `FROM` token.
pub fn referenced_table_names(query: impl AsRef<str>) -> SqlResult<Vec<String>> {
    let query = query.as_ref();
    if query.trim().is_empty() {
        return Err(SqlError::EmptyQuery);
    }

    let statements =
        Parser::parse_sql(&GenericDialect {}, query).map_err(|e| SqlError::DataFusion {
            message: format!("SQL parse error: {e}"),
        })?;
    let mut names = BTreeSet::new();
    let _ = visit_relations(&statements, |relation| {
        names.insert(relation.to_string());
        ControlFlow::<()>::Continue(())
    });
    Ok(names.into_iter().collect())
}

/// Format Arrow batches for CLI and tests.
pub fn pretty_batches(batches: &[RecordBatch]) -> SqlResult<String> {
    Ok(pretty_format_batches(batches)
        .map_err(|error| SqlError::DataFusion {
            message: error.to_string(),
        })?
        .to_string())
}

#[cfg(test)]
mod sql_tests;