use crate::catalog::{ArrowStream, InputCollectionHandle};
use crate::format::InputBuffer;
use crate::integrated::delta_table::{delta_input_serde_config, register_storage_handlers};
use crate::transport::{InputEndpoint, InputQueue, InputReaderCommand, IntegratedInputEndpoint};
use crate::util::JobQueue;
use crate::{ControllerError, InputConsumer, InputReader, PipelineState};
use anyhow::{Error as AnyError, Result as AnyResult, anyhow, bail};
use arrow::array::BooleanArray;
use chrono::{DateTime, Utc};
use datafusion::catalog::TableProvider;
use datafusion::common::arrow::array::RecordBatch;
use datafusion::datasource::file_format::parquet::ParquetFormat;
use datafusion::datasource::listing::{
    ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl,
};
use datafusion::logical_expr::LogicalPlan;
use datafusion::physical_plan::PhysicalExpr;
use datafusion::physical_planner::{DefaultPhysicalPlanner, PhysicalPlanner};
use datafusion::prelude::SessionConfig;
use dbsp::circuit::tokio::TOKIO;
use deltalake::datafusion::dataframe::DataFrame;
use deltalake::datafusion::execution::context::SQLOptions;
use deltalake::datafusion::prelude::SessionContext;
use deltalake::kernel::Action;
use deltalake::logstore::{self, IORuntime};
use deltalake::table::builder::ensure_table_uri;
use deltalake::{DeltaTable, DeltaTableBuilder, datafusion};
use feldera_adapterlib::format::{ParseError, StagedInputBuffer};
use feldera_adapterlib::metrics::{ConnectorMetrics, ValueType};
use feldera_adapterlib::transport::{InputQueueEntry, Resume, Watermark, parse_resume_info};
use feldera_adapterlib::utils::datafusion::{
    array_to_string, execute_query_collect, execute_singleton_query, timestamp_to_sql_expression,
    validate_sql_expression, validate_timestamp_column,
};
use feldera_storage::tokio::TOKIO_DEDICATED_IO;
use feldera_types::config::FtModel;
use feldera_types::program_schema::Relation;
use feldera_types::transport::delta_table::{DeltaTableReaderConfig, DeltaTableTransactionMode};
use futures_util::StreamExt;
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;
use std::cmp::min;
use std::collections::{BTreeSet, HashMap};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::select;
use tokio::sync::watch::{Receiver, Sender, channel};
use tokio::sync::{Semaphore, mpsc};
use tokio::time::sleep;
use tracing::{debug, info, trace, warn};
use url::Url;

/// Polling interval when following a delta table.
const POLL_INTERVAL: Duration = Duration::from_millis(1000);

/// Calculate exponential backoff delay for retrying delta log reads.
/// Starts at 0.5s, doubles each retry, caps at 32s.
fn calculate_backoff_delay(retry_count: u32) -> Duration {
    let base_delay_ms = 500; // 0.5 seconds
    let max_delay_ms = 32_000; // 32 seconds
    let delay_ms = std::cmp::min(base_delay_ms << retry_count, max_delay_ms);
    Duration::from_millis(delay_ms)
}

/// Default object store timeout. When not explicitly set by the user,
/// we use a large timeout value to avoid this issue:
/// https://github.com/delta-io/delta-rs/issues/2595, which is common for
/// tables with large files.
const DEFAULT_OBJECT_STORE_TIMEOUT: &str = "120s";

const REPORT_ERROR: &str =
    "please report this error to developers (https://github.com/feldera/feldera/issues)";

/// Semaphore used to control the number of concurrent reads to the object store
/// as a workaround for https://github.com/apache/arrow-rs-object-store/issues/14.
/// (see `max_concurrent_readers` in `DeltaTableReaderConfig`).
static DELTA_READER_SEMAPHORE: std::sync::LazyLock<Semaphore> =
    std::sync::LazyLock::new(|| Semaphore::new(6));

/// True if the `max_concurrent_readers` attribute was set by one of the connectors.
/// Used to detect conflicting values of `max_concurrent_readers`.
static MAX_CONCURRENT_READERS_SET: AtomicBool = AtomicBool::new(false);

/// Takes a column name from a DeltaLake schema and returns a qouted string
/// that can be used in datafusion queries like `select "foo""bar" from my_table`.
fn quote_sql_identifier<S: AsRef<str>>(ident: S) -> String {
    format!("\"{}\"", ident.as_ref().replace("\"", "\"\""))
}

/// Integrated input connector that reads from a delta table.
pub struct DeltaTableInputEndpoint {
    endpoint_name: String,
    config: DeltaTableReaderConfig,
    consumer: Box<dyn InputConsumer>,
}

impl DeltaTableInputEndpoint {
    pub fn new(
        endpoint_name: &str,
        config: &DeltaTableReaderConfig,
        consumer: Box<dyn InputConsumer>,
    ) -> Self {
        register_storage_handlers();

        Self {
            endpoint_name: endpoint_name.to_string(),
            config: config.clone(),
            consumer,
        }
    }
}

impl InputEndpoint for DeltaTableInputEndpoint {
    fn fault_tolerance(&self) -> Option<FtModel> {
        Some(FtModel::AtLeastOnce)
    }
}
impl IntegratedInputEndpoint for DeltaTableInputEndpoint {
    fn open(
        self: Box<Self>,
        input_handle: &InputCollectionHandle,
        resume_info: Option<JsonValue>,
    ) -> AnyResult<Box<dyn InputReader>> {
        Ok(Box::new(DeltaTableInputReader::new(
            self.endpoint_name,
            self.config,
            self.consumer,
            input_handle,
            resume_info,
        )?))
    }
}

struct DeltaTableInputReader {
    sender: Sender<PipelineState>,
    inner: Arc<DeltaTableInputEndpointInner>,
}

impl DeltaTableInputReader {
    fn new(
        endpoint_name: String,
        mut config: DeltaTableReaderConfig,
        consumer: Box<dyn InputConsumer>,
        input_handle: &InputCollectionHandle,
        resume_info: Option<JsonValue>,
    ) -> AnyResult<Self> {
        let (sender, receiver) = channel(PipelineState::Paused);
        let receiver_clone = receiver.clone();

        let resume_info = if let Some(resume_info) = resume_info {
            let resume_info = parse_resume_info::<DeltaResumeInfo>(&resume_info)?;
            match &resume_info {
                DeltaResumeInfo { eoi: true, .. } => {
                    info!(
                        "delta_table {endpoint_name}: skipping connector initialization because the connector is already in the end-of-input state"
                    );
                }
                DeltaResumeInfo {
                    version: Some(version),
                    snapshot_timestamp: Some(snapshot_timestamp),
                    ..
                } => {
                    info!(
                        "delta_table {endpoint_name}: resuming to ingest the initial snapshot from table version {version} at timestamp {snapshot_timestamp}"
                    );
                }
                DeltaResumeInfo {
                    version: Some(version),
                    ..
                } => {
                    info!(
                        "delta_table {endpoint_name}: resuming in follow mode from table version {version}"
                    );
                }
                DeltaResumeInfo { version: None, .. } => {
                    info!("delta_table {endpoint_name}: resuming from clean state");
                }
            }

            Some(resume_info)
        } else {
            None
        };

        let eoi = if let Some(resume_info) = &resume_info {
            resume_info.eoi
        } else {
            false
        };

        // Used to communicate the status of connector initialization.
        let (init_status_sender, mut init_status_receiver) =
            mpsc::channel::<Result<(), ControllerError>>(1);

        if config.num_parsers == 0 {
            bail!("invalid 'num_parsers' value: 'num_parsers' must be greater than 0");
        }

        if config.end_version.is_some() && !config.follow() {
            bail!(
                "the 'end_version' property is not valid in '{}' mode; it can only be used when the DeltaLake connector is configured with 'follow', 'snapshot_and_follow', or 'cdc' mode",
                config.mode
            )
        }

        if config.version.is_some()
            && config.end_version.is_some()
            && config.version.unwrap() >= config.end_version.unwrap()
        {
            bail!("'end_version' must be greater than 'version'")
        }

        // If the config specifies max_concurrent_connectors, adjust the number of tokens
        // in the semaphore.
        if let Some(max_concurrent_readers) = config.max_concurrent_readers {
            let max_concurrent_readers = max_concurrent_readers as usize;
            let available_permits = DELTA_READER_SEMAPHORE.available_permits();

            if max_concurrent_readers == 0 {
                bail!(
                    "invalid 'max_concurrent_readers' value: 'max_concurrent_readers' must be greater than 0"
                );
            }

            if MAX_CONCURRENT_READERS_SET.load(Ordering::Acquire)
                && max_concurrent_readers != available_permits
            {
                bail!(
                    "found conflicting values of the `max_concurrent_readers` attribute: this is a global setting that affects all Delta Lake connectors, and not just the connector where it is specified; if multiple connectors specify `max_concurrent_readers`, they must all use the same value."
                );
            }

            MAX_CONCURRENT_READERS_SET.store(true, Ordering::Release);

            info!(
                "delta_table {endpoint_name}: adjusting the number of concurrent readers to {max_concurrent_readers}"
            );

            // The semaphore doesn't allow changing the total token count directly, but at this point,
            // while initializing connectors, none of the tokens have been acquired, so we can adjust
            // the total number of tokens by adjusting currently available tokens.
            if max_concurrent_readers > available_permits {
                DELTA_READER_SEMAPHORE.add_permits(max_concurrent_readers - available_permits);
            } else {
                DELTA_READER_SEMAPHORE.forget_permits(available_permits - max_concurrent_readers);
            }
        };

        if !config.object_store_config.contains_key("timeout") {
            config.object_store_config.insert(
                "timeout".to_string(),
                DEFAULT_OBJECT_STORE_TIMEOUT.to_string(),
            );
        }

        Self::prepare_unity_config(&config.object_store_config);

        let input_stream = input_handle
            .handle
            .configure_arrow_deserializer(delta_input_serde_config())?;
        let schema = input_handle.schema.clone();

        let endpoint = Arc::new(DeltaTableInputEndpointInner::new(
            &endpoint_name,
            config,
            consumer,
            schema,
            resume_info.clone(),
        ));

        // This is needed to initialize completed_frontier for the connector. It's not ideal, as the completion
        // status for the frontier will be set to the current time instead of whenever the table version in resume_info
        // was actually processed. The right solution is to checkpoint the frontier with the connector.
        if resume_info.is_some() {
            endpoint
                .queue
                .push_with_aux((None, Vec::new()), Utc::now(), resume_info);
        }

        if eoi {
            endpoint
                .metrics
                .phase
                .store(DeltaPhase::Completed as u64, Ordering::Relaxed);
            endpoint.consumer.eoi();
        } else {
            let endpoint_clone = endpoint.clone();
            thread::Builder::new()
                .name(format!("{endpoint_name}-delta-input-tokio-wrapper"))
                .spawn(move || {
                    TOKIO.block_on(async {
                        let _ = endpoint_clone
                            .worker_task(input_stream, receiver_clone, init_status_sender)
                            .await;
                    })
                })
                .expect("failed to spawn delta connector tokio wrapper thread");

            init_status_receiver.blocking_recv().ok_or_else(|| {
                ControllerError::input_transport_error(
                    &endpoint.endpoint_name,
                    true,
                    anyhow!("worker thread terminated unexpectedly during initialization"),
                )
            })??;
        }

        Ok(Self {
            sender,
            inner: endpoint.clone(),
        })
    }

    /// deltalake_catalog_unity expects all configuration settings to be provided as environment variables.
    /// This function sets environment variables for all configuration settings that start with "unity_" or "databricks_".
    fn prepare_unity_config(object_store_config: &HashMap<String, String>) {
        for (key, val) in object_store_config.iter() {
            if key.to_lowercase().starts_with("unity_")
                || key.to_lowercase().starts_with("databricks_")
            {
                // TODO: Audit that the environment access only happens in single-threaded code.
                unsafe { std::env::set_var(key.to_uppercase(), val) };
            }
        }
    }
}

impl InputReader for DeltaTableInputReader {
    fn as_any(self: Arc<Self>) -> Arc<dyn std::any::Any + Send + Sync> {
        self
    }

    fn request(&self, command: InputReaderCommand) {
        match command {
            InputReaderCommand::Replay { .. } => panic!(
                "replay command is not supported by DeltaTableInputReader; this is a bug, please report it to developers"
            ),
            InputReaderCommand::Extend => {
                let _ = self.sender.send_replace(PipelineState::Running);
            }
            InputReaderCommand::Pause => {
                let _ = self.sender.send_replace(PipelineState::Paused);
            }
            InputReaderCommand::Queue {
                checkpoint_requested,
            } => {
                // When initiating a checkpoint, try to stop at a delta table transaction boundary.
                let stop_at: &dyn Fn(&Option<DeltaResumeInfo>) -> bool = if checkpoint_requested {
                    &|resume_info: &Option<DeltaResumeInfo>| resume_info.is_some()
                } else {
                    &|_: &Option<DeltaResumeInfo>| false
                };

                let (total, _, resume_info) = self.inner.queue.flush_with_aux_until(stop_at);
                let resume_status = resume_info
                    .last()
                    .map(|(_ts, resume_info)| resume_info.clone())
                    .unwrap_or_else(|| self.inner.last_resume_status.lock().unwrap().clone());
                *self.inner.last_resume_status.lock().unwrap() = resume_status.clone();

                let resume = match resume_status {
                    None => Resume::Barrier,
                    Some(delta_resume_info) => delta_resume_info.to_resume(),
                };

                // We use the same format (DeltaResumeInfo) for resume info and watermark metadata.
                self.inner.consumer.extended(
                    total,
                    Some(resume),
                    resume_info
                        .into_iter()
                        .map(|(timestamp, metadata)| {
                            Watermark::new(
                                timestamp,
                                metadata.map(|m| serde_json::to_value(m).unwrap()),
                            )
                        })
                        .collect(),
                );
            }
            InputReaderCommand::Disconnect => {
                let _ = self.sender.send_replace(PipelineState::Terminated);
            }
        }
    }

    fn is_closed(&self) -> bool {
        self.inner.queue.is_empty() && self.sender.is_closed()
    }
}

impl Drop for DeltaTableInputReader {
    fn drop(&mut self) {
        self.disconnect();
    }
}

/// Resume info stored in each checkpoint for the DeltaTableInputEndpoint.
#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone)]
struct DeltaResumeInfo {
    /// Table version where the connector stopped reading before the checkpoint,
    /// `None` if the connector hasn't started reading yet.
    ///
    /// If the connector is still processing the initial snapshot, this is the version of the snapshot.
    version: Option<i64>,

    /// The upper bound of timestamps previously ingested as part of the initial snapshot.
    ///
    /// Only set if the connector is still processing the initial snapshot,
    /// and if `timestamp_column` is set in the configuration.
    ///
    /// The connector will resume loading the snapshot from this timestamp after
    /// resuming from a checkpoint.
    snapshot_timestamp: Option<String>,

    /// True if the connector reached the end-of-input state.
    eoi: bool,
}

impl DeltaResumeInfo {
    /// Checkpoint taken before the connector started ingesting data.
    fn initial() -> Self {
        Self {
            version: None,
            snapshot_timestamp: None,
            eoi: false,
        }
    }

    /// Checkpoint taken after the connector reached the end-of-input state.
    fn eoi() -> Self {
        Self {
            version: None,
            snapshot_timestamp: None,
            eoi: true,
        }
    }

    /// Checkpoint taken in follow mode, after the connector has ingested the initial snapshot
    /// (if any).
    ///
    /// * `version` - the last processed version of the table.
    /// * `eoi` - true if the connector reached the end-of-input state
    ///   (can happen if `end_version` is set in the configuration or if the connector is
    ///   configured to ingest the initial snapshot only).
    fn follow_mode(version: i64, eoi: bool) -> Self {
        Self {
            version: Some(version),
            snapshot_timestamp: None,
            eoi,
        }
    }

    /// Checkpoint taken while the connector is ingesting the initial snapshot.
    ///
    /// * `version` is the version of the table being snapshotted.
    /// * `timestamp` is the timestamp of the last record ingested as part of the initial snapshot.
    fn snapshot_mode(version: i64, timestamp: &str) -> Self {
        Self {
            version: Some(version),
            snapshot_timestamp: Some(timestamp.to_string()),
            eoi: false,
        }
    }

    fn to_resume(&self) -> Resume {
        Resume::Seek {
            seek: serde_json::to_value(self).unwrap(),
        }
    }
}

/// Current phase of a delta table input connector.
// Using repr u64 as we using AtomicU64 to store the phase in metrics
#[repr(u64)]
enum DeltaPhase {
    LoadingSnapshot = 0,
    Follow = 1,
    Completed = 2,
}

struct DeltaTableMetrics {
    /// Current phase of the connector (see [DeltaPhase]).
    phase: AtomicU64,
    /// Unix epoch seconds when snapshot phase finished; 0 if not yet complete.
    snapshot_completed_ts: AtomicU64,
    /// Total records loaded during snapshot phase.
    snapshot_records_total: AtomicU64,
}

impl DeltaTableMetrics {
    fn new() -> Self {
        Self {
            phase: AtomicU64::new(DeltaPhase::LoadingSnapshot as u64),
            snapshot_completed_ts: AtomicU64::new(0),
            snapshot_records_total: AtomicU64::new(0),
        }
    }
}

impl ConnectorMetrics for DeltaTableMetrics {
    fn metrics(&self) -> Vec<(&'static str, &'static str, ValueType, f64)> {
        vec![
            (
                "input_connector_delta_phase",
                "Current phase: 0=loading_snapshot, 1=follow/streaming, 2=completed.",
                ValueType::Gauge,
                self.phase.load(Ordering::Relaxed) as f64,
            ),
            (
                "input_connector_delta_snapshot_completed_seconds",
                "Unix epoch seconds when the snapshot phase finished (0 if not yet complete).",
                ValueType::Gauge,
                self.snapshot_completed_ts.load(Ordering::Relaxed) as f64,
            ),
            (
                "input_connector_delta_snapshot_records_total",
                "Total records loaded during the snapshot phase.",
                ValueType::Counter,
                self.snapshot_records_total.load(Ordering::Relaxed) as f64,
            ),
        ]
    }
}

fn now_unix_secs() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

struct DeltaTableInputEndpointInner {
    endpoint_name: String,
    schema: Relation,
    config: DeltaTableReaderConfig,
    consumer: Box<dyn InputConsumer>,
    datafusion: SessionContext,

    /// Index of the last transaction initiated by this connector.
    ///
    /// This is different from the transaction ID assigned to the transaction by the controller.
    /// Used for debugging to track the number of transactions initiated by this connector.
    transaction_index: AtomicUsize,

    /// The latest resume status of this endpoint:
    /// * Initialized to `None` or `Some(version)` (when resume_info is specified) on initialization.
    /// * Updated to `Some(new_version)` after advancing to the next table version in the transaction log
    ///   in follow mode or after ingesting the initial snapshot.
    last_resume_status: Mutex<Option<DeltaResumeInfo>>,
    queue: Arc<InputQueue<Option<DeltaResumeInfo>, StagedInputBuffer>>,
    metrics: Arc<DeltaTableMetrics>,
}

impl DeltaTableInputEndpointInner {
    fn new(
        endpoint_name: &str,
        config: DeltaTableReaderConfig,
        consumer: Box<dyn InputConsumer>,
        schema: Relation,
        resume_info: Option<DeltaResumeInfo>,
    ) -> Self {
        let queue = Arc::new(InputQueue::new(consumer.clone()));
        // Configure datafusion not to generate Utf8View arrow types, which are not yet
        // supported by the `serde_arrow` crate.
        let session_config = SessionConfig::new().set_bool(
            "datafusion.execution.parquet.schema_force_view_types",
            false,
        );

        let metrics = Arc::new(DeltaTableMetrics::new());
        consumer.set_custom_metrics(Arc::clone(&metrics) as Arc<dyn ConnectorMetrics>);

        Self {
            endpoint_name: endpoint_name.to_string(),
            schema,
            config,
            consumer,
            datafusion: SessionContext::new_with_config(session_config),
            transaction_index: AtomicUsize::new(0),

            // Set version to None by default so that the connector is checkpointable in the initial state.
            last_resume_status: Mutex::new(Some(
                resume_info.unwrap_or_else(DeltaResumeInfo::initial),
            )),
            queue,
            metrics,
        }
    }

    fn skip_unused_columns(&self) -> bool {
        // Old-style: property in the connector configuration.
        // New-style: property in the table definition.
        self.config.skip_unused_columns
            || self.schema.get_property("skip_unused_columns") == Some("true")
    }

    fn allocate_follow_transaction_label(&self) -> Option<Option<String>> {
        if self.config.transaction_mode == DeltaTableTransactionMode::Always {
            Some(Some(format!(
                "follow-{}",
                self.transaction_index.fetch_add(1, Ordering::Release)
            )))
        } else {
            None
        }
    }

    fn allocate_snapshot_transaction_label(&self) -> Option<Option<String>> {
        if matches!(
            self.config.transaction_mode,
            DeltaTableTransactionMode::Always | DeltaTableTransactionMode::Snapshot
        ) {
            Some(Some(format!(
                "snapshot-{}",
                self.transaction_index.fetch_add(1, Ordering::Release),
            )))
        } else {
            None
        }
    }

    fn validate_cdc_config(&self) -> Result<(), ControllerError> {
        if self.config.is_cdc() {
            if self.config.cdc_order_by.is_none() {
                return Err(ControllerError::invalid_transport_configuration(
                    &self.endpoint_name,
                    "the DeltaLake connector is configured in 'cdc' mode, but the 'cdc_order_by' property is not set",
                ));
            };

            self.validate_cdc_order_by()?;

            if self.config.cdc_delete_filter.is_none() {
                return Err(ControllerError::invalid_transport_configuration(
                    &self.endpoint_name,
                    "the DeltaLake connector is configured in 'cdc' mode, but the 'cdc_delete_filter' property is not set",
                ));
            }
        } else {
            if self.config.cdc_delete_filter.is_some() {
                return Err(ControllerError::invalid_transport_configuration(
                    &self.endpoint_name,
                    "the 'cdc_delete_filter' property can only be used when 'mode=cdc'",
                ));
            }
            if self.config.cdc_order_by.is_some() {
                return Err(ControllerError::invalid_transport_configuration(
                    &self.endpoint_name,
                    "the 'cdc_order_by' property can only be used with 'mode=cdc'",
                ));
            }
        }

        Ok(())
    }

    async fn worker_task(
        self: Arc<Self>,
        input_stream: Box<dyn ArrowStream>,
        receiver: Receiver<PipelineState>,
        init_status_sender: mpsc::Sender<Result<(), ControllerError>>,
    ) {
        let mut receiver_clone = receiver.clone();
        select! {
            _ = Self::worker_task_inner(self.clone(), input_stream, receiver, init_status_sender) => {
                debug!("delta_table {}: worker task terminated",
                    &self.endpoint_name,
                );
            }
            _ = receiver_clone.wait_for(|state| state == &PipelineState::Terminated) => {
                debug!("delta_table {}: received termination command; worker task canceled",
                    &self.endpoint_name,
                );
            }
        }
    }

    /// Compute the subset of columns in the Delta table schema that occur in the SQL
    /// table declaration.
    fn used_columns(&self, table: &DeltaTable) -> Vec<String> {
        // Column names in the SQL schema.
        let sql_columns = if self.skip_unused_columns() {
            self.schema
                .fields
                .iter()
                // skip unused columns as long as they are nullable or have a default value.
                .filter(|f| !f.unused || (!f.columntype.nullable && f.default.is_none()))
                .map(|field| field.name.name())
                .collect::<Vec<String>>()
        } else {
            self.schema
                .fields
                .iter()
                .map(|field| field.name.name())
                .collect::<Vec<String>>()
        };

        let table_schema = table.schema();
        let delta_columns = table_schema
            .fields()
            .iter()
            .map(|f| f.name().to_string())
            .collect::<Vec<String>>();

        // We need to be careful in checking whether a Delta column name occurs in
        // sql_columns.  Delta doesn't seem to include case sensitivity information any
        // any form in its schema.  So we conservatively check whether column name
        // occurs in the SQL tables as is or whether a lowercase column name occurs
        // in the set of SQL columns converted to lowercase.

        let sql_columns = sql_columns.iter().cloned().collect::<BTreeSet<_>>();
        let sql_columns_lowercase = sql_columns
            .iter()
            .map(|c| c.to_lowercase())
            .collect::<BTreeSet<_>>();

        delta_columns
            .into_iter()
            .filter(|c| {
                sql_columns.contains(c) || sql_columns_lowercase.contains(&c.to_lowercase())
            })
            .collect::<Vec<_>>()
    }

    fn used_column_list(&self, table: &DeltaTable) -> String {
        let columns = self.used_columns(table);

        columns
            .iter()
            .map(quote_sql_identifier)
            .collect::<Vec<_>>()
            .join(", ")
    }

    /// Load the entire table snapshot as a single "select * where <filter>" query.
    /// Returns the total number of records processed.
    async fn read_unordered_snapshot(
        &self,
        table: &DeltaTable,
        input_stream: &mut dyn ArrowStream,
        receiver: &mut Receiver<PipelineState>,
    ) -> usize {
        let column_names = self.used_column_list(table);

        let mut snapshot_query = format!("select {column_names} from snapshot");
        if let Some(filter) = self.effective_snapshot_filter() {
            snapshot_query = format!("{snapshot_query} where {filter}");
        }
        // Execute the snapshot query; push snapshot data to the circuit.
        info!(
            "delta_table {}: reading initial snapshot: {snapshot_query}",
            &self.endpoint_name,
        );

        // Use the time when we started reading the snapshot as the ingestion timestamp for the snapshot.
        let timestamp = Utc::now();

        let record_count = self
            .execute_snapshot_query(&snapshot_query, "initial snapshot", input_stream, receiver)
            .await;
        self.metrics
            .snapshot_records_total
            .fetch_add(record_count as u64, Ordering::Relaxed);

        // Empty buffer to indicate checkpointable state.
        self.queue.push_entry(
            InputQueueEntry::new_with_aux(
                timestamp,
                Some(DeltaResumeInfo::follow_mode(
                    // We verified that the table version is not None in the open_table method.
                    table.version().unwrap(),
                    !self.config.follow(),
                )),
            )
            // If we started a transaction while processing the snapshot, commit it now.
            .with_commit_transaction(true),
            Vec::new(),
        );

        //let _ = self.datafusion.deregister_table("snapshot");
        info!(
            "delta_table {}: snapshot load completed (records: {}, version: {})",
            &self.endpoint_name,
            record_count,
            table.version().unwrap()
        );
        record_count
    }

    /// Load the initial snapshot by issuing a sequence of queries for monotonically
    /// increasing timestamp ranges.
    /// Returns the total number of records processed.
    async fn read_ordered_snapshot(
        &self,
        table: &DeltaTable,
        input_stream: &mut dyn ArrowStream,
        receiver: &mut Receiver<PipelineState>,
    ) -> usize {
        // Use the time when we started reading the snapshot as the ingestion timestamp for the snapshot.
        let timestamp = Utc::now();

        let total_records = self
            .read_ordered_snapshot_inner(table, input_stream, receiver)
            .await
            .unwrap_or_else(|e| {
                self.consumer.error(true, e, None);
                0
            });

        // Empty buffer to indicate checkpointable state.
        self.queue.push_entry(
            InputQueueEntry::new_with_aux(
                timestamp,
                Some(DeltaResumeInfo::follow_mode(
                    // We verified that the table version is not None in the open_table method.
                    table.version().unwrap(),
                    !self.config.follow(),
                )),
            )
            // If we started a transaction while processing the snapshot, commit it now.
            .with_commit_transaction(true),
            Vec::new(),
        );

        info!(
            "delta_table {}: snapshot load completed (records: {}, version: {})",
            &self.endpoint_name,
            total_records,
            table.version().unwrap()
        );
        total_records
    }

    async fn read_ordered_snapshot_inner(
        &self,
        table: &DeltaTable,
        input_stream: &mut dyn ArrowStream,
        receiver: &mut Receiver<PipelineState>,
    ) -> Result<usize, AnyError> {
        let timestamp_column = self.config.timestamp_column.as_ref().unwrap();

        let timestamp_field = self.schema.field(timestamp_column).unwrap();

        // The following unwraps are safe, as validated in `validate_timestamp_column`.
        let lateness = timestamp_field.lateness.as_ref().unwrap();

        // Query the table for min and max values of the timestamp column that satisfy the filter.
        let bounds_query = format!(
            "select * from (select cast(min({timestamp_column}) as string) as start_ts, cast(max({timestamp_column}) as string) as end_ts from snapshot {}) where start_ts is not null",
            if let Some(filter) = &self.effective_snapshot_filter() {
                format!("where {filter}")
            } else {
                String::new()
            }
        );

        let bounds = execute_query_collect(&self.datafusion, &bounds_query).await?;

        info!(
            "delta_table {}: querying the table for min and max timestamp values",
            &self.endpoint_name,
        );

        if bounds.len() != 1 || bounds[0].num_rows() != 1 {
            info!(
                "delta_table {}: initial snapshot is empty; the Delta table contains no records{}",
                &self.endpoint_name,
                if let Some(filter) = &self.effective_snapshot_filter() {
                    format!(" that satisfy the filter condition '{filter}'")
                } else {
                    String::new()
                }
            );
            return Ok(0);
        }

        if bounds[0].num_columns() != 2 {
            // Should never happen.
            return Err(anyhow!(
                "internal error: query '{bounds_query}' returned a result with {} columns; expected 2 columns",
                bounds[0].num_columns()
            ));
        }

        // If the connector is restarting from a checkpoint where the initial snapshot has been partially ingested,
        // resume from the checkpointed timestamp value; otherwise use the earliest timestamp in the table.
        let min = match &*self.last_resume_status.lock().unwrap()
        { Some(DeltaResumeInfo {
            snapshot_timestamp: Some(snapshot_timestamp),
            ..
        }) => {
            snapshot_timestamp.clone()
        } _ => {
            array_to_string(bounds[0].column(0))
            .ok_or_else(|| anyhow!("internal error: cannot retrieve the first column in the output of query '{bounds_query}' as a string"))?
        }};

        let max = array_to_string(bounds[0].column(1)).ok_or_else(|| {
            anyhow!(
                "internal error: cannot retrieve the second column in the output of query '{bounds_query}' as a string"
            )
        })?;

        let columns = self.used_columns(table);
        let column_names = columns
            .iter()
            .map(quote_sql_identifier)
            .collect::<Vec<_>>()
            .join(", ");

        info!(
            "delta_table {}: reading table snapshot in the range '{min} <= {timestamp_column} < {max}'",
            &self.endpoint_name,
        );

        let min = timestamp_to_sql_expression(&timestamp_field.columntype, &min);
        let max = timestamp_to_sql_expression(&timestamp_field.columntype, &max);

        let mut start = min.clone();
        let mut total_records = 0usize;

        loop {
            // Evaluate SQL expression for the new end of the interval.
            let end = execute_singleton_query(
                &self.datafusion,
                &format!("select cast(({start} + {lateness}) as string)"),
            )
            .await?;
            let end = timestamp_to_sql_expression(&timestamp_field.columntype, &end);

            // Query the table for the range.
            let mut range_query = format!(
                "select {column_names} from snapshot where {timestamp_column} >= {start} and {timestamp_column} < {end}"
            );
            if let Some(filter) = &self.effective_snapshot_filter() {
                range_query = format!("{range_query} and {filter}");
            }

            let range_record_count = self
                .execute_snapshot_query(&range_query, "range", input_stream, receiver)
                .await;
            self.metrics
                .snapshot_records_total
                .fetch_add(range_record_count as u64, Ordering::Relaxed);
            total_records += range_record_count;

            start = end.clone();

            let done = execute_singleton_query(
                &self.datafusion,
                &format!("select cast({start} > {max} as string)"),
            )
            .await?;

            if done == "true" {
                break;
            }

            self.queue.push_entry(
                InputQueueEntry::new_with_aux(
                    Utc::now(),
                    Some(DeltaResumeInfo::snapshot_mode(
                        // We verified that the table version is not None in the open_table method.
                        table.version().unwrap(),
                        &start,
                    )),
                )
                // If we started a transaction while processing the range query, commit it now.
                .with_commit_transaction(true),
                Vec::new(),
            );
        }

        Ok(total_records)
    }

    async fn worker_task_inner(
        self: Arc<Self>,
        mut input_stream: Box<dyn ArrowStream>,
        mut receiver: Receiver<PipelineState>,
        init_status_sender: mpsc::Sender<Result<(), ControllerError>>,
    ) {
        let table = match self.open_table().await {
            Err(e) => {
                let _ = init_status_sender.send(Err(e)).await;
                return;
            }
            Ok(table) => table,
        };

        let table = Arc::new(table);

        if let Err(e) = self.prepare_snapshot_query(&table).await {
            let _ = init_status_sender.send(Err(e)).await;
            return;
        };

        let cdc_delete_filter = match self.prepare_cdc().await {
            Err(e) => {
                let _ = init_status_sender.send(Err(e)).await;
                return;
            }
            Ok(cdc_delete_filter) => cdc_delete_filter,
        };

        // Code before this point is part of endpoint initialization.
        // After this point, the thread should continue running until it receives a
        // shutdown command from the controller.
        let _ = init_status_sender.send(Ok(())).await;

        // Wait for the pipeline to start before reading.
        wait_running(&mut receiver).await;

        let last_resume_status = self.last_resume_status.lock().unwrap().clone();

        // We verified that the table version is not None in the open_table method.
        let mut version = table.version().unwrap();

        // We haven't completed a snapshot before the checkpoint was taken if
        // - there is no checkpoint
        // - the checkpoint was taken in the initial state
        // - the checkpoint was taken mid-snapshot (snapshot_timestamp is set)
        let snapshot_incomplete = matches!(
            last_resume_status,
            None | Some(DeltaResumeInfo { version: None, .. })
                | Some(DeltaResumeInfo {
                    snapshot_timestamp: Some(_),
                    ..
                })
        );

        let mut snapshot_record_count = 0usize;

        if snapshot_incomplete && self.config.snapshot() && self.config.timestamp_column.is_none() {
            // Read snapshot chunk-by-chunk.
            snapshot_record_count = self
                .read_unordered_snapshot(&table, input_stream.as_mut(), &mut receiver)
                .await;
        } else if snapshot_incomplete && self.config.snapshot() {
            // Read the entire snapshot in one query.
            snapshot_record_count = self
                .read_ordered_snapshot(&table, input_stream.as_mut(), &mut receiver)
                .await;
        }

        // Start following the table if required by the configuration.
        if self.config.follow() {
            // Log the appropriate message based on whether we just completed a snapshot
            if self.config.snapshot() && snapshot_incomplete {
                // We just completed reading a snapshot, now switching to follow mode
                self.metrics
                    .snapshot_completed_ts
                    .store(now_unix_secs(), Ordering::Relaxed);
                self.metrics
                    .phase
                    .store(DeltaPhase::Follow as u64, Ordering::Relaxed);
                info!(
                    "delta_table {}: snapshot phase completed, switching to follow mode (records: {}, version: {})",
                    &self.endpoint_name, snapshot_record_count, version
                );
            } else if !self.config.snapshot() {
                // Pure CDC follow mode (no snapshot configured)
                self.metrics
                    .phase
                    .store(DeltaPhase::Follow as u64, Ordering::Relaxed);
                info!(
                    "delta_table {}: CDC follow started (from version: {})",
                    &self.endpoint_name, version
                );
            }
            // Note: If self.config.snapshot() && !snapshot_incomplete, we're resuming from a checkpoint
            // where the snapshot was already completed, so no special log needed

            let mut retry_count = 0;

            // If we haven't previously read a snapshot of the table, report initial frontier.
            // This makes sure that even if the current version of the table is the final version,
            // we will report the frontier.
            if !self.config.snapshot() {
                self.queue.push_with_aux(
                    (None, Vec::new()),
                    Utc::now(),
                    Some(DeltaResumeInfo::follow_mode(version, false)),
                );
            }

            loop {
                wait_running(&mut receiver).await;
                let new_version = version + 1;

                match table.log_store().read_commit_entry(new_version).await {
                    Ok(None) => sleep(POLL_INTERVAL).await,
                    Ok(Some(bytes))
                        if self.config.end_version.is_none()
                            || self.config.end_version.unwrap() >= new_version =>
                    {
                        retry_count = 0;

                        let actions = match logstore::get_actions(new_version, &bytes) {
                            Ok(actions) => actions,
                            Err(e) => {
                                self.consumer.error(
                                    true,
                                    anyhow!("error parsing log entry for table version {new_version}: {e}"),
                                    Some("delta-parse-log")
                                );
                                break;
                            }
                        };

                        version = new_version;
                        self.process_log_entry(
                            new_version,
                            &actions,
                            &table,
                            cdc_delete_filter.clone(),
                            input_stream.as_mut(),
                            &mut receiver,
                        )
                        .await;

                        if let Some(end_version) = self.config.end_version
                            && end_version <= new_version
                        {
                            info!(
                                "delta_table {}: reached table version {} specified as 'end_version' in connector config: stopping the connector",
                                &self.endpoint_name, end_version
                            );
                            self.metrics
                                .phase
                                .store(DeltaPhase::Completed as u64, Ordering::Relaxed);
                            self.consumer.eoi();
                            break;
                        }
                    }
                    Ok(Some(_bytes)) => {
                        info!(
                            "delta_table {}: reached table version {new_version}, which is greater than the 'end_version' {} specified in connector config: stopping the connector",
                            &self.endpoint_name,
                            self.config.end_version.unwrap()
                        );
                        self.metrics
                            .phase
                            .store(DeltaPhase::Completed as u64, Ordering::Relaxed);
                        self.consumer.eoi();

                        // Empty buffer to indicate eoi.
                        self.queue.push_with_aux(
                            (None, Vec::new()),
                            Utc::now(),
                            Some(DeltaResumeInfo::eoi()),
                        );

                        break;
                    }
                    Err(e) => {
                        // Transient timeouts are common when reading the next log entry from S3.
                        retry_count += 1;

                        if retry_count == 20 {
                            self.consumer.error(
                                true,
                                anyhow!("error reading the next log entry after {retry_count} attempts (current table version: {version}): {e}"),
                                Some("delta-next-log")
                            );
                            break;
                        } else {
                            let backoff_delay = calculate_backoff_delay(retry_count - 1);
                            warn!(
                                "delta_table {}: error reading the next log entry after {retry_count} attempts (current table version: {version}): {e}; retrying in {:?}",
                                &self.endpoint_name, backoff_delay
                            );
                            sleep(backoff_delay).await;
                        }
                    }
                }
            }
        } else {
            // Snapshot-only mode: record completion timestamp and transition to Completed.
            if self.config.snapshot() && snapshot_incomplete {
                self.metrics
                    .snapshot_completed_ts
                    .store(now_unix_secs(), Ordering::Relaxed);
            }
            self.metrics
                .phase
                .store(DeltaPhase::Completed as u64, Ordering::Relaxed);
            self.consumer.eoi();
        }
    }

    /// Open existing delta table.  Use version or timestamp specified in the configuration, if any.
    async fn open_table(&self) -> Result<DeltaTable, ControllerError> {
        debug!(
            "delta_table {}: opening delta table '{}'",
            &self.endpoint_name, &self.config.uri
        );

        // DeltaTableBuilder::from_uri panics on error, so we use `from_valid_uri` instead.
        let url = ensure_table_uri(&self.config.uri).map_err(|e| {
            ControllerError::invalid_transport_configuration(
                &self.endpoint_name,
                &format!("invalid Delta table uri '{}': {e}", &self.config.uri),
            )
        })?;

        let delta_table: DeltaTable = {
            let mut retry_count = 0;
            const MAX_RETRIES: u32 = 10;

            // We've seen the table builder get stuck forever in S3 authentication for some configurations
            // (see https://github.com/delta-io/delta-rs/issues/3768). So we add a timeout and retry logic
            // in that case.
            //
            // In other situations, the operation fails returning a timeout error. There is no easy way to
            // distinguish such errors from permanent failures such as incorrect credentials, we therefore
            // resort to checking the returned error message for the word "timeout".
            let mut operation_timeout: Duration = Duration::from_secs(60);

            loop {
                // DeltaTableBuilder doesn't implement Clone, so we need to create a new instance every time.
                let table_builder = DeltaTableBuilder::from_url(url.clone())
                    .map_err(|e| {
                        ControllerError::invalid_transport_configuration(
                            &self.endpoint_name,
                            &format!("invalid Delta table URL '{url}': {e}"),
                        )
                    })?
                    .with_storage_options(self.config.object_store_config.clone())
                    .with_io_runtime(IORuntime::RT(TOKIO_DEDICATED_IO.handle().clone()));

                let table_builder = if let Some(DeltaResumeInfo {
                    version: Some(version),
                    ..
                }) = self.last_resume_status.lock().unwrap().clone()
                {
                    // If we are resuming from a checkpoint, use the version specified in the checkpoint.
                    table_builder.with_version(version)
                } else {
                    match &self.config {
                        DeltaTableReaderConfig {
                            version: Some(_),
                            datetime: Some(_),
                            ..
                        } => {
                            return Err(ControllerError::invalid_transport_configuration(
                                &self.endpoint_name,
                                "at most one of 'version' and 'datetime' options can be specified",
                            ));
                        }
                        DeltaTableReaderConfig {
                            version: None,
                            datetime: None,
                            ..
                        } => table_builder,
                        DeltaTableReaderConfig {
                            version: Some(version),
                            datetime: None,
                            ..
                        } => table_builder.with_version(*version),
                        DeltaTableReaderConfig {
                            version: None,
                            datetime: Some(datetime),
                            ..
                        } => table_builder.with_datestring(datetime).map_err(|e| {
                            ControllerError::invalid_transport_configuration(
                                &self.endpoint_name,
                                &format!(
                            "invalid 'datetime' format (expected ISO-8601/RFC-3339 timestamp): {e}"
                        ),
                            )
                        })?,
                    }
                };

                match tokio::time::timeout(operation_timeout, table_builder.load()).await {
                    Ok(Ok(table)) => break table,
                    Ok(Err(e)) => {
                        // Timeout errors can originate from multiple transitive dependencies. There is no easy
                        // way to identify them in a strongly-typed fashion. Instead, we check for the "timeout"
                        // substring in a formatter representation of the error.
                        //
                        // Debug-format `e` as the timeout error is often found toward the end of the error chain.
                        let is_timeout = format!("{:?}", e).to_lowercase().contains("timeout");

                        if is_timeout && retry_count < MAX_RETRIES {
                            retry_count += 1;
                            let backoff_ms = min(1000 * (1 << (retry_count - 1)), 10_000);
                            warn!(
                                "delta_table {}: error loading delta table '{}' after {retry_count} attempts (retrying in {backoff_ms} ms): {e:?}",
                                &self.endpoint_name, &self.config.uri,
                            );

                            sleep(Duration::from_millis(backoff_ms)).await;
                        } else {
                            return Err(ControllerError::invalid_transport_configuration(
                                &self.endpoint_name,
                                &format!("error opening delta table '{}': {e:?}", &self.config.uri),
                            ));
                        }
                    }
                    Err(_timeout) => {
                        if retry_count >= MAX_RETRIES {
                            return Err(ControllerError::invalid_transport_configuration(
                                &self.endpoint_name,
                                &format!(
                                    "timeout loading delta table '{}' after {retry_count} attempts",
                                    &self.config.uri,
                                ),
                            ));
                        } else {
                            warn!(
                                "delta_table {}: timeout loading delta table '{}' after {retry_count} attempts, retrying",
                                &self.endpoint_name, &self.config.uri,
                            );
                            retry_count += 1;
                            if operation_timeout < Duration::from_secs(240) {
                                operation_timeout *= 2;
                            }
                        }
                    }
                }
            }
        };

        let version = delta_table.version().ok_or_else(|| {
            ControllerError::invalid_transport_configuration(
                &self.endpoint_name,
                "internal error: table version is not available",
            )
        })?;

        // If we are about to follow the table, set resume state to the current table version, otherwise
        // the connector will remain in the barrier state until at least one transaction is added to the log.
        if !self.config.snapshot() {
            *self.last_resume_status.lock().unwrap() =
                Some(DeltaResumeInfo::follow_mode(version, false));
        }

        // Register object store with datafusion, so it will recognize individual parquet
        // file URIs when processing transaction log.  The `object_store_url` function
        // generates a unique URL, which only makes sense to datafusion.  We must append
        // the same string to the relative file path we read from the log below to make
        // sure datafusion links it to this object store.
        let object_store_url = delta_table.log_store().object_store_url();
        let url: &Url = object_store_url.as_ref();
        self.datafusion
            .runtime_env()
            .register_object_store(url, delta_table.log_store().object_store(None));

        // if let Some(schema) = delta_table.schema() {
        //     info!("Delta table schema: {schema:?}");
        // }

        info!(
            "delta_table {}: opened delta table '{}' (current table version {})",
            &self.endpoint_name, &self.config.uri, version
        );

        // TODO: Validate that table schema matches relation schema

        // TODO: Validate that timestamp is a valid column.

        Ok(delta_table)
    }

    /// Register `table` as a Datafusion table named "snapshot".
    async fn register_snapshot_table(
        &self,
        table: &Arc<DeltaTable>,
    ) -> Result<(), ControllerError> {
        trace!(
            "delta_table {}: registering table with Datafusion",
            &self.endpoint_name,
        );

        self.datafusion
            .register_table("snapshot", table.clone())
            .map_err(|e| {
                ControllerError::input_transport_error(
                    &self.endpoint_name,
                    true,
                    anyhow!("failed to register table snapshot with datafusion: {e}"),
                )
            })?;

        Ok(())
    }

    /// Validate the filter expression specified in the 'snapshot_filter' parameter.
    fn validate_snapshot_filter(&self) -> Result<(), ControllerError> {
        if let Some(filter) = &self.config.snapshot_filter {
            validate_sql_expression(filter).map_err(|e| {
                ControllerError::invalid_transport_configuration(
                    &self.endpoint_name,
                    &format!("error parsing 'snapshot_filter' expression '{filter}': {e}"),
                )
            })?;
        }

        Ok(())
    }

    /// Validate the filter expression specified in the 'filter' parameter.
    fn validate_filter(&self) -> Result<(), ControllerError> {
        if let Some(filter) = &self.config.filter {
            validate_sql_expression(filter).map_err(|e| {
                ControllerError::invalid_transport_configuration(
                    &self.endpoint_name,
                    &format!("error parsing 'filter' expression '{filter}': {e}"),
                )
            })?;
        }

        Ok(())
    }

    /// SQL expression that combines the snapshot filter and the filter is either or both are specified.
    fn effective_snapshot_filter(&self) -> Option<String> {
        match (&self.config.snapshot_filter, &self.config.filter) {
            (Some(snapshot_filter), Some(filter)) => {
                Some(format!("({snapshot_filter}) and ({filter})"))
            }
            (Some(snapshot_filter), None) => Some(snapshot_filter.to_string()),
            (None, Some(filter)) => Some(filter.to_string()),
            (None, None) => None,
        }
    }

    /// Validate the cdc_order_by expression.
    fn validate_cdc_order_by(&self) -> Result<(), ControllerError> {
        if let Some(order_by) = &self.config.cdc_order_by {
            validate_sql_expression(order_by).map_err(|e| {
                ControllerError::invalid_transport_configuration(
                    &self.endpoint_name,
                    &format!("error parsing 'cdc_order_by' expression '{order_by}': {e}"),
                )
            })?;
        }

        Ok(())
    }

    /// Convert the delete filter SQL expression into a Datafusion PhysicalExpr.
    ///
    /// To do this, we generate a query plan for the `select * from snapshot where <delete_filter>`
    /// query and extract the logical expression from the plan and then convert it to
    /// a physical expression.
    //
    // FIXME: I wonder if there's a simpler way that doesn't require registering the `snapshot`
    // table just so we can compile this query.
    async fn extract_delete_filter_expr(
        &self,
    ) -> Result<Option<Arc<dyn PhysicalExpr>>, ControllerError> {
        let Some(delete_filter) = &self.config.cdc_delete_filter else {
            return Ok(None);
        };

        let query = format!("SELECT * FROM snapshot WHERE {}", delete_filter);

        let logical_plan = self
            .datafusion
            .sql(&query)
            .await
            .map_err(|e| {
                ControllerError::invalid_transport_configuration(
                    &self.endpoint_name,
                    &format!("invalid delete filter '{delete_filter}': the 'cdc_delete_filter' expression must be a valid SQL expression that can be used in a 'SELECT * FROM snapshot WHERE <cdc_delete_filter>' query, but the following error was encountered when compiling '{query}': {e}"),
                )
            })?
            .logical_plan()
            .clone();

        let filter_expr = if let LogicalPlan::Projection(filter_plan) = &logical_plan {
            if let LogicalPlan::Filter(filter) = filter_plan.input.as_ref() {
                filter.predicate.clone()
            } else {
                return Err(ControllerError::invalid_encoder_configuration(
                    &self.endpoint_name,
                    &format!(
                        "internal error when compiling the 'cdc_delete_filter' expression '{delete_filter}'; {REPORT_ERROR}: unexpected logical plan {logical_plan}"
                    ),
                ));
            }
        } else {
            return Err(ControllerError::invalid_encoder_configuration(
                &self.endpoint_name,
                &format!(
                    "internal error when compiling the 'cdc_delete_filter' expression '{delete_filter}'; {REPORT_ERROR}: unexpected logical plan {logical_plan}"
                ),
            ));
        };

        let schema = self
            .datafusion
            .table("snapshot")
            .await
            .map_err(|_e| {
                ControllerError::invalid_transport_configuration(&self.endpoint_name, &format!("internal error when compiling the 'cdc_delete_filter' expression '{delete_filter}'; {REPORT_ERROR}: table 'snapshot' not found"))
            })?
            .schema()
            .clone();

        let physical_expr = DefaultPhysicalPlanner::default()
            .create_physical_expr(&filter_expr, &schema, &self.datafusion.state())
            .map_err(|e| ControllerError::invalid_transport_configuration(&self.endpoint_name, &format!("internal error when compiling the 'cdc_delete_filter' expression '{delete_filter}'; {REPORT_ERROR}: error generating physical plan: {e}")))?;

        Ok(Some(physical_expr))
    }

    /// Evaluate delete filter expression against a batch of updates; returns a vector of polarities.
    async fn eval_delete_filter(
        delete_filter: &dyn PhysicalExpr,
        batch: &RecordBatch,
    ) -> AnyResult<Vec<bool>> {
        let deletions = delete_filter.evaluate(batch).map_err(|e| {
            anyhow!("internal error evaluating the delete filter expression; {REPORT_ERROR}: {e}")
        })?;

        let array = deletions
            .into_array(batch.num_rows())
            .map_err(|e| {
                anyhow!(
                    "internal error converting the result of the delete filter expressions to array; {REPORT_ERROR}: {e}"
                )
            })?;
        let bitmask = array
            .as_any()
            .downcast_ref::<BooleanArray>()
            .ok_or_else(|| {
                anyhow!(
                    "internal error converting the result of the delete filter expression to BooleanArray; {REPORT_ERROR}: expected Boolean, found {:?}", array.data_type()
                )
            })?;

        let polarities = bitmask
            .into_iter()
            // treat NULLs as inserts.
            .map(|b| !b.unwrap_or(false))
            .collect::<Vec<bool>>();

        Ok(polarities)
    }

    /// Prepare to read initial snapshot, if required by endpoint configuration.
    ///
    /// * register snapshot as a datafusion table
    /// * validate snapshot config: filter condition and timestamp column
    async fn prepare_snapshot_query(&self, table: &Arc<DeltaTable>) -> Result<(), ControllerError> {
        if !self.config.snapshot() && !self.config.is_cdc() {
            return Ok(());
        }

        self.register_snapshot_table(table).await?;
        self.validate_snapshot_filter()?;
        self.validate_filter()?;

        if let Some(timestamp_column) = &self.config.timestamp_column {
            validate_timestamp_column(
                &self.endpoint_name,
                timestamp_column,
                &self.datafusion,
                &self.schema,
                "see DeltaLake connector documentation for more details: https://docs.feldera.com/connectors/sources/delta"
            )
            .await?;
        };

        Ok(())
    }

    /// Prepare to process CDC stream.
    async fn prepare_cdc(&self) -> Result<Option<Arc<dyn PhysicalExpr>>, ControllerError> {
        self.validate_cdc_config()?;

        let delete_expr = self.extract_delete_filter_expr().await?;
        Ok(delete_expr)
    }

    /// Execute a SQL query to load a complete or partial snapshot of the DeltaTable.
    /// Returns the total number of records processed.
    async fn execute_snapshot_query(
        &self,
        query: &str,
        descr: &str,
        input_stream: &mut dyn ArrowStream,
        receiver: &mut Receiver<PipelineState>,
    ) -> usize {
        let descr = format!("{descr} query '{query}'");
        debug!(
            "delta_table {}: retrieving data from the Delta table snapshot using {descr}",
            &self.endpoint_name,
        );

        let options: SQLOptions = SQLOptions::new()
            .with_allow_ddl(false)
            .with_allow_dml(false);

        let df = match self.datafusion.sql_with_options(query, options).await {
            Ok(df) => df,
            Err(e) => {
                self.consumer
                    .error(true, anyhow!("error compiling query '{query}': {e}"), None);
                return 0;
            }
        };

        self.execute_df(
            df,
            true,
            None,
            &descr,
            input_stream,
            receiver,
            self.allocate_snapshot_transaction_label(),
        )
        .await
    }

    /// Execute a prepared dataframe and push data from it to the circuit.
    ///
    /// * `polarity` - determines whether records in the dataframe should be
    ///   inserted to or deleted from the table.
    ///
    /// * `descr` - dataframe description used to construct error message.
    ///
    /// * `input_stream` - handle to push updates to.
    ///
    /// * `receiver` - used to block the function until the endpoint is unpaused.
    ///
    /// * `transaction` - execute the dataframe as part of a transaction with the given label (is `Some`).
    ///
    /// Returns the total number of records processed.
    #[allow(clippy::too_many_arguments)]
    async fn execute_df(
        &self,
        dataframe: DataFrame,
        polarity: bool,
        cdc_delete_filter: Option<Arc<dyn PhysicalExpr>>,
        descr: &str,
        input_stream: &mut dyn ArrowStream,
        receiver: &mut Receiver<PipelineState>,
        transaction: Option<Option<String>>,
    ) -> usize {
        wait_running(receiver).await;

        // Limit the number of connectors simultaneously reading from Delta Lake.
        let _token = DELTA_READER_SEMAPHORE.acquire().await.unwrap();

        let mut stream = match dataframe.execute_stream().await {
            Err(e) => {
                self.consumer
                    .error(true, anyhow!("error retrieving {descr}: {e:?}"), None);
                return 0;
            }
            Ok(stream) => stream,
        };

        let mut num_batches = 0;
        let mut total_records = 0usize;

        let queue = self.queue.clone();

        let num_parsers = self.config.num_parsers as usize;

        // Create a job queue to efficiently parse record batches retrieved by the query.
        let job_queue = JobQueue::<
            (RecordBatch, DateTime<Utc>),
            (Option<StagedInputBuffer>, Vec<ParseError>, DateTime<Utc>),
        >::new(
            num_parsers,
            move || {
                let cdc_delete_filter: Option<Arc<dyn PhysicalExpr>> = cdc_delete_filter.clone();
                let input_stream = input_stream.fork();

                Box::new(move |(batch, timestamp)| {
                    Box::pin({
                        let cdc_delete_filter: Option<Arc<dyn PhysicalExpr>> =
                            cdc_delete_filter.clone();
                        let mut input_stream = input_stream.fork();

                        async move {
                            let (parsed_buffer, errors) = Self::parse_record_batch(
                                batch,
                                polarity,
                                &cdc_delete_filter,
                                input_stream.as_mut(),
                            )
                            .await;
                            let staged_buffer = parsed_buffer.map(|buffer| {
                                let len = buffer.len();
                                StagedInputBuffer::new(input_stream.stage(vec![buffer]), len)
                            });
                            (staged_buffer, errors, timestamp)
                        }
                    })
                })
            },
            move |(buffer, errors, timestamp)| {
                queue.push_entry(
                    InputQueueEntry::new_with_aux(timestamp, None)
                        .with_buffer(buffer)
                        .with_start_transaction(transaction.clone()),
                    errors,
                )
            },
        );

        // Use the timestamp when we start retrieving the next batch as the ingestion timestamp.
        let mut timestamp = Utc::now();

        while let Some(batch) = stream.next().await {
            wait_running(receiver).await;
            let batch = match batch {
                Ok(batch) => batch,
                Err(e) => {
                    self.consumer.error(
                        false,
                        anyhow!("error retrieving batch {num_batches} of {descr}: {e:?}"),
                        Some("delta-batch"),
                    );

                    continue;
                }
            };
            // info!("schema: {}", batch.schema());
            num_batches += 1;
            total_records += batch.num_rows();

            // Use the timestamp when the batch was retrieved as the ingestion timestamp.
            job_queue.push_job((batch, timestamp)).await;
            timestamp = Utc::now();
        }

        job_queue.flush().await;
        total_records
    }

    async fn parse_record_batch(
        batch: RecordBatch,
        polarity: bool,
        cdc_delete_filter: &Option<Arc<dyn PhysicalExpr>>,
        input_stream: &mut dyn ArrowStream,
    ) -> (Option<Box<dyn InputBuffer>>, Vec<ParseError>) {
        let result = if polarity {
            if let Some(delete_filter_expr) = cdc_delete_filter {
                let polarities =
                    match Self::eval_delete_filter(delete_filter_expr.as_ref(), &batch).await {
                        Ok(polarities) => polarities,
                        Err(e) => {
                            return (
                                None,
                                vec![ParseError::bin_envelope_error(e.to_string(), &[], None)],
                            );
                        }
                    };
                // println!(
                //     "insert_with_polarities: {} updates, {} insertions",
                //     polarities.len(),
                //     polarities.iter().filter(|x| **x == true).count()
                // );
                input_stream.insert_with_polarities(&batch, &polarities, &None)
            } else {
                input_stream.insert(&batch, &None)
            }
        } else {
            input_stream.delete(&batch, &None)
        };
        let errors = result.map_or_else(
            |e| {
                vec![ParseError::bin_envelope_error(
                    format!("error deserializing table records from Parquet data: {e}"),
                    &[],
                    None,
                )]
            },
            |()| Vec::new(),
        );

        (input_stream.take_all(), errors)
    }

    /// Apply actions from a transaction log entry.
    ///
    /// Only `Add` and `Remove` actions are picked up.
    async fn process_log_entry(
        &self,
        new_version: i64,
        actions: &[Action],
        table: &DeltaTable,
        cdc_delete_filter: Option<Arc<dyn PhysicalExpr>>,
        input_stream: &mut dyn ArrowStream,
        receiver: &mut Receiver<PipelineState>,
    ) {
        if self.config.verbose > 0 {
            // Don't log actions we ignore to limit spurious logging. E.g., delta lake
            // optimization passes can generate thousand of noop actions.
            let data_change_actions = actions
                .iter()
                .filter(|action| match action {
                    Action::Add(add) if add.data_change => true,
                    Action::Remove(remove) if remove.data_change => true,
                    _ => false,
                })
                .collect::<Vec<_>>();
            info!(
                "delta_table {}: log entry for table version {new_version}: {data_change_actions:?}{}",
                &self.endpoint_name,
                if actions.len() > data_change_actions.len() {
                    format!(
                        " ({} other actions)",
                        actions.len() - data_change_actions.len()
                    )
                } else {
                    "".to_string()
                }
            );
        }

        // Use the time when we _started_ reading transaction data as the ingestion timestamp.
        let timestamp = Utc::now();

        if self.config.is_cdc() {
            self.process_cdc_transaction(actions, table, cdc_delete_filter, input_stream, receiver)
                .await;
        } else {
            let column_names = self.used_column_list(table);

            let start_transaction = self.allocate_follow_transaction_label();

            // TODO: consider processing all Add actions and all Remove actions in one
            // go using `ListingTable`, which understands partitioning and can probably
            // parallelize the load.

            // Process deletes before inserts. Semantically, delete actions in
            // are applied before insert actions; however the delta standard doesn't
            // guarantee that actions occur in any particular order in the transaction log
            // entry.
            for action in actions {
                if matches!(action, Action::Remove(_)) {
                    self.process_action(
                        action,
                        table,
                        &column_names,
                        input_stream,
                        receiver,
                        start_transaction.clone(),
                    )
                    .await;
                }
            }

            for action in actions {
                if matches!(action, Action::Add(_)) {
                    self.process_action(
                        action,
                        table,
                        &column_names,
                        input_stream,
                        receiver,
                        start_transaction.clone(),
                    )
                    .await;
                }
            }
        }

        // Empty buffer to indicate checkpointable state.
        self.queue.push_entry(
            InputQueueEntry::new_with_aux(
                timestamp,
                Some(DeltaResumeInfo::follow_mode(
                    new_version,
                    self.config.end_version == Some(new_version),
                )),
            )
            // If we started a transaction while processing the log entry, commit it now.
            .with_commit_transaction(true),
            Vec::new(),
        );
    }

    /// Process a DeltaLake transaction in CDC mode:
    ///
    /// * Ignore delete actions (the assumption is that CDC tables are append-only,
    ///   where only old records are deleted as part of the table maintenance)
    /// * Order all rows from all insert actions by `cdc_order_by`, compute the
    ///   polarity of each row using `cdc_delete_filter` and insert the row with
    ///   appropriate polarity.
    async fn process_cdc_transaction(
        &self,
        actions: &[Action],
        table: &DeltaTable,
        cdc_delete_filter: Option<Arc<dyn PhysicalExpr>>,
        input_stream: &mut dyn ArrowStream,
        receiver: &mut Receiver<PipelineState>,
    ) {
        let result = self
            .do_process_cdc_transaction(actions, table, cdc_delete_filter, input_stream, receiver)
            .await;

        // Deregister the table registered by `do_process_cdc_transaction`.
        // If the table does not exist, there's no harm.
        let _ = self.datafusion.deregister_table("tmp_table");

        if let Err(e) = result {
            self.consumer.error(false, e, Some("delta-cdc"));
        }
    }

    async fn do_process_cdc_transaction(
        &self,
        actions: &[Action],
        table: &DeltaTable,
        cdc_delete_filter: Option<Arc<dyn PhysicalExpr>>,
        input_stream: &mut dyn ArrowStream,
        receiver: &mut Receiver<PipelineState>,
    ) -> AnyResult<()> {
        // List all files that occur in Add actions.
        let files = actions
            .iter()
            .flat_map(|action| match action {
                Action::Add(add) if add.data_change => Some(format!(
                    "{}{}",
                    table.log_store().object_store_url().as_str(),
                    add.path
                )),
                _ => None,
            })
            .collect::<Vec<_>>();

        let description = format!(
            "CDC transaction consisting of {} files {:?}",
            files.len(),
            &files
        );

        // Create a datafusion table backed by these files.
        let table = Arc::new(
            self.create_parquet_table(table, files, &description)
                .await?,
        );

        self.datafusion.register_table("tmp_table", table).map_err(|e| {
            anyhow!("internal error processing {description}; {REPORT_ERROR}; error registering Parquet table: {e}")
        })?;

        let where_clause = if let Some(filter) = &self.config.filter {
            format!("where {filter}")
        } else {
            "".to_string()
        };

        // Order the table by the `cdc_order_by` expression.
        // TODO: We don't use `used_column_list` here, as the resulting dataframe will have a different
        // schema than the original table, and the `cdc_delete_filter` physical expression won't be valid for it.
        let order_by = self.config.cdc_order_by.as_ref().unwrap();
        let query = format!("SELECT * FROM tmp_table {where_clause} ORDER BY {order_by}");

        let df = self.datafusion.sql(&query).await.map_err(|e| {
            anyhow!("invalid 'cdc_order_by' or 'filter' expression: 'cdc_order_by' and 'filter' (when specified) must be valid SQL expressions that can be used in a 'SELECT * FROM <table> WHERE <filter> ORDER BY <cdc_order_by>' query, but the following error was encountered when compiling '{query}': {e}")
        })?;

        let _record_count = self
            .execute_df(
                df,
                true,
                cdc_delete_filter,
                &description,
                input_stream,
                receiver,
                self.allocate_follow_transaction_label(),
            )
            .await;

        Ok(())
    }

    /// Create a table provider from a list of Parquet files.
    async fn create_parquet_table(
        &self,
        table: &DeltaTable,
        files: Vec<String>,
        description: &str,
    ) -> AnyResult<ListingTable> {
        let schema = table.schema();

        let mut urls = Vec::with_capacity(files.len());
        for file in files.iter() {
            urls.push(ListingTableUrl::parse(file).map_err(|e| anyhow!("internal error processing {description}; {REPORT_ERROR}; error converting file path '{file}' to table URL: {e}"))?);
        }

        let listing_options = ListingOptions::new(Arc::new(ParquetFormat::default()))
            .with_file_extension_opt(Some(".parquet"));

        let table_config = ListingTableConfig::new_with_multi_paths(urls)
            .with_listing_options(listing_options)
            .with_schema(schema);

        ListingTable::try_new(table_config).map_err(|e| {
            anyhow!("internal error processing {description}; {REPORT_ERROR}; error creating Parquet table: {e}")
        })
    }

    async fn process_action(
        &self,
        action: &Action,
        table: &DeltaTable,
        column_names: &str,
        input_stream: &mut dyn ArrowStream,
        receiver: &mut Receiver<PipelineState>,
        start_transaction: Option<Option<String>>,
    ) {
        let result = match action {
            Action::Add(add) if add.data_change => {
                self.add_with_polarity(
                    &add.path,
                    true,
                    table,
                    column_names,
                    input_stream,
                    receiver,
                    start_transaction,
                )
                .await
            }
            Action::Remove(remove)
                if remove.data_change && self.config.cdc_delete_filter.is_none() =>
            {
                self.add_with_polarity(
                    &remove.path,
                    false,
                    table,
                    column_names,
                    input_stream,
                    receiver,
                    start_transaction,
                )
                .await
            }
            _ => return,
        };

        // Deregister the table registered by `add_with_polarity`.
        // If the table does not exist, there's no harm.
        let _ = self.datafusion.deregister_table("tmp_table");

        if let Err(e) = result {
            self.consumer.error(false, e, Some("delta-action"));
        }
    }

    // TODO: here, as well as in `process_cdc_transaction`, we can get some potential speedup by only reading a subset
    // of columns that occurs in the SQL declaration, similar to how we already do when reading the snapshot.  However,
    // this requires some extra care, since different transactions in the log can have different schemas. The implementation
    // should therefore monitor for schema changes and update the set of relevant columns appropriately.
    #[allow(clippy::too_many_arguments)]
    async fn add_with_polarity(
        &self,
        path: &str,
        polarity: bool,
        table: &DeltaTable,
        column_names: &str,
        input_stream: &mut dyn ArrowStream,
        receiver: &mut Receiver<PipelineState>,
        start_transaction: Option<Option<String>>,
    ) -> AnyResult<()> {
        let description = format!("file '{path}'");

        // See comment about `object_store_url` above.
        let full_path = format!("{}{}", table.log_store().object_store_url().as_str(), path);

        // Create a datafusion table backed by these files.
        let table = Arc::new(
            self.create_parquet_table(table, vec![full_path.clone()], &description)
                .await?,
        );

        self.datafusion.register_table("tmp_table", table).map_err(|e| {
            anyhow!("internal error processing file {full_path}; {REPORT_ERROR}; error registering Parquet table: {e}")
        })?;

        let df = if let Some(filter) = &self.config.filter {
            let query = format!("SELECT {column_names} FROM tmp_table where {filter}");
            self.datafusion.sql(&query).await.map_err(|e| {
                anyhow!("invalid 'cdc_order_by' filter expression '{filter}': 'filter' must be a valid SQL expression that can be used in a 'SELECT * FROM <table> ORDER BY <cdc_order_by>' query, but the following error was encountered when compiling '{query}': {e}")
            })?
        } else {
            let query = format!("SELECT {column_names} FROM tmp_table");
            self.datafusion.sql(&query).await.map_err(|e| {
                anyhow!("internal error processing file {full_path}'; {REPORT_ERROR}; error compiling query '{query}': {e}")
            })?
        };

        let _record_count = self
            .execute_df(
                df,
                polarity,
                None,
                &description,
                input_stream,
                receiver,
                start_transaction,
            )
            .await;

        Ok(())
    }
}

/// Block until the state is `Running`.
async fn wait_running(receiver: &mut Receiver<PipelineState>) {
    // An error indicates that the channel was closed.  It's ok to ignore
    // the error as this situation will be handled by the top-level select,
    // which will abort the worker thread.
    let _ = receiver
        .wait_for(|state| state == &PipelineState::Running)
        .await;
}
