datum-sql 0.10.3

#![forbid(unsafe_code)]
#![doc = include_str!("../README.md")]

mod join;
mod sink;
mod time;
mod window;

use std::collections::BTreeMap;
use std::fmt;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use arrow::array::{Array, RecordBatch};
use arrow::compute::filter_record_batch;
use arrow::datatypes::{DataType, Field, FieldRef, SchemaRef};
use async_trait::async_trait;
use datafusion::catalog::{Session, TableProvider};
use datafusion::common::ColumnStatistics;
use datafusion::common::ScalarValue;
use datafusion::common::assert_or_internal_err;
use datafusion::common::cast::as_boolean_array;
use datafusion::common::stats::Precision;
use datafusion::common::{DataFusionError, Result, project_schema};
use datafusion::execution::TaskContext;
use datafusion::execution::context::SessionContext;
use datafusion::logical_expr::{
    ColumnarValue, Expr, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature,
    TableProviderFilterPushDown, TableType, Volatility,
};
use datafusion::physical_expr::EquivalenceProperties;
use datafusion::physical_expr::projection::ProjectionExpr;
use datafusion::physical_plan::PhysicalExpr;
use datafusion::physical_plan::Statistics;
use datafusion::physical_plan::aggregates::AggregateExec;
#[allow(deprecated)]
use datafusion::physical_plan::coalesce_batches::CoalesceBatchesExec;
use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
use datafusion::physical_plan::execution_plan::{Boundedness, EmissionType, SchedulingType};
use datafusion::physical_plan::filter::FilterExec;
use datafusion::physical_plan::joins::HashJoinExec;
use datafusion::physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
use datafusion::physical_plan::memory::MemoryStream;
use datafusion::physical_plan::projection::ProjectionExec;
use datafusion::physical_plan::{
    DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties,
    SendableRecordBatchStream,
};
use datafusion::prelude::SessionConfig;
use datafusion::sql::parser::{DFParser, Statement as DfStatement};
use datafusion::sql::sqlparser::ast::{Statement as SqlStatement, TableObject as SqlTableObject};
use datum::{Keep, KillSwitches, Sink, Source, StreamError, StreamResult};

pub use join::{
    StreamingJoinConfig, StreamingJoinMetrics, StreamingJoinMode, StreamingJoinStateLimits,
    StreamingJoinWindow,
};
use sink::{AppendSinkSource, RegisteredSqlSink, sink_registry_error};
pub use sink::{CommittableRecordBatch, SourceCommit, SqlSourcePosition};
pub use time::{
    ContinuousQueryHandle, EventTimeConfig, SqlBarrier, SqlEvent, SqlEventPayload, Watermark,
    WatermarkStrategy, assign_event_time_watermarks, data_events, map_sql_event_data,
};
pub use window::WindowedAggregationMetrics;

pub mod connect;
#[cfg(feature = "cdc")]
pub use connect::cdc::{CdcChangelogBatch, CdcSourcePosition, CdcTableAdapter};
#[cfg(feature = "mq")]
pub use connect::mq::{
    JsonRowFormat, KafkaPartitionOffset, KafkaRecordBatch, KafkaSourcePosition, MqPayloadFormat,
};
pub use connect::{BatchEnvelope, EnvelopedBatch, EnvelopedRecordBatch};

type SourceFactory = Arc<dyn Fn() -> Source<RecordBatch> + Send + Sync>;
type CommittableSourceFactory = Arc<dyn Fn() -> Source<CommittableRecordBatch> + Send + Sync>;
type ChangelogSourceFactory = Arc<dyn Fn() -> Source<ChangelogBatch> + Send + Sync>;

/// Row-level operation for SQL streams that can revise prior output.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChangeOp {
    /// Adds a row that was not previously present.
    Insert,
    /// Retracts a row without a replacement.
    Delete,
    /// First half of an adjacent update pair, carrying the old row.
    UpdateDelete,
    /// Second half of an adjacent update pair, carrying the new row.
    UpdateInsert,
}

impl ChangeOp {
    /// Returns true for rows that add state to an updating stream.
    #[must_use]
    pub const fn is_positive(self) -> bool {
        matches!(self, Self::Insert | Self::UpdateInsert)
    }

    /// Returns true for rows that retract state from an updating stream.
    #[must_use]
    pub const fn is_retraction(self) -> bool {
        matches!(self, Self::Delete | Self::UpdateDelete)
    }
}

/// A typed changelog envelope over a plain Arrow `RecordBatch`.
///
/// Append-only SQL stays as `Source<RecordBatch>`. This type is only used once a
/// source or operator can emit retractions.
#[derive(Clone)]
pub struct ChangelogBatch {
    ops: Arc<[ChangeOp]>,
    batch: RecordBatch,
}

impl ChangelogBatch {
    /// Builds a changelog batch and validates that row operations match the
    /// payload row count and update-pair adjacency rules.
    pub fn try_new(ops: Vec<ChangeOp>, batch: RecordBatch) -> Result<Self> {
        validate_change_ops(&ops, batch.num_rows())?;
        Ok(Self {
            ops: Arc::from(ops),
            batch,
        })
    }

    /// Wraps an append-only Arrow batch as insert-only changelog data.
    #[must_use]
    pub fn insert_only(batch: RecordBatch) -> Self {
        Self {
            ops: vec![ChangeOp::Insert; batch.num_rows()].into(),
            batch,
        }
    }

    /// Returns the row-level changelog operations.
    #[must_use]
    pub fn ops(&self) -> &[ChangeOp] {
        &self.ops
    }

    /// Returns the Arrow payload batch.
    #[must_use]
    pub fn batch(&self) -> &RecordBatch {
        &self.batch
    }

    /// Returns the number of rows in the payload batch.
    #[must_use]
    pub fn num_rows(&self) -> usize {
        self.batch.num_rows()
    }

    /// Returns true when every row is an insert.
    #[must_use]
    pub fn is_insert_only(&self) -> bool {
        self.ops.iter().all(|op| *op == ChangeOp::Insert)
    }
}

impl fmt::Debug for ChangelogBatch {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ChangelogBatch")
            .field("ops", &self.ops)
            .field("batch", &self.batch)
            .finish()
    }
}

fn validate_change_ops(ops: &[ChangeOp], rows: usize) -> Result<()> {
    if ops.len() != rows {
        return Err(DataFusionError::Plan(format!(
            "changelog batch has {} ops for {rows} rows",
            ops.len()
        )));
    }

    let mut row = 0;
    while row < ops.len() {
        match ops[row] {
            ChangeOp::UpdateDelete => {
                if ops.get(row + 1) != Some(&ChangeOp::UpdateInsert) {
                    return Err(DataFusionError::Plan(format!(
                        "UpdateDelete at row {row} must be immediately followed by UpdateInsert"
                    )));
                }
                row += 2;
            }
            ChangeOp::UpdateInsert => {
                return Err(DataFusionError::Plan(format!(
                    "UpdateInsert at row {row} must be immediately preceded by UpdateDelete"
                )));
            }
            ChangeOp::Insert | ChangeOp::Delete => row += 1,
        }
    }

    Ok(())
}

/// Per-query counters populated by SQL lowering and native Datum stages.
#[derive(Debug, Clone, Default)]
pub struct SqlExecutionMetrics {
    late_dropped_rows: Arc<AtomicU64>,
    join_state_rows: Arc<AtomicU64>,
    join_evicted_rows: Arc<AtomicU64>,
}

impl SqlExecutionMetrics {
    /// Rows dropped after the active watermark in windowed or joined streams.
    #[must_use]
    pub fn late_dropped_rows(&self) -> u64 {
        self.late_dropped_rows.load(Ordering::Relaxed)
    }

    pub(crate) fn window_metrics(&self) -> WindowedAggregationMetrics {
        WindowedAggregationMetrics::new(Arc::clone(&self.late_dropped_rows))
    }

    /// Returns a snapshot handle for streaming join counters.
    #[must_use]
    pub fn streaming_join_metrics(&self) -> StreamingJoinMetrics {
        self.join_metrics()
    }

    /// Rows currently retained by streaming join state.
    #[must_use]
    pub fn join_state_rows(&self) -> u64 {
        self.join_state_rows.load(Ordering::Relaxed)
    }

    /// Rows evicted from streaming join state by watermark progress.
    #[must_use]
    pub fn join_evicted_rows(&self) -> u64 {
        self.join_evicted_rows.load(Ordering::Relaxed)
    }

    pub(crate) fn join_metrics(&self) -> StreamingJoinMetrics {
        StreamingJoinMetrics::new(
            Arc::clone(&self.join_state_rows),
            Arc::clone(&self.join_evicted_rows),
            Arc::clone(&self.late_dropped_rows),
        )
    }
}

/// SQL planning context that registers Datum sources and executes supported
/// DataFusion physical plans on Datum.
pub struct DatumSqlContext {
    inner: SessionContext,
    sinks: Arc<Mutex<BTreeMap<String, RegisteredSqlSink>>>,
    streaming_join_config: StreamingJoinConfig,
}

impl DatumSqlContext {
    /// Creates a SQL context with one DataFusion target partition and Datum's
    /// planner placeholder functions registered.
    #[must_use]
    pub fn new() -> Self {
        let config = SessionConfig::new().with_target_partitions(1);
        let inner = SessionContext::new_with_config(config);
        register_window_placeholder_udfs(&inner);
        Self {
            inner,
            sinks: Arc::new(Mutex::new(BTreeMap::new())),
            streaming_join_config: StreamingJoinConfig::default(),
        }
    }

    /// Returns this context with a different streaming join policy.
    #[must_use]
    pub fn with_streaming_join_config(mut self, config: StreamingJoinConfig) -> Self {
        self.streaming_join_config = config;
        self
    }

    /// Replaces the streaming join policy used by future query lowering.
    pub fn set_streaming_join_config(&mut self, config: StreamingJoinConfig) {
        self.streaming_join_config = config;
    }

    /// Returns the configured streaming join policy.
    #[must_use]
    pub const fn streaming_join_config(&self) -> &StreamingJoinConfig {
        &self.streaming_join_config
    }

    /// Registers an append-only Datum source as a SQL table.
    pub fn register_source(
        &self,
        name: &str,
        schema: SchemaRef,
        source: Source<RecordBatch>,
    ) -> Result<()> {
        self.register_source_factory(name, schema, move || source.clone())
    }

    /// Registers an append-only Datum source and event-time declaration as a
    /// SQL table.
    pub fn register_source_with_event_time(
        &self,
        name: &str,
        schema: SchemaRef,
        source: Source<RecordBatch>,
        event_time: EventTimeConfig,
    ) -> Result<()> {
        self.register_source_factory_with_event_time(
            name,
            schema,
            move || source.clone(),
            event_time,
        )
    }

    /// Registers a factory for append-only Datum sources as a SQL table.
    pub fn register_source_factory<F>(
        &self,
        name: &str,
        schema: SchemaRef,
        source_factory: F,
    ) -> Result<()>
    where
        F: Fn() -> Source<RecordBatch> + Send + Sync + 'static,
    {
        let table = DatumTable::new(schema, Arc::new(source_factory), None);
        self.inner.register_table(name, Arc::new(table))?;
        Ok(())
    }

    /// Registers a factory for append-only Datum sources with event time.
    pub fn register_source_factory_with_event_time<F>(
        &self,
        name: &str,
        schema: SchemaRef,
        source_factory: F,
        event_time: EventTimeConfig,
    ) -> Result<()>
    where
        F: Fn() -> Source<RecordBatch> + Send + Sync + 'static,
    {
        event_time.resolve(schema.as_ref())?;
        let table = DatumTable::new(schema, Arc::new(source_factory), Some(event_time));
        self.inner.register_table(name, Arc::new(table))?;
        Ok(())
    }

    /// Registers a committable append-only source as a SQL table.
    pub fn register_committable_source(
        &self,
        name: &str,
        schema: SchemaRef,
        source: Source<CommittableRecordBatch>,
    ) -> Result<()> {
        self.register_committable_source_factory(name, schema, move || source.clone())
    }

    /// Registers a factory for committable append-only sources as a SQL table.
    pub fn register_committable_source_factory<F>(
        &self,
        name: &str,
        schema: SchemaRef,
        source_factory: F,
    ) -> Result<()>
    where
        F: Fn() -> Source<CommittableRecordBatch> + Send + Sync + 'static,
    {
        let table = DatumCommittableTable::new(schema, Arc::new(source_factory), None);
        self.inner.register_table(name, Arc::new(table))?;
        Ok(())
    }

    /// Registers a committable append-only source and event-time declaration as
    /// a SQL table.
    pub fn register_committable_source_with_event_time(
        &self,
        name: &str,
        schema: SchemaRef,
        source: Source<CommittableRecordBatch>,
        event_time: EventTimeConfig,
    ) -> Result<()> {
        self.register_committable_source_factory_with_event_time(
            name,
            schema,
            move || source.clone(),
            event_time,
        )
    }

    /// Registers a factory for committable append-only sources with event time.
    pub fn register_committable_source_factory_with_event_time<F>(
        &self,
        name: &str,
        schema: SchemaRef,
        source_factory: F,
        event_time: EventTimeConfig,
    ) -> Result<()>
    where
        F: Fn() -> Source<CommittableRecordBatch> + Send + Sync + 'static,
    {
        event_time.resolve(schema.as_ref())?;
        let table = DatumCommittableTable::new(schema, Arc::new(source_factory), Some(event_time));
        self.inner.register_table(name, Arc::new(table))?;
        Ok(())
    }

    /// Registers an updating source as a SQL table.
    pub fn register_changelog_source(
        &self,
        name: &str,
        schema: SchemaRef,
        source: Source<ChangelogBatch>,
    ) -> Result<()> {
        self.register_changelog_source_factory(name, schema, move || source.clone())
    }

    /// Registers an updating source and event-time declaration as a SQL table.
    pub fn register_changelog_source_with_event_time(
        &self,
        name: &str,
        schema: SchemaRef,
        source: Source<ChangelogBatch>,
        event_time: EventTimeConfig,
    ) -> Result<()> {
        self.register_changelog_source_factory_with_event_time(
            name,
            schema,
            move || source.clone(),
            event_time,
        )
    }

    /// Registers a factory for updating sources as a SQL table.
    pub fn register_changelog_source_factory<F>(
        &self,
        name: &str,
        schema: SchemaRef,
        source_factory: F,
    ) -> Result<()>
    where
        F: Fn() -> Source<ChangelogBatch> + Send + Sync + 'static,
    {
        let table = DatumChangelogTable::new(schema, Arc::new(source_factory), None);
        self.inner.register_table(name, Arc::new(table))?;
        Ok(())
    }

    /// Registers a factory for updating sources with event time.
    pub fn register_changelog_source_factory_with_event_time<F>(
        &self,
        name: &str,
        schema: SchemaRef,
        source_factory: F,
        event_time: EventTimeConfig,
    ) -> Result<()>
    where
        F: Fn() -> Source<ChangelogBatch> + Send + Sync + 'static,
    {
        event_time.resolve(schema.as_ref())?;
        let table = DatumChangelogTable::new(schema, Arc::new(source_factory), Some(event_time));
        self.inner.register_table(name, Arc::new(table))?;
        Ok(())
    }

    #[cfg(feature = "mq")]
    /// Registers a Kafka JSON topic as a committable SQL table.
    pub fn register_mq_topic(
        &self,
        name: &str,
        settings: datum_mq::KafkaConsumerSettings,
        topic: impl Into<String>,
        format: connect::mq::JsonRowFormat,
    ) -> Result<()> {
        let topic = topic.into();
        self.register_committable_source_factory(name, format.schema(), move || {
            connect::mq::kafka_json_committable_source_uncontrolled(
                settings.clone(),
                topic.clone(),
                format.clone(),
            )
        })
    }

    #[cfg(feature = "mq")]
    /// Registers a Kafka JSON topic and event-time declaration as a
    /// committable SQL table.
    pub fn register_mq_topic_with_event_time(
        &self,
        name: &str,
        settings: datum_mq::KafkaConsumerSettings,
        topic: impl Into<String>,
        format: connect::mq::JsonRowFormat,
        event_time: EventTimeConfig,
    ) -> Result<()> {
        let topic = topic.into();
        self.register_committable_source_factory_with_event_time(
            name,
            format.schema(),
            move || {
                connect::mq::kafka_json_committable_source_uncontrolled(
                    settings.clone(),
                    topic.clone(),
                    format.clone(),
                )
            },
            event_time,
        )
    }

    #[cfg(feature = "mq")]
    /// Registers a Kafka JSON producer as an append-only SQL sink.
    pub fn register_mq_json_sink(
        &self,
        name: &str,
        settings: datum_mq::KafkaProducerSettings,
        topic: impl Into<String>,
        format: connect::mq::JsonRowFormat,
    ) -> Result<()> {
        self.register_sql_sink(
            name,
            RegisteredSqlSink::kafka_json(settings, topic.into(), format),
        )
    }

    #[cfg(feature = "cdc")]
    /// Registers a CDC source factory as an updating SQL table.
    pub fn register_cdc_source<F>(
        &self,
        name: &str,
        schema: SchemaRef,
        source_factory: F,
    ) -> Result<()>
    where
        F: Fn() -> Source<datum_cdc::ChangeEvent, datum_cdc::CdcHandle> + Send + Sync + 'static,
    {
        self.register_changelog_source_factory(name, Arc::clone(&schema), move || {
            connect::cdc::cdc_changelog_batch_source_uncontrolled(
                source_factory(),
                Arc::clone(&schema),
            )
        })
    }

    /// Returns DataFusion's physical plan for SQL text without lowering it.
    pub async fn physical_plan(&self, sql: &str) -> Result<Arc<dyn ExecutionPlan>> {
        self.inner.sql(sql).await?.create_physical_plan().await
    }

    /// Executes a bounded append-only query and collects the output batches.
    pub async fn execute(&self, sql: &str) -> Result<Vec<RecordBatch>> {
        let plan = self.physical_plan(sql).await?;
        let source = lower_physical_plan(plan.as_ref())?;
        source
            .run_with(Sink::collect())
            .map_err(datafusion_error)?
            .wait()
            .map_err(datafusion_error)
    }

    /// Lowers a continuous SQL query into a Datum source of data/control events.
    pub async fn streaming_source(&self, sql: &str) -> Result<Source<SqlEvent<RecordBatch>>> {
        self.streaming_source_with_metrics(sql)
            .await
            .map(|(source, _metrics)| source)
    }

    /// Lowers a continuous SQL query and returns shared execution metrics.
    pub async fn streaming_source_with_metrics(
        &self,
        sql: &str,
    ) -> Result<(Source<SqlEvent<RecordBatch>>, SqlExecutionMetrics)> {
        let plan = self.physical_plan(sql).await?;
        let metrics = SqlExecutionMetrics::default();
        let source = lower_physical_plan_streaming_with_metrics_and_config(
            plan.as_ref(),
            &metrics,
            &self.streaming_join_config,
        )?;
        Ok((source, metrics))
    }

    /// Materializes a continuous SQL query, or an `INSERT INTO` sink query.
    pub async fn execute_streaming(&self, sql: &str) -> Result<ContinuousQueryHandle> {
        if looks_like_insert_into(sql) {
            return self.execute_insert_into(sql).await;
        }
        let source = self.streaming_source(sql).await?;
        let (kill_switch, completion) = source
            .via_mat(KillSwitches::single(), Keep::right)
            .to_mat(Sink::ignore(), Keep::both)
            .run()
            .map_err(datafusion_error)?;
        Ok(ContinuousQueryHandle::new(kill_switch, completion))
    }

    /// Parses and materializes `INSERT INTO <sink> SELECT ...`.
    pub async fn execute_insert_into(&self, sql: &str) -> Result<ContinuousQueryHandle> {
        let insert = parse_insert_into(sql)?;
        self.select_into(&insert.select_sql, &insert.sink_name)
            .await
    }

    /// Materializes a `SELECT` body into a registered sink.
    pub async fn select_into(
        &self,
        select_sql: &str,
        sink_name: &str,
    ) -> Result<ContinuousQueryHandle> {
        let sink = self.lookup_sink(sink_name)?;
        let plan = self.physical_plan(select_sql).await?;
        self.run_plan_into_sink(plan.as_ref(), sink)
    }

    /// Registers an append-only sink callback.
    pub fn register_append_sink<F>(&self, name: &str, writer: F) -> Result<()>
    where
        F: Fn(RecordBatch) -> StreamResult<()> + Send + Sync + 'static,
    {
        self.register_sql_sink(name, RegisteredSqlSink::append_only(writer))
    }

    /// Registers a sink callback that can consume both append-only and updating
    /// streams.
    pub fn register_changelog_sink<F, G>(
        &self,
        name: &str,
        append_writer: F,
        changelog_writer: G,
    ) -> Result<()>
    where
        F: Fn(RecordBatch) -> StreamResult<()> + Send + Sync + 'static,
        G: Fn(ChangelogBatch) -> StreamResult<()> + Send + Sync + 'static,
    {
        self.register_sql_sink(
            name,
            RegisteredSqlSink::changelog(append_writer, changelog_writer),
        )
    }

    fn register_sql_sink(&self, name: &str, sink: RegisteredSqlSink) -> Result<()> {
        self.sinks
            .lock()
            .map_err(|_| sink_registry_error("datum-sql sink registry lock poisoned"))?
            .insert(name.to_owned(), sink);
        Ok(())
    }

    fn lookup_sink(&self, name: &str) -> Result<RegisteredSqlSink> {
        self.sinks
            .lock()
            .map_err(|_| sink_registry_error("datum-sql sink registry lock poisoned"))?
            .get(name)
            .cloned()
            .ok_or_else(|| {
                DataFusionError::Plan(format!("datum-sql sink '{name}' is not registered"))
            })
    }

    fn run_plan_into_sink(
        &self,
        plan: &dyn ExecutionPlan,
        sink: RegisteredSqlSink,
    ) -> Result<ContinuousQueryHandle> {
        if plan_contains_changelog_scan(plan) {
            if !sink.accepts_changelog() {
                return Err(DataFusionError::Plan(
                    "datum-sql append-only sink cannot consume an updating stream; register a changelog-aware sink to accept retractions".into(),
                ));
            }
            let source = lower_changelog_physical_plan(plan)?;
            let (kill_switch, completion) = source
                .via_mat(KillSwitches::single(), Keep::right)
                .to_mat(
                    Sink::foreach_result(move |changes| sink.write_changelog(changes)),
                    Keep::both,
                )
                .run()
                .map_err(datafusion_error)?;
            return Ok(ContinuousQueryHandle::new(kill_switch, completion));
        }

        let source = lower_append_sink_plan(plan)?.into_committable();
        let (kill_switch, completion) = source
            .via_mat(KillSwitches::single(), Keep::right)
            .to_mat(
                Sink::foreach_result(move |batch: CommittableRecordBatch| {
                    sink.write_append(batch.batch().clone())?;
                    batch.commit()
                }),
                Keep::both,
            )
            .run()
            .map_err(datafusion_error)?;
        Ok(ContinuousQueryHandle::new(kill_switch, completion))
    }
}

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

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

fn register_window_placeholder_udfs(context: &SessionContext) {
    context.register_udf(ScalarUDF::from(WindowPlaceholderUdf::new("tumble", 2)));
    context.register_udf(ScalarUDF::from(WindowPlaceholderUdf::new("hop", 3)));
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct WindowPlaceholderUdf {
    name: &'static str,
    arity: usize,
    signature: Signature,
}

impl WindowPlaceholderUdf {
    fn new(name: &'static str, arity: usize) -> Self {
        Self {
            name,
            arity,
            signature: Signature::variadic_any(Volatility::Immutable),
        }
    }

    fn timestamp_arg<'a>(&self, args: &'a [FieldRef]) -> Result<&'a Field> {
        if args.len() != self.arity {
            return Err(DataFusionError::Plan(format!(
                "{} expects {} arguments, found {}",
                self.name,
                self.arity,
                args.len()
            )));
        }
        let timestamp_index = self.arity - 1;
        let field = args[timestamp_index].as_ref();
        if !matches!(field.data_type(), DataType::Timestamp(_, _)) {
            return Err(DataFusionError::Plan(format!(
                "{} expects its final argument to be a timestamp, found {:?}",
                self.name,
                field.data_type()
            )));
        }
        Ok(field)
    }
}

impl ScalarUDFImpl for WindowPlaceholderUdf {
    fn name(&self) -> &str {
        self.name
    }

    fn signature(&self) -> &Signature {
        &self.signature
    }

    fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> {
        if arg_types.len() != self.arity {
            return Err(DataFusionError::Plan(format!(
                "{} expects {} arguments, found {}",
                self.name,
                self.arity,
                arg_types.len()
            )));
        }
        match &arg_types[self.arity - 1] {
            DataType::Timestamp(unit, timezone) => Ok(DataType::Timestamp(*unit, timezone.clone())),
            other => Err(DataFusionError::Plan(format!(
                "{} expects its final argument to be a timestamp, found {other:?}",
                self.name
            ))),
        }
    }

    fn return_field_from_args(&self, args: ReturnFieldArgs<'_>) -> Result<FieldRef> {
        let timestamp = self.timestamp_arg(args.arg_fields)?;
        Ok(Arc::new(Field::new(
            self.name,
            timestamp.data_type().clone(),
            timestamp.is_nullable(),
        )))
    }

    fn invoke_with_args(&self, args: ScalarFunctionArgs) -> Result<ColumnarValue> {
        args.args.get(self.arity - 1).cloned().ok_or_else(|| {
            DataFusionError::Plan(format!("{} missing timestamp argument", self.name))
        })
    }
}

#[derive(Clone)]
struct DatumTable {
    schema: SchemaRef,
    source_factory: SourceFactory,
    event_time: Option<EventTimeConfig>,
}

impl DatumTable {
    fn new(
        schema: SchemaRef,
        source_factory: SourceFactory,
        event_time: Option<EventTimeConfig>,
    ) -> Self {
        Self {
            schema,
            source_factory,
            event_time,
        }
    }
}

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

#[derive(Clone)]
struct DatumCommittableTable {
    schema: SchemaRef,
    source_factory: CommittableSourceFactory,
    event_time: Option<EventTimeConfig>,
}

impl DatumCommittableTable {
    fn new(
        schema: SchemaRef,
        source_factory: CommittableSourceFactory,
        event_time: Option<EventTimeConfig>,
    ) -> Self {
        Self {
            schema,
            source_factory,
            event_time,
        }
    }
}

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

#[derive(Clone)]
struct DatumChangelogTable {
    schema: SchemaRef,
    source_factory: ChangelogSourceFactory,
    event_time: Option<EventTimeConfig>,
}

impl DatumChangelogTable {
    fn new(
        schema: SchemaRef,
        source_factory: ChangelogSourceFactory,
        event_time: Option<EventTimeConfig>,
    ) -> Self {
        Self {
            schema,
            source_factory,
            event_time,
        }
    }
}

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

#[async_trait]
impl TableProvider for DatumTable {
    fn schema(&self) -> SchemaRef {
        Arc::clone(&self.schema)
    }

    fn table_type(&self) -> TableType {
        TableType::Base
    }

    async fn scan(
        &self,
        _state: &dyn Session,
        projection: Option<&Vec<usize>>,
        filters: &[Expr],
        limit: Option<usize>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        if !filters.is_empty() {
            return Err(DataFusionError::Internal(
                "datum-sql scan received pushed filters despite reporting them unsupported".into(),
            ));
        }
        Ok(Arc::new(DatumScanExec::try_new(
            Arc::clone(&self.schema),
            Arc::clone(&self.source_factory),
            projection.cloned(),
            limit,
            self.event_time.clone(),
        )?))
    }

    fn supports_filters_pushdown(
        &self,
        filters: &[&Expr],
    ) -> Result<Vec<TableProviderFilterPushDown>> {
        Ok(vec![
            TableProviderFilterPushDown::Unsupported;
            filters.len()
        ])
    }
}

#[async_trait]
impl TableProvider for DatumCommittableTable {
    fn schema(&self) -> SchemaRef {
        Arc::clone(&self.schema)
    }

    fn table_type(&self) -> TableType {
        TableType::Base
    }

    async fn scan(
        &self,
        _state: &dyn Session,
        projection: Option<&Vec<usize>>,
        filters: &[Expr],
        limit: Option<usize>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        if !filters.is_empty() {
            return Err(DataFusionError::Internal(
                "datum-sql committable scan received pushed filters despite reporting them unsupported"
                    .into(),
            ));
        }
        Ok(Arc::new(DatumCommittableScanExec::try_new(
            Arc::clone(&self.schema),
            Arc::clone(&self.source_factory),
            projection.cloned(),
            limit,
            self.event_time.clone(),
        )?))
    }

    fn supports_filters_pushdown(
        &self,
        filters: &[&Expr],
    ) -> Result<Vec<TableProviderFilterPushDown>> {
        Ok(vec![
            TableProviderFilterPushDown::Unsupported;
            filters.len()
        ])
    }
}

#[derive(Clone)]
struct DatumCommittableScanExec {
    input_schema: SchemaRef,
    schema: SchemaRef,
    source_factory: CommittableSourceFactory,
    projection: Option<Vec<usize>>,
    limit: Option<usize>,
    event_time: Option<EventTimeConfig>,
    cache: Arc<PlanProperties>,
}

impl DatumCommittableScanExec {
    fn try_new(
        input_schema: SchemaRef,
        source_factory: CommittableSourceFactory,
        projection: Option<Vec<usize>>,
        limit: Option<usize>,
        event_time: Option<EventTimeConfig>,
    ) -> Result<Self> {
        let schema = project_schema(&input_schema, projection.as_ref())?;
        if let Some(event_time) = &event_time {
            event_time.resolve(input_schema.as_ref())?;
        }
        let cache = Arc::new(DatumScanExec::compute_properties(Arc::clone(&schema)));
        Ok(Self {
            input_schema,
            schema,
            source_factory,
            projection,
            limit,
            event_time,
            cache,
        })
    }

    fn source(&self) -> Source<RecordBatch> {
        self.committable_source().map(|batch| batch.batch().clone())
    }

    fn committable_source(&self) -> Source<CommittableRecordBatch> {
        let mut source = (self.source_factory)();
        if let Some(projection) = self.projection.clone() {
            source = source.try_map(move |batch| {
                batch
                    .try_map_batch(|record_batch| {
                        record_batch
                            .project(&projection)
                            .map_err(DataFusionError::from)
                    })
                    .map_err(stream_error)
            });
        }
        if let Some(limit) = self.limit {
            source = apply_committable_row_window(source, 0, Some(limit));
        }
        source
    }

    fn event_source(&self) -> Result<Source<SqlEvent<RecordBatch>>> {
        let raw_source = (self.source_factory)();
        let source = match &self.event_time {
            Some(event_time) => {
                let resolved = event_time.resolve(self.input_schema.as_ref())?;
                time::assign_resolved_event_time_watermarks(raw_source, resolved)
            }
            None => data_events(raw_source),
        };
        let mut source = map_sql_event_data(source, |batch: CommittableRecordBatch| {
            Ok(batch.batch().clone())
        });
        if let Some(projection) = self.projection.clone() {
            source = map_sql_event_data(source, move |batch| {
                batch.project(&projection).map_err(stream_error)
            });
        }
        if let Some(limit) = self.limit {
            source = apply_event_row_window(source, 0, Some(limit));
        }
        Ok(source)
    }
}

impl fmt::Debug for DatumCommittableScanExec {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("DatumCommittableScanExec")
            .field("schema", &self.schema)
            .field("projection", &self.projection)
            .field("limit", &self.limit)
            .field("event_time", &self.event_time)
            .finish_non_exhaustive()
    }
}

impl DisplayAs for DatumCommittableScanExec {
    fn fmt_as(&self, t: DisplayFormatType, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match t {
            DisplayFormatType::Default | DisplayFormatType::Verbose => write!(
                f,
                "DatumCommittableScanExec: projection={:?}, limit={:?}",
                self.projection, self.limit
            ),
            DisplayFormatType::TreeRender => write!(f, "DatumCommittableScanExec"),
        }
    }
}

impl ExecutionPlan for DatumCommittableScanExec {
    fn name(&self) -> &'static str {
        "DatumCommittableScanExec"
    }

    fn properties(&self) -> &Arc<PlanProperties> {
        &self.cache
    }

    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
        vec![]
    }

    fn with_new_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        if children.is_empty() {
            Ok(self)
        } else {
            Err(DataFusionError::Internal(
                "DatumCommittableScanExec does not accept child plans".into(),
            ))
        }
    }

    fn execute(
        &self,
        partition: usize,
        _context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        assert_or_internal_err!(
            partition == 0,
            "DatumCommittableScanExec invalid partition {partition}; only partition 0 exists"
        );
        let batches = self.source().run_collect().map_err(datafusion_error)?;
        Ok(Box::pin(MemoryStream::try_new(
            batches,
            Arc::clone(&self.schema),
            None,
        )?))
    }

    fn partition_statistics(&self, partition: Option<usize>) -> Result<Arc<Statistics>> {
        if let Some(partition) = partition {
            assert_or_internal_err!(
                partition == 0,
                "DatumCommittableScanExec invalid partition {partition}; only partition 0 exists"
            );
        }

        let mut stats = Statistics::default();
        for _ in self.schema.fields() {
            stats = stats.add_column_statistics(ColumnStatistics {
                null_count: Precision::Absent,
                distinct_count: Precision::Absent,
                min_value: Precision::<ScalarValue>::Absent,
                max_value: Precision::<ScalarValue>::Absent,
                sum_value: Precision::<ScalarValue>::Absent,
                byte_size: Precision::Absent,
            });
        }
        Ok(Arc::new(stats))
    }
}

#[async_trait]
impl TableProvider for DatumChangelogTable {
    fn schema(&self) -> SchemaRef {
        Arc::clone(&self.schema)
    }

    fn table_type(&self) -> TableType {
        TableType::Base
    }

    async fn scan(
        &self,
        _state: &dyn Session,
        projection: Option<&Vec<usize>>,
        filters: &[Expr],
        limit: Option<usize>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        if !filters.is_empty() {
            return Err(DataFusionError::Internal(
                "datum-sql changelog scan received pushed filters despite reporting them unsupported"
                    .into(),
            ));
        }
        Ok(Arc::new(DatumChangelogScanExec::try_new(
            Arc::clone(&self.schema),
            Arc::clone(&self.source_factory),
            projection.cloned(),
            limit,
            self.event_time.clone(),
        )?))
    }

    fn supports_filters_pushdown(
        &self,
        filters: &[&Expr],
    ) -> Result<Vec<TableProviderFilterPushDown>> {
        Ok(vec![
            TableProviderFilterPushDown::Unsupported;
            filters.len()
        ])
    }
}

#[derive(Clone)]
struct DatumScanExec {
    input_schema: SchemaRef,
    schema: SchemaRef,
    source_factory: SourceFactory,
    projection: Option<Vec<usize>>,
    limit: Option<usize>,
    event_time: Option<EventTimeConfig>,
    cache: Arc<PlanProperties>,
}

impl DatumScanExec {
    fn try_new(
        input_schema: SchemaRef,
        source_factory: SourceFactory,
        projection: Option<Vec<usize>>,
        limit: Option<usize>,
        event_time: Option<EventTimeConfig>,
    ) -> Result<Self> {
        let schema = project_schema(&input_schema, projection.as_ref())?;
        if let Some(event_time) = &event_time {
            event_time.resolve(input_schema.as_ref())?;
        }
        let cache = Arc::new(Self::compute_properties(Arc::clone(&schema)));
        Ok(Self {
            input_schema,
            schema,
            source_factory,
            projection,
            limit,
            event_time,
            cache,
        })
    }

    fn compute_properties(schema: SchemaRef) -> PlanProperties {
        PlanProperties::new(
            EquivalenceProperties::new(schema),
            Partitioning::UnknownPartitioning(1),
            EmissionType::Incremental,
            Boundedness::Bounded,
        )
        .with_scheduling_type(SchedulingType::Cooperative)
    }

    fn source(&self) -> Source<RecordBatch> {
        let mut source = (self.source_factory)();
        if let Some(projection) = self.projection.clone() {
            source = source.try_map(move |batch| batch.project(&projection).map_err(stream_error));
        }
        if let Some(limit) = self.limit {
            source = apply_row_window(source, 0, Some(limit));
        }
        source
    }

    fn event_source(&self) -> Result<Source<SqlEvent<RecordBatch>>> {
        let raw_source = (self.source_factory)();
        let mut source = match &self.event_time {
            Some(event_time) => {
                let resolved = event_time.resolve(self.input_schema.as_ref())?;
                time::assign_resolved_event_time_watermarks(raw_source, resolved)
            }
            None => data_events(raw_source),
        };
        if let Some(projection) = self.projection.clone() {
            source = map_sql_event_data(source, move |batch| {
                batch.project(&projection).map_err(stream_error)
            });
        }
        if let Some(limit) = self.limit {
            source = apply_event_row_window(source, 0, Some(limit));
        }
        Ok(source)
    }
}

#[derive(Clone)]
struct DatumChangelogScanExec {
    input_schema: SchemaRef,
    schema: SchemaRef,
    source_factory: ChangelogSourceFactory,
    projection: Option<Vec<usize>>,
    limit: Option<usize>,
    event_time: Option<EventTimeConfig>,
    cache: Arc<PlanProperties>,
}

impl DatumChangelogScanExec {
    fn try_new(
        input_schema: SchemaRef,
        source_factory: ChangelogSourceFactory,
        projection: Option<Vec<usize>>,
        limit: Option<usize>,
        event_time: Option<EventTimeConfig>,
    ) -> Result<Self> {
        let schema = project_schema(&input_schema, projection.as_ref())?;
        if let Some(event_time) = &event_time {
            event_time.resolve(input_schema.as_ref())?;
        }
        let cache = Arc::new(DatumScanExec::compute_properties(Arc::clone(&schema)));
        Ok(Self {
            input_schema,
            schema,
            source_factory,
            projection,
            limit,
            event_time,
            cache,
        })
    }

    fn source(&self) -> Source<ChangelogBatch> {
        let mut source = (self.source_factory)();
        if let Some(projection) = self.projection.clone() {
            source = source.try_map(move |changes| {
                project_changelog_batch(&changes, &projection).map_err(stream_error)
            });
        }
        if let Some(limit) = self.limit {
            source = apply_changelog_row_window(source, 0, Some(limit));
        }
        source
    }

    fn event_source(&self) -> Result<Source<SqlEvent<ChangelogBatch>>> {
        let raw_source = (self.source_factory)();
        let mut source = match &self.event_time {
            Some(event_time) => {
                let resolved = event_time.resolve(self.input_schema.as_ref())?;
                time::assign_resolved_event_time_watermarks(raw_source, resolved)
            }
            None => data_events(raw_source),
        };
        if let Some(projection) = self.projection.clone() {
            source = map_sql_event_data(source, move |changes| {
                project_changelog_batch(&changes, &projection).map_err(stream_error)
            });
        }
        if let Some(limit) = self.limit {
            source = apply_changelog_event_row_window(source, 0, Some(limit));
        }
        Ok(source)
    }
}

impl fmt::Debug for DatumChangelogScanExec {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("DatumChangelogScanExec")
            .field("schema", &self.schema)
            .field("projection", &self.projection)
            .field("limit", &self.limit)
            .field("event_time", &self.event_time)
            .finish_non_exhaustive()
    }
}

impl DisplayAs for DatumChangelogScanExec {
    fn fmt_as(&self, t: DisplayFormatType, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match t {
            DisplayFormatType::Default | DisplayFormatType::Verbose => write!(
                f,
                "DatumChangelogScanExec: projection={:?}, limit={:?}",
                self.projection, self.limit
            ),
            DisplayFormatType::TreeRender => write!(f, "DatumChangelogScanExec"),
        }
    }
}

impl ExecutionPlan for DatumChangelogScanExec {
    fn name(&self) -> &'static str {
        "DatumChangelogScanExec"
    }

    fn properties(&self) -> &Arc<PlanProperties> {
        &self.cache
    }

    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
        vec![]
    }

    fn with_new_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        if children.is_empty() {
            Ok(self)
        } else {
            Err(DataFusionError::Internal(
                "DatumChangelogScanExec does not accept child plans".into(),
            ))
        }
    }

    fn execute(
        &self,
        _partition: usize,
        _context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        Err(updating_scan_error())
    }

    fn partition_statistics(&self, partition: Option<usize>) -> Result<Arc<Statistics>> {
        if let Some(partition) = partition {
            assert_or_internal_err!(
                partition == 0,
                "DatumChangelogScanExec invalid partition {partition}; only partition 0 exists"
            );
        }

        let mut stats = Statistics::default();
        for _ in self.schema.fields() {
            stats = stats.add_column_statistics(ColumnStatistics {
                null_count: Precision::Absent,
                distinct_count: Precision::Absent,
                min_value: Precision::<ScalarValue>::Absent,
                max_value: Precision::<ScalarValue>::Absent,
                sum_value: Precision::<ScalarValue>::Absent,
                byte_size: Precision::Absent,
            });
        }
        Ok(Arc::new(stats))
    }
}

impl fmt::Debug for DatumScanExec {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("DatumScanExec")
            .field("schema", &self.schema)
            .field("projection", &self.projection)
            .field("limit", &self.limit)
            .field("event_time", &self.event_time)
            .finish_non_exhaustive()
    }
}

impl DisplayAs for DatumScanExec {
    fn fmt_as(&self, t: DisplayFormatType, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match t {
            DisplayFormatType::Default | DisplayFormatType::Verbose => write!(
                f,
                "DatumScanExec: projection={:?}, limit={:?}",
                self.projection, self.limit
            ),
            DisplayFormatType::TreeRender => write!(f, "DatumScanExec"),
        }
    }
}

impl ExecutionPlan for DatumScanExec {
    fn name(&self) -> &'static str {
        "DatumScanExec"
    }

    fn properties(&self) -> &Arc<PlanProperties> {
        &self.cache
    }

    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
        vec![]
    }

    fn with_new_children(
        self: Arc<Self>,
        children: Vec<Arc<dyn ExecutionPlan>>,
    ) -> Result<Arc<dyn ExecutionPlan>> {
        if children.is_empty() {
            Ok(self)
        } else {
            Err(DataFusionError::Internal(
                "DatumScanExec does not accept child plans".into(),
            ))
        }
    }

    fn execute(
        &self,
        partition: usize,
        _context: Arc<TaskContext>,
    ) -> Result<SendableRecordBatchStream> {
        assert_or_internal_err!(
            partition == 0,
            "DatumScanExec invalid partition {partition}; only partition 0 exists"
        );
        let batches = self.source().run_collect().map_err(datafusion_error)?;
        Ok(Box::pin(MemoryStream::try_new(
            batches,
            Arc::clone(&self.schema),
            None,
        )?))
    }

    fn partition_statistics(&self, partition: Option<usize>) -> Result<Arc<Statistics>> {
        if let Some(partition) = partition {
            assert_or_internal_err!(
                partition == 0,
                "DatumScanExec invalid partition {partition}; only partition 0 exists"
            );
        }

        let mut stats = Statistics::default();
        for _ in self.schema.fields() {
            stats = stats.add_column_statistics(ColumnStatistics {
                null_count: Precision::Absent,
                distinct_count: Precision::Absent,
                min_value: Precision::<ScalarValue>::Absent,
                max_value: Precision::<ScalarValue>::Absent,
                sum_value: Precision::<ScalarValue>::Absent,
                byte_size: Precision::Absent,
            });
        }
        Ok(Arc::new(stats))
    }
}

fn lower_physical_plan(plan: &dyn ExecutionPlan) -> Result<Source<RecordBatch>> {
    if let Some(scan) = plan.downcast_ref::<DatumScanExec>() {
        return Ok(scan.source());
    }

    if let Some(scan) = plan.downcast_ref::<DatumCommittableScanExec>() {
        return Ok(scan.source());
    }

    if plan.downcast_ref::<DatumChangelogScanExec>().is_some() {
        return Err(updating_scan_error());
    }

    if let Some(projection) = plan.downcast_ref::<ProjectionExec>() {
        let input = lower_physical_plan(projection.input().as_ref())?;
        let exprs = projection.expr().to_vec();
        let schema = projection.schema();
        return Ok(input.try_map(move |batch| {
            project_batch(&batch, &exprs, Arc::clone(&schema)).map_err(stream_error)
        }));
    }

    if let Some(filter) = plan.downcast_ref::<FilterExec>() {
        let input = lower_physical_plan(filter.input().as_ref())?;
        let predicate = Arc::clone(filter.predicate());
        let projection = filter.projection().as_ref().map(|p| p.as_ref().to_vec());
        return Ok(input.try_map(move |batch| {
            filter_batch(&batch, predicate.as_ref(), projection.as_deref()).map_err(stream_error)
        }));
    }

    #[allow(deprecated)]
    if let Some(coalesce) = plan.downcast_ref::<CoalesceBatchesExec>() {
        return lower_physical_plan(coalesce.input().as_ref());
    }

    if let Some(coalesce) = plan.downcast_ref::<CoalescePartitionsExec>() {
        return lower_physical_plan(coalesce.input().as_ref());
    }

    if let Some(limit) = plan.downcast_ref::<GlobalLimitExec>() {
        let input = lower_physical_plan(limit.input().as_ref())?;
        return Ok(apply_row_window(input, limit.skip(), limit.fetch()));
    }

    if let Some(limit) = plan.downcast_ref::<LocalLimitExec>() {
        let input = lower_physical_plan(limit.input().as_ref())?;
        return Ok(apply_row_window(input, 0, Some(limit.fetch())));
    }

    Err(DataFusionError::NotImplemented(format!(
        "datum-sql does not lower DataFusion physical node {} yet",
        plan.name()
    )))
}

fn lower_physical_plan_streaming_with_metrics_and_config(
    plan: &dyn ExecutionPlan,
    metrics: &SqlExecutionMetrics,
    join_config: &StreamingJoinConfig,
) -> Result<Source<SqlEvent<RecordBatch>>> {
    if let Some(scan) = plan.downcast_ref::<DatumScanExec>() {
        return scan.event_source();
    }

    if let Some(scan) = plan.downcast_ref::<DatumCommittableScanExec>() {
        return scan.event_source();
    }

    if let Some(projection) = plan.downcast_ref::<ProjectionExec>() {
        let input = lower_physical_plan_streaming_with_metrics_and_config(
            projection.input().as_ref(),
            metrics,
            join_config,
        )?;
        let exprs = projection.expr().to_vec();
        let schema = projection.schema();
        return Ok(map_sql_event_data(input, move |batch| {
            project_batch(&batch, &exprs, Arc::clone(&schema)).map_err(stream_error)
        }));
    }

    if let Some(filter) = plan.downcast_ref::<FilterExec>() {
        let input = lower_physical_plan_streaming_with_metrics_and_config(
            filter.input().as_ref(),
            metrics,
            join_config,
        )?;
        let predicate = Arc::clone(filter.predicate());
        let projection = filter.projection().as_ref().map(|p| p.as_ref().to_vec());
        return Ok(map_sql_event_data(input, move |batch| {
            filter_batch(&batch, predicate.as_ref(), projection.as_deref()).map_err(stream_error)
        }));
    }

    if let Some(join) = plan.downcast_ref::<HashJoinExec>() {
        if plan_contains_changelog_scan(join.left().as_ref())
            || plan_contains_changelog_scan(join.right().as_ref())
        {
            return Err(DataFusionError::NotImplemented(
                "datum-sql streaming joins do not support changelog inputs yet; append-only inputs are required"
                    .into(),
            ));
        }
        let left = lower_physical_plan_streaming_with_metrics_and_config(
            join.left().as_ref(),
            metrics,
            join_config,
        )?;
        let right = lower_physical_plan_streaming_with_metrics_and_config(
            join.right().as_ref(),
            metrics,
            join_config,
        )?;
        return join::streaming_hash_join_source(
            left,
            right,
            join,
            join_config,
            metrics.join_metrics(),
        );
    }

    if let Some(aggregate) = plan.downcast_ref::<AggregateExec>() {
        if plan_contains_changelog_scan(aggregate.input().as_ref()) {
            let input = lower_changelog_physical_plan_streaming(aggregate.input().as_ref())?;
            return window::windowed_changelog_aggregate_source(
                input,
                aggregate,
                metrics.window_metrics(),
            );
        }
        let input = lower_physical_plan_streaming_with_metrics_and_config(
            aggregate.input().as_ref(),
            metrics,
            join_config,
        )?;
        return window::windowed_aggregate_source(input, aggregate, metrics.window_metrics());
    }

    #[allow(deprecated)]
    if let Some(coalesce) = plan.downcast_ref::<CoalesceBatchesExec>() {
        return lower_physical_plan_streaming_with_metrics_and_config(
            coalesce.input().as_ref(),
            metrics,
            join_config,
        );
    }

    if let Some(coalesce) = plan.downcast_ref::<CoalescePartitionsExec>() {
        return lower_physical_plan_streaming_with_metrics_and_config(
            coalesce.input().as_ref(),
            metrics,
            join_config,
        );
    }

    if let Some(limit) = plan.downcast_ref::<GlobalLimitExec>() {
        let input = lower_physical_plan_streaming_with_metrics_and_config(
            limit.input().as_ref(),
            metrics,
            join_config,
        )?;
        return Ok(apply_event_row_window(input, limit.skip(), limit.fetch()));
    }

    if let Some(limit) = plan.downcast_ref::<LocalLimitExec>() {
        let input = lower_physical_plan_streaming_with_metrics_and_config(
            limit.input().as_ref(),
            metrics,
            join_config,
        )?;
        return Ok(apply_event_row_window(input, 0, Some(limit.fetch())));
    }

    Err(DataFusionError::NotImplemented(format!(
        "datum-sql does not lower DataFusion physical node {} yet",
        plan.name()
    )))
}

fn lower_changelog_physical_plan_streaming(
    plan: &dyn ExecutionPlan,
) -> Result<Source<SqlEvent<ChangelogBatch>>> {
    if let Some(scan) = plan.downcast_ref::<DatumChangelogScanExec>() {
        return scan.event_source();
    }

    if let Some(projection) = plan.downcast_ref::<ProjectionExec>() {
        let input = lower_changelog_physical_plan_streaming(projection.input().as_ref())?;
        let exprs = projection.expr().to_vec();
        let schema = projection.schema();
        return Ok(map_sql_event_data(input, move |changes| {
            let batch = project_batch(changes.batch(), &exprs, Arc::clone(&schema))
                .map_err(stream_error)?;
            ChangelogBatch::try_new(changes.ops().to_vec(), batch).map_err(stream_error)
        }));
    }

    if let Some(filter) = plan.downcast_ref::<FilterExec>() {
        let input = lower_changelog_physical_plan_streaming(filter.input().as_ref())?;
        let predicate = Arc::clone(filter.predicate());
        let projection = filter.projection().as_ref().map(|p| p.as_ref().to_vec());
        return Ok(map_sql_event_data(input, move |changes| {
            filter_changelog_batch(&changes, predicate.as_ref(), projection.as_deref())
                .map_err(stream_error)
        }));
    }

    #[allow(deprecated)]
    if let Some(coalesce) = plan.downcast_ref::<CoalesceBatchesExec>() {
        return lower_changelog_physical_plan_streaming(coalesce.input().as_ref());
    }

    if let Some(coalesce) = plan.downcast_ref::<CoalescePartitionsExec>() {
        return lower_changelog_physical_plan_streaming(coalesce.input().as_ref());
    }

    if let Some(limit) = plan.downcast_ref::<GlobalLimitExec>() {
        let input = lower_changelog_physical_plan_streaming(limit.input().as_ref())?;
        return Ok(apply_changelog_event_row_window(
            input,
            limit.skip(),
            limit.fetch(),
        ));
    }

    if let Some(limit) = plan.downcast_ref::<LocalLimitExec>() {
        let input = lower_changelog_physical_plan_streaming(limit.input().as_ref())?;
        return Ok(apply_changelog_event_row_window(
            input,
            0,
            Some(limit.fetch()),
        ));
    }

    Err(DataFusionError::NotImplemented(format!(
        "datum-sql does not lower updating DataFusion physical node {} yet",
        plan.name()
    )))
}

fn lower_changelog_physical_plan(plan: &dyn ExecutionPlan) -> Result<Source<ChangelogBatch>> {
    if let Some(scan) = plan.downcast_ref::<DatumChangelogScanExec>() {
        return Ok(scan.source());
    }

    if let Some(projection) = plan.downcast_ref::<ProjectionExec>() {
        let input = lower_changelog_physical_plan(projection.input().as_ref())?;
        let exprs = projection.expr().to_vec();
        let schema = projection.schema();
        return Ok(input.try_map(move |changes| {
            let batch = project_batch(changes.batch(), &exprs, Arc::clone(&schema))
                .map_err(stream_error)?;
            ChangelogBatch::try_new(changes.ops().to_vec(), batch).map_err(stream_error)
        }));
    }

    if let Some(filter) = plan.downcast_ref::<FilterExec>() {
        let input = lower_changelog_physical_plan(filter.input().as_ref())?;
        let predicate = Arc::clone(filter.predicate());
        let projection = filter.projection().as_ref().map(|p| p.as_ref().to_vec());
        return Ok(input.try_map(move |changes| {
            filter_changelog_batch(&changes, predicate.as_ref(), projection.as_deref())
                .map_err(stream_error)
        }));
    }

    #[allow(deprecated)]
    if let Some(coalesce) = plan.downcast_ref::<CoalesceBatchesExec>() {
        return lower_changelog_physical_plan(coalesce.input().as_ref());
    }

    if let Some(coalesce) = plan.downcast_ref::<CoalescePartitionsExec>() {
        return lower_changelog_physical_plan(coalesce.input().as_ref());
    }

    if let Some(limit) = plan.downcast_ref::<GlobalLimitExec>() {
        let input = lower_changelog_physical_plan(limit.input().as_ref())?;
        return Ok(apply_changelog_row_window(
            input,
            limit.skip(),
            limit.fetch(),
        ));
    }

    if let Some(limit) = plan.downcast_ref::<LocalLimitExec>() {
        let input = lower_changelog_physical_plan(limit.input().as_ref())?;
        return Ok(apply_changelog_row_window(input, 0, Some(limit.fetch())));
    }

    Err(DataFusionError::NotImplemented(format!(
        "datum-sql does not lower updating DataFusion physical node {} yet",
        plan.name()
    )))
}

fn lower_append_sink_plan(plan: &dyn ExecutionPlan) -> Result<AppendSinkSource> {
    if !plan_contains_committable_scan(plan) {
        return lower_physical_plan(plan).map(AppendSinkSource::Plain);
    }

    if let Some(scan) = plan.downcast_ref::<DatumCommittableScanExec>() {
        return Ok(AppendSinkSource::Committable(scan.committable_source()));
    }

    if let Some(projection) = plan.downcast_ref::<ProjectionExec>() {
        let input = lower_append_sink_plan(projection.input().as_ref())?.into_committable();
        let exprs = projection.expr().to_vec();
        let schema = projection.schema();
        return Ok(AppendSinkSource::Committable(input.try_map(move |batch| {
            batch
                .try_map_batch(|record_batch| {
                    project_batch(record_batch, &exprs, Arc::clone(&schema))
                })
                .map_err(stream_error)
        })));
    }

    if let Some(filter) = plan.downcast_ref::<FilterExec>() {
        let input = lower_append_sink_plan(filter.input().as_ref())?.into_committable();
        let predicate = Arc::clone(filter.predicate());
        let projection = filter.projection().as_ref().map(|p| p.as_ref().to_vec());
        return Ok(AppendSinkSource::Committable(input.try_map(move |batch| {
            batch
                .try_map_batch(|record_batch| {
                    filter_batch(record_batch, predicate.as_ref(), projection.as_deref())
                })
                .map_err(stream_error)
        })));
    }

    #[allow(deprecated)]
    if let Some(coalesce) = plan.downcast_ref::<CoalesceBatchesExec>() {
        return lower_append_sink_plan(coalesce.input().as_ref());
    }

    if let Some(coalesce) = plan.downcast_ref::<CoalescePartitionsExec>() {
        return lower_append_sink_plan(coalesce.input().as_ref());
    }

    if let Some(limit) = plan.downcast_ref::<GlobalLimitExec>() {
        let input = lower_append_sink_plan(limit.input().as_ref())?.into_committable();
        return Ok(AppendSinkSource::Committable(apply_committable_row_window(
            input,
            limit.skip(),
            limit.fetch(),
        )));
    }

    if let Some(limit) = plan.downcast_ref::<LocalLimitExec>() {
        let input = lower_append_sink_plan(limit.input().as_ref())?.into_committable();
        return Ok(AppendSinkSource::Committable(apply_committable_row_window(
            input,
            0,
            Some(limit.fetch()),
        )));
    }

    Err(DataFusionError::NotImplemented(format!(
        "datum-sql does not lower committable DataFusion physical node {} yet",
        plan.name()
    )))
}

fn project_batch(
    batch: &RecordBatch,
    exprs: &[ProjectionExpr],
    schema: SchemaRef,
) -> Result<RecordBatch> {
    let columns = exprs
        .iter()
        .map(|expr| expr.expr.evaluate(batch)?.into_array(batch.num_rows()))
        .collect::<Result<Vec<_>>>()?;
    RecordBatch::try_new(schema, columns).map_err(DataFusionError::from)
}

fn filter_batch(
    batch: &RecordBatch,
    predicate: &dyn PhysicalExpr,
    projection: Option<&[usize]>,
) -> Result<RecordBatch> {
    let array = predicate.evaluate(batch)?.into_array(batch.num_rows())?;
    let filter = as_boolean_array(&array)?;
    match projection {
        Some(projection) => {
            let projected = batch.project(projection)?;
            filter_record_batch(&projected, filter).map_err(DataFusionError::from)
        }
        None => filter_record_batch(batch, filter).map_err(DataFusionError::from),
    }
}

fn project_changelog_batch(
    changes: &ChangelogBatch,
    projection: &[usize],
) -> Result<ChangelogBatch> {
    let batch = changes.batch().project(projection)?;
    ChangelogBatch::try_new(changes.ops().to_vec(), batch)
}

fn filter_changelog_batch(
    changes: &ChangelogBatch,
    predicate: &dyn PhysicalExpr,
    projection: Option<&[usize]>,
) -> Result<ChangelogBatch> {
    let array = predicate
        .evaluate(changes.batch())?
        .into_array(changes.num_rows())?;
    let filter = as_boolean_array(&array)?;
    let payload = match projection {
        Some(projection) => changes.batch().project(projection)?,
        None => changes.batch().clone(),
    };
    let batch = filter_record_batch(&payload, filter)?;
    let ops = (0..changes.num_rows())
        .filter(|row| filter.is_valid(*row) && filter.value(*row))
        .map(|row| changes.ops()[row])
        .collect::<Vec<_>>();
    ChangelogBatch::try_new(ops, batch)
}

fn plan_contains_changelog_scan(plan: &dyn ExecutionPlan) -> bool {
    plan.downcast_ref::<DatumChangelogScanExec>().is_some()
        || plan
            .children()
            .into_iter()
            .any(|child| plan_contains_changelog_scan(child.as_ref()))
}

fn plan_contains_committable_scan(plan: &dyn ExecutionPlan) -> bool {
    plan.downcast_ref::<DatumCommittableScanExec>().is_some()
        || plan
            .children()
            .into_iter()
            .any(|child| plan_contains_committable_scan(child.as_ref()))
}

#[derive(Clone)]
struct RowWindow {
    skip: usize,
    fetch: Option<usize>,
}

fn apply_row_window(
    source: Source<RecordBatch>,
    skip: usize,
    fetch: Option<usize>,
) -> Source<RecordBatch> {
    source.stateful_map_concat(RowWindow { skip, fetch }, |window, batch| {
        let mut offset = 0;
        if window.skip > 0 {
            let skipped = window.skip.min(batch.num_rows());
            window.skip -= skipped;
            offset = skipped;
        }

        if offset == batch.num_rows() {
            return Vec::new();
        }

        let mut len = batch.num_rows() - offset;
        if let Some(remaining) = &mut window.fetch {
            if *remaining == 0 {
                return Vec::new();
            }
            len = len.min(*remaining);
            *remaining -= len;
        }

        vec![batch.slice(offset, len)]
    })
}

fn apply_event_row_window(
    source: Source<SqlEvent<RecordBatch>>,
    skip: usize,
    fetch: Option<usize>,
) -> Source<SqlEvent<RecordBatch>> {
    source.stateful_map_concat(RowWindow { skip, fetch }, |window, event| match event {
        SqlEvent::Data(batch) => {
            let mut offset = 0;
            if window.skip > 0 {
                let skipped = window.skip.min(batch.num_rows());
                window.skip -= skipped;
                offset = skipped;
            }

            if offset == batch.num_rows() {
                return Vec::new();
            }

            let mut len = batch.num_rows() - offset;
            if let Some(remaining) = &mut window.fetch {
                if *remaining == 0 {
                    return Vec::new();
                }
                len = len.min(*remaining);
                *remaining -= len;
            }

            vec![SqlEvent::Data(batch.slice(offset, len))]
        }
        SqlEvent::Watermark(watermark) => vec![SqlEvent::Watermark(watermark)],
        SqlEvent::Barrier(barrier) => vec![SqlEvent::Barrier(barrier)],
    })
}

fn apply_changelog_row_window(
    source: Source<ChangelogBatch>,
    skip: usize,
    fetch: Option<usize>,
) -> Source<ChangelogBatch> {
    source.try_stateful_map_concat(RowWindow { skip, fetch }, |window, changes| {
        let mut offset = 0;
        if window.skip > 0 {
            let skipped = window.skip.min(changes.num_rows());
            window.skip -= skipped;
            offset = skipped;
        }

        if offset == changes.num_rows() {
            return Ok(Vec::new());
        }

        let mut len = changes.num_rows() - offset;
        if let Some(remaining) = &mut window.fetch {
            if *remaining == 0 {
                return Ok(Vec::new());
            }
            len = len.min(*remaining);
            *remaining -= len;
        }

        let ops = changes.ops()[offset..offset + len].to_vec();
        let batch = changes.batch().slice(offset, len);
        let changes = ChangelogBatch::try_new(ops, batch).map_err(stream_error)?;
        Ok(vec![changes])
    })
}

fn apply_committable_row_window(
    source: Source<CommittableRecordBatch>,
    skip: usize,
    fetch: Option<usize>,
) -> Source<CommittableRecordBatch> {
    source.try_stateful_map_concat(RowWindow { skip, fetch }, |window, batch| {
        apply_committable_row_window_to_batch(window, batch)
    })
}

fn apply_committable_row_window_to_batch(
    window: &mut RowWindow,
    batch: CommittableRecordBatch,
) -> StreamResult<Vec<CommittableRecordBatch>> {
    let mut offset = 0;
    if window.skip > 0 {
        let skipped = window.skip.min(batch.batch().num_rows());
        window.skip -= skipped;
        offset = skipped;
    }

    let mut len = batch.batch().num_rows().saturating_sub(offset);
    if let Some(remaining) = &mut window.fetch {
        if *remaining == 0 {
            return empty_committable_batch(batch).map(|batch| vec![batch]);
        }
        len = len.min(*remaining);
        *remaining -= len;
    }

    if len == 0 {
        return empty_committable_batch(batch).map(|batch| vec![batch]);
    }

    batch
        .try_map_batch(|record_batch| Ok(record_batch.slice(offset, len)))
        .map(|batch| vec![batch])
        .map_err(stream_error)
}

fn empty_committable_batch(batch: CommittableRecordBatch) -> StreamResult<CommittableRecordBatch> {
    batch
        .try_map_batch(|record_batch| Ok(record_batch.slice(0, 0)))
        .map_err(stream_error)
}

fn apply_changelog_event_row_window(
    source: Source<SqlEvent<ChangelogBatch>>,
    skip: usize,
    fetch: Option<usize>,
) -> Source<SqlEvent<ChangelogBatch>> {
    source.try_stateful_map_concat(RowWindow { skip, fetch }, |window, event| match event {
        SqlEvent::Data(changes) => apply_changelog_row_window_to_batch(window, changes)
            .map(|batches| batches.into_iter().map(SqlEvent::Data).collect::<Vec<_>>()),
        SqlEvent::Watermark(watermark) => Ok(vec![SqlEvent::Watermark(watermark)]),
        SqlEvent::Barrier(barrier) => Ok(vec![SqlEvent::Barrier(barrier)]),
    })
}

fn apply_changelog_row_window_to_batch(
    window: &mut RowWindow,
    changes: ChangelogBatch,
) -> StreamResult<Vec<ChangelogBatch>> {
    let mut offset = 0;
    if window.skip > 0 {
        let skipped = window.skip.min(changes.num_rows());
        window.skip -= skipped;
        offset = skipped;
    }

    if offset == changes.num_rows() {
        return Ok(Vec::new());
    }

    let mut len = changes.num_rows() - offset;
    if let Some(remaining) = &mut window.fetch {
        if *remaining == 0 {
            return Ok(Vec::new());
        }
        len = len.min(*remaining);
        *remaining -= len;
    }

    let ops = changes.ops()[offset..offset + len].to_vec();
    let batch = changes.batch().slice(offset, len);
    let changes = ChangelogBatch::try_new(ops, batch).map_err(stream_error)?;
    Ok(vec![changes])
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct ParsedInsertInto {
    sink_name: String,
    select_sql: String,
}

fn looks_like_insert_into(sql: &str) -> bool {
    sql.trim_start()
        .get(..11)
        .is_some_and(|prefix| prefix.eq_ignore_ascii_case("insert into"))
}

fn parse_insert_into(sql: &str) -> Result<ParsedInsertInto> {
    let mut statements = DFParser::parse_sql(sql)?;
    if statements.len() != 1 {
        return Err(DataFusionError::Plan(format!(
            "datum-sql INSERT INTO sink expects exactly one statement, found {}",
            statements.len()
        )));
    }

    let statement = statements
        .pop_front()
        .expect("statement length was checked above");
    let DfStatement::Statement(statement) = statement else {
        return Err(DataFusionError::Plan(
            "datum-sql INSERT INTO sink supports SQL INSERT statements only".into(),
        ));
    };
    let SqlStatement::Insert(insert) = *statement else {
        return Err(DataFusionError::Plan(
            "datum-sql INSERT INTO sink supports INSERT INTO <sink> SELECT ... only".into(),
        ));
    };

    if !insert.into {
        return Err(DataFusionError::Plan(
            "datum-sql sink SQL requires INSERT INTO <sink> SELECT ...".into(),
        ));
    }
    if !insert.optimizer_hints.is_empty()
        || insert.or.is_some()
        || insert.ignore
        || insert.table_alias.is_some()
        || !insert.columns.is_empty()
        || insert.overwrite
        || !insert.assignments.is_empty()
        || insert.partitioned.is_some()
        || !insert.after_columns.is_empty()
        || insert.has_table_keyword
        || insert.on.is_some()
        || insert.returning.is_some()
        || insert.output.is_some()
        || insert.replace_into
        || insert.priority.is_some()
        || insert.insert_alias.is_some()
        || insert.settings.is_some()
        || insert.format_clause.is_some()
        || insert.multi_table_insert_type.is_some()
        || !insert.multi_table_into_clauses.is_empty()
        || !insert.multi_table_when_clauses.is_empty()
        || insert.multi_table_else_clause.is_some()
    {
        return Err(DataFusionError::Plan(
            "datum-sql sink SQL supports only plain INSERT INTO <sink> SELECT ...".into(),
        ));
    }

    let SqlTableObject::TableName(table_name) = insert.table else {
        return Err(DataFusionError::Plan(
            "datum-sql sink SQL requires a registered sink name after INSERT INTO".into(),
        ));
    };
    let sink_name = object_name_to_registry_key(table_name)?;
    let source = insert.source.ok_or_else(|| {
        DataFusionError::Plan("datum-sql sink SQL requires INSERT INTO <sink> SELECT ...".into())
    })?;

    Ok(ParsedInsertInto {
        sink_name,
        select_sql: source.to_string(),
    })
}

fn object_name_to_registry_key(
    name: datafusion::sql::sqlparser::ast::ObjectName,
) -> Result<String> {
    let mut parts = Vec::with_capacity(name.0.len());
    for part in name.0 {
        let Some(ident) = part.as_ident() else {
            return Err(DataFusionError::Plan(
                "datum-sql sink names must be identifiers".into(),
            ));
        };
        parts.push(ident.value.clone());
    }
    if parts.is_empty() {
        return Err(DataFusionError::Plan(
            "datum-sql sink name cannot be empty".into(),
        ));
    }
    Ok(parts.join("."))
}

fn updating_scan_error() -> DataFusionError {
    DataFusionError::Plan(
        "datum-sql updating tables produce Source<ChangelogBatch>; use changelog-aware lowering or a changelog-aware sink instead of the append-only RecordBatch executor"
            .into(),
    )
}

fn datafusion_error(error: impl fmt::Display) -> DataFusionError {
    DataFusionError::External(Box::new(StreamBridgeError(error.to_string())))
}

fn stream_error(error: impl fmt::Display) -> StreamError {
    StreamError::Failed(error.to_string())
}

#[derive(Debug)]
struct StreamBridgeError(String);

impl fmt::Display for StreamBridgeError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.write_str(&self.0)
    }
}

impl std::error::Error for StreamBridgeError {}