deltalake-core 0.32.0

//! Helper module to check if a transaction can be committed in case of conflicting commits.
use std::collections::HashSet;

use delta_kernel::table_properties::IsolationLevel;

use super::CommitInfo;
use crate::DeltaTableError;
#[cfg(feature = "datafusion")]
use crate::delta_datafusion::DataFusionMixins;
use crate::errors::DeltaResult;
use crate::kernel::{
    Action, Add, LogDataHandler, Metadata, Protocol, Remove, Transaction, Version,
};
use crate::logstore::{LogStore, get_actions};
use crate::protocol::DeltaOperation;
use crate::table::config::TablePropertiesExt as _;

#[cfg(feature = "datafusion")]
use super::state::AddContainer;
#[cfg(feature = "datafusion")]
use datafusion::logical_expr::Expr;
#[cfg(feature = "datafusion")]
use itertools::Either;

/// Exceptions raised during commit conflict resolution
#[derive(thiserror::Error, Debug)]
pub enum CommitConflictError {
    /// This exception occurs when a concurrent operation adds files in the same partition
    /// (or anywhere in an un-partitioned table) that your operation reads. The file additions
    /// can be caused by INSERT, DELETE, UPDATE, or MERGE operations.
    #[error(
        "Commit failed: a concurrent transactions added new data.\nHelp: This transaction's query must be rerun to include the new data. Also, if you don't care to require this check to pass in the future, the isolation level can be set to Snapshot Isolation."
    )]
    ConcurrentAppend,

    /// This exception occurs when a concurrent operation deleted a file that your operation read.
    /// Common causes are a DELETE, UPDATE, or MERGE operation that rewrites files.
    #[error(
        "Commit failed: a concurrent transaction deleted data this operation read.\nHelp: This transaction's query must be rerun to exclude the removed data. Also, if you don't care to require this check to pass in the future, the isolation level can be set to Snapshot Isolation."
    )]
    ConcurrentDeleteRead,

    /// This exception occurs when a concurrent operation deleted a file that your operation also deletes.
    /// This could be caused by two concurrent compaction operations rewriting the same files.
    #[error(
        "Commit failed: a concurrent transaction deleted the same data your transaction deletes.\nHelp: you should retry this write operation. If it was based on data contained in the table, you should rerun the query generating the data."
    )]
    ConcurrentDeleteDelete,

    /// This exception occurs when a concurrent transaction updates the metadata of a Delta table.
    /// Common causes are ALTER TABLE operations or writes to your Delta table that update the schema of the table.
    #[error("Metadata changed since last commit.")]
    MetadataChanged,

    /// If a streaming query using the same checkpoint location is started multiple times concurrently
    /// and tries to write to the Delta table at the same time. You should never have two streaming
    /// queries use the same checkpoint location and run at the same time.
    #[error("Concurrent transaction failed.")]
    ConcurrentTransaction,

    /// This exception can occur in the following cases:
    /// - When your Delta table is upgraded to a new version. For future operations to succeed
    ///   you may need to upgrade your Delta Lake version.
    /// - When multiple writers are creating or replacing a table at the same time.
    /// - When multiple writers are writing to an empty path at the same time.
    #[error("Protocol changed since last commit: {0}")]
    ProtocolChanged(String),

    /// Error returned when the table requires an unsupported writer version
    #[error("Delta-rs does not support writer version {0}")]
    UnsupportedWriterVersion(i32),

    /// Error returned when the table requires an unsupported writer version
    #[error("Delta-rs does not support reader version {0}")]
    UnsupportedReaderVersion(i32),

    /// Error returned when the snapshot has missing or corrupted data
    #[error("Snapshot is corrupted: {source}")]
    CorruptedState {
        /// Source error
        source: Box<dyn std::error::Error + Send + Sync + 'static>,
    },

    /// Error returned when evaluating predicate
    #[error("Error evaluating predicate: {source}")]
    Predicate {
        /// Source error
        source: Box<dyn std::error::Error + Send + Sync + 'static>,
    },

    /// Error returned when no metadata was found in the DeltaTable.
    #[error("No metadata found, please make sure table is loaded.")]
    NoMetadata,
}

/// A struct representing different attributes of current transaction needed for conflict detection.
#[allow(unused)]
pub(crate) struct TransactionInfo<'a> {
    txn_id: String,
    /// partition predicates by which files have been queried by the transaction
    ///
    /// If any new data files or removed data files match this predicate, the
    /// transaction should fail.
    #[cfg(not(feature = "datafusion"))]
    read_predicates: Option<String>,
    /// partition predicates by which files have been queried by the transaction
    #[cfg(feature = "datafusion")]
    read_predicates: Option<Expr>,
    /// appIds that have been seen by the transaction
    read_app_ids: HashSet<String>,
    /// delta log actions that the transaction wants to commit
    actions: &'a [Action],
    /// read [`DeltaTableState`] used for the transaction
    read_snapshot: LogDataHandler<'a>,
    /// Whether the transaction tainted the whole table
    read_whole_table: bool,
}

impl<'a> TransactionInfo<'a> {
    #[cfg(feature = "datafusion")]
    pub fn try_new(
        read_snapshot: LogDataHandler<'a>,
        read_predicates: Option<String>,
        actions: &'a [Action],
        read_whole_table: bool,
    ) -> DeltaResult<Self> {
        use datafusion::prelude::SessionContext;

        let session = SessionContext::new();
        let read_predicates = read_predicates
            .map(|pred| read_snapshot.parse_predicate_expression(pred, &session.state()))
            .transpose()?;

        let mut read_app_ids = HashSet::<String>::new();
        for action in actions.iter() {
            if let Action::Txn(Transaction { app_id, .. }) = action {
                read_app_ids.insert(app_id.clone());
            }
        }

        Ok(Self::new(
            read_snapshot,
            read_predicates,
            actions,
            read_whole_table,
        ))
    }

    #[cfg(feature = "datafusion")]
    pub fn new(
        read_snapshot: LogDataHandler<'a>,
        read_predicates: Option<Expr>,
        actions: &'a [Action],
        read_whole_table: bool,
    ) -> Self {
        let mut read_app_ids = HashSet::<String>::new();
        for action in actions.iter() {
            if let Action::Txn(Transaction { app_id, .. }) = action {
                read_app_ids.insert(app_id.clone());
            }
        }
        Self {
            txn_id: "".into(),
            read_predicates,
            read_app_ids,
            actions,
            read_snapshot,
            read_whole_table,
        }
    }

    #[cfg(not(feature = "datafusion"))]
    pub fn try_new(
        read_snapshot: LogDataHandler<'a>,
        read_predicates: Option<String>,
        actions: &'a Vec<Action>,
        read_whole_table: bool,
    ) -> DeltaResult<Self> {
        let mut read_app_ids = HashSet::<String>::new();
        for action in actions.iter() {
            if let Action::Txn(Transaction { app_id, .. }) = action {
                read_app_ids.insert(app_id.clone());
            }
        }
        Ok(Self {
            txn_id: "".into(),
            read_predicates,
            read_app_ids,
            actions,
            read_snapshot,
            read_whole_table,
        })
    }

    /// Whether the transaction changed the tables metadatas
    pub fn metadata_changed(&self) -> bool {
        self.actions
            .iter()
            .any(|a| matches!(a, Action::Metadata(_)))
    }

    #[cfg(feature = "datafusion")]
    /// Files read by the transaction
    pub fn read_files(&self) -> Result<impl Iterator<Item = Add> + '_, CommitConflictError> {
        use crate::delta_datafusion::files_matching_predicate;

        if let Some(predicate) = &self.read_predicates {
            Ok(Either::Left(
                files_matching_predicate(
                    self.read_snapshot.clone(),
                    std::slice::from_ref(predicate),
                )
                .map_err(|err| CommitConflictError::Predicate {
                    source: Box::new(err),
                })?,
            ))
        } else {
            Ok(Either::Right(self.read_snapshot.iter().map(|f| f.to_add())))
        }
    }

    #[cfg(not(feature = "datafusion"))]
    /// Files read by the transaction
    pub fn read_files(&self) -> Result<impl Iterator<Item = Add> + '_, CommitConflictError> {
        Ok(self.read_snapshot.iter().map(|f| f.to_add()))
    }

    /// Whether the whole table was read during the transaction
    pub fn read_whole_table(&self) -> bool {
        self.read_whole_table
    }
}

/// Summary of the Winning commit against which we want to check the conflict
#[derive(Debug)]
pub(crate) struct WinningCommitSummary {
    pub actions: Vec<Action>,
    pub commit_info: Option<CommitInfo>,
}

impl WinningCommitSummary {
    pub async fn try_new(
        log_store: &dyn LogStore,
        read_version: Version,
        winning_commit_version: Version,
    ) -> DeltaResult<Self> {
        // NOTE using assert, since a wrong version would right now mean a bug in our code.
        assert_eq!(winning_commit_version, read_version + 1);

        let commit_log_bytes = log_store.read_commit_entry(winning_commit_version).await?;
        match commit_log_bytes {
            Some(bytes) => {
                let actions = get_actions(winning_commit_version, &bytes)?; // ← ADD ? HERE
                let commit_info = actions
                    .iter()
                    .find(|action| matches!(action, Action::CommitInfo(_)))
                    .map(|action| match action {
                        Action::CommitInfo(info) => info.clone(),
                        _ => unreachable!(),
                    });

                Ok(Self {
                    actions,
                    commit_info,
                })
            }
            None => Err(DeltaTableError::InvalidVersion(winning_commit_version)),
        }
    }

    pub fn metadata_updates(&self) -> Vec<Metadata> {
        self.actions
            .iter()
            .cloned()
            .filter_map(|action| match action {
                Action::Metadata(metadata) => Some(metadata),
                _ => None,
            })
            .collect()
    }

    pub fn app_level_transactions(&self) -> HashSet<String> {
        self.actions
            .iter()
            .cloned()
            .filter_map(|action| match action {
                Action::Txn(txn) => Some(txn.app_id),
                _ => None,
            })
            .collect()
    }

    pub fn protocol(&self) -> Vec<Protocol> {
        self.actions
            .iter()
            .cloned()
            .filter_map(|action| match action {
                Action::Protocol(protocol) => Some(protocol),
                _ => None,
            })
            .collect()
    }

    pub fn removed_files(&self) -> Vec<Remove> {
        self.actions
            .iter()
            .cloned()
            .filter_map(|action| match action {
                Action::Remove(remove) => Some(remove),
                _ => None,
            })
            .collect()
    }

    pub fn added_files(&self) -> Vec<Add> {
        self.actions
            .iter()
            .cloned()
            .filter_map(|action| match action {
                Action::Add(add) => Some(add),
                _ => None,
            })
            .collect()
    }

    pub fn blind_append_added_files(&self) -> Vec<Add> {
        if self.is_blind_append().unwrap_or(false) {
            self.added_files()
        } else {
            vec![]
        }
    }

    pub fn changed_data_added_files(&self) -> Vec<Add> {
        if self.is_blind_append().unwrap_or(false) {
            vec![]
        } else {
            self.added_files()
        }
    }

    pub fn is_blind_append(&self) -> Option<bool> {
        self.commit_info
            .as_ref()
            .map(|opt| opt.is_blind_append.unwrap_or(false))
    }
}

/// Checks if a failed commit may be committed after a conflicting winning commit
pub(crate) struct ConflictChecker<'a> {
    /// transaction information for current transaction at start of check
    txn_info: TransactionInfo<'a>,
    /// Summary of the transaction, that has been committed ahead of the current transaction
    winning_commit_summary: WinningCommitSummary,
    /// Isolation level for the current transaction
    isolation_level: IsolationLevel,
}

impl<'a> ConflictChecker<'a> {
    pub fn new(
        transaction_info: TransactionInfo<'a>,
        winning_commit_summary: WinningCommitSummary,
        operation: Option<&DeltaOperation>,
    ) -> ConflictChecker<'a> {
        let isolation_level = operation
            .and_then(|op| {
                if can_downgrade_to_snapshot_isolation(
                    &winning_commit_summary.actions,
                    op,
                    &transaction_info
                        .read_snapshot
                        .table_properties()
                        .isolation_level(),
                ) {
                    Some(IsolationLevel::SnapshotIsolation)
                } else {
                    None
                }
            })
            .unwrap_or_else(|| {
                transaction_info
                    .read_snapshot
                    .table_properties()
                    .isolation_level()
            });

        Self {
            txn_info: transaction_info,
            winning_commit_summary,
            isolation_level,
        }
    }

    /// This function checks conflict of the `initial_current_transaction_info` against the
    /// `winning_commit_version` and returns an updated [`TransactionInfo`] that represents
    /// the transaction as if it had started while reading the `winning_commit_version`.
    pub fn check_conflicts(&self) -> Result<(), CommitConflictError> {
        self.check_protocol_compatibility()?;
        self.check_no_metadata_updates()?;
        self.check_for_added_files_that_should_have_been_read_by_current_txn()?;
        self.check_for_deleted_files_against_current_txn_read_files()?;
        self.check_for_deleted_files_against_current_txn_deleted_files()?;
        self.check_for_updated_application_transaction_ids_that_current_txn_depends_on()?;
        Ok(())
    }

    /// Asserts that the client is up to date with the protocol and is allowed
    /// to read and write against the protocol set by the committed transaction.
    fn check_protocol_compatibility(&self) -> Result<(), CommitConflictError> {
        for p in self.winning_commit_summary.protocol() {
            let (win_read, curr_read) = (
                p.min_reader_version(),
                self.txn_info.read_snapshot.protocol().min_reader_version(),
            );
            let (win_write, curr_write) = (
                p.min_writer_version(),
                self.txn_info.read_snapshot.protocol().min_writer_version(),
            );
            if curr_read < win_read || win_write < curr_write {
                return Err(CommitConflictError::ProtocolChanged(format!(
                    "required read/write {win_read}/{win_write}, current read/write {curr_read}/{curr_write}"
                )));
            };
        }
        if !self.winning_commit_summary.protocol().is_empty()
            && self
                .txn_info
                .actions
                .iter()
                .any(|a| matches!(a, Action::Protocol(_)))
        {
            return Err(CommitConflictError::ProtocolChanged(
                "protocol changed".into(),
            ));
        };
        Ok(())
    }

    /// Check if the committed transaction has changed metadata.
    fn check_no_metadata_updates(&self) -> Result<(), CommitConflictError> {
        // Fail if the metadata is different than what the txn read.
        if !self.winning_commit_summary.metadata_updates().is_empty() {
            Err(CommitConflictError::MetadataChanged)
        } else {
            Ok(())
        }
    }

    /// Check if the new files added by the already committed transactions
    /// should have been read by the current transaction.
    fn check_for_added_files_that_should_have_been_read_by_current_txn(
        &self,
    ) -> Result<(), CommitConflictError> {
        // Skip check, if the operation can be downgraded to snapshot isolation
        if matches!(self.isolation_level, IsolationLevel::SnapshotIsolation) {
            return Ok(());
        }

        // Fail if new files have been added that the txn should have read.
        let added_files_to_check = match self.isolation_level {
            IsolationLevel::WriteSerializable if !self.txn_info.metadata_changed() => {
                // don't conflict with blind appends
                self.winning_commit_summary.changed_data_added_files()
            }
            IsolationLevel::Serializable | IsolationLevel::WriteSerializable => {
                let mut files = self.winning_commit_summary.changed_data_added_files();
                files.extend(self.winning_commit_summary.blind_append_added_files());
                files
            }
            IsolationLevel::SnapshotIsolation => vec![],
        };

        // Here we need to check if the current transaction would have read the
        // added files. for this we need to be able to evaluate predicates. Err on the safe side is
        // to assume all files match
        cfg_if::cfg_if! {
            if #[cfg(feature = "datafusion")] {
                let added_files_matching_predicates = if let (Some(predicate), false) = (
                    &self.txn_info.read_predicates,
                    self.txn_info.read_whole_table(),
                ) {
                    let arrow_schema = self.txn_info.read_snapshot.read_schema();
                    let partition_columns = self
                        .txn_info
                        .read_snapshot
                        .metadata()
                        .partition_columns()
                        .to_vec();
                    AddContainer::new(&added_files_to_check, &partition_columns, arrow_schema)
                        .predicate_matches(predicate.clone())
                        .map_err(|err| CommitConflictError::Predicate {
                            source: Box::new(err),
                        })?
                        .cloned()
                        .collect::<Vec<_>>()
                } else if self.txn_info.read_whole_table() {
                    added_files_to_check
                } else {
                    vec![]
                };
            } else {
                let added_files_matching_predicates = if self.txn_info.read_whole_table()
                {
                    added_files_to_check
                } else {
                    vec![]
                };
            }
        }

        if !added_files_matching_predicates.is_empty() {
            Err(CommitConflictError::ConcurrentAppend)
        } else {
            Ok(())
        }
    }

    /// Check if [Remove] actions added by already committed transactions
    /// conflicts with files read by the current transaction.
    fn check_for_deleted_files_against_current_txn_read_files(
        &self,
    ) -> Result<(), CommitConflictError> {
        // Fail if files have been deleted that the txn read.
        let read_file_path: HashSet<String> = self
            .txn_info
            .read_files()?
            .map(|f| f.path.clone())
            .collect();

        // Only consider removals with data_change = true as conflicts.
        // Removals with data_change = false (e.g., from OPTIMIZE/compaction)
        // don't change the logical data, only the physical layout, so they
        // shouldn't conflict with concurrent read operations.
        let removed_files_with_data_change: Vec<Remove> = self
            .winning_commit_summary
            .removed_files()
            .into_iter()
            .filter(|r| r.data_change)
            .collect();

        let deleted_read_overlap = removed_files_with_data_change
            .iter()
            .find(|f| read_file_path.contains(&f.path));

        if deleted_read_overlap.is_some()
            || (!removed_files_with_data_change.is_empty() && self.txn_info.read_whole_table())
        {
            Err(CommitConflictError::ConcurrentDeleteRead)
        } else {
            Ok(())
        }
    }

    /// Check if [Remove] actions added by already committed transactions conflicts
    /// with [Remove] actions this transaction is trying to add.
    fn check_for_deleted_files_against_current_txn_deleted_files(
        &self,
    ) -> Result<(), CommitConflictError> {
        // Fail if a file is deleted twice.
        let txn_deleted_files: HashSet<String> = self
            .txn_info
            .actions
            .iter()
            .cloned()
            .filter_map(|action| match action {
                Action::Remove(remove) => Some(remove.path),
                _ => None,
            })
            .collect();
        let winning_deleted_files: HashSet<String> = self
            .winning_commit_summary
            .removed_files()
            .iter()
            .cloned()
            .map(|r| r.path)
            .collect();
        let intersection: HashSet<&String> = txn_deleted_files
            .intersection(&winning_deleted_files)
            .collect();

        if !intersection.is_empty() {
            Err(CommitConflictError::ConcurrentDeleteDelete)
        } else {
            Ok(())
        }
    }

    /// Checks if the winning transaction corresponds to some AppId on which
    /// current transaction also depends.
    fn check_for_updated_application_transaction_ids_that_current_txn_depends_on(
        &self,
    ) -> Result<(), CommitConflictError> {
        // Fail if the appIds seen by the current transaction has been updated by the winning
        // transaction i.e. the winning transaction have [Txn] corresponding to
        // some appId on which current transaction depends on. Example - This can happen when
        // multiple instances of the same streaming query are running at the same time.
        let winning_txns = self.winning_commit_summary.app_level_transactions();
        let txn_overlap: HashSet<&String> = winning_txns
            .intersection(&self.txn_info.read_app_ids)
            .collect();
        if !txn_overlap.is_empty() {
            Err(CommitConflictError::ConcurrentTransaction)
        } else {
            Ok(())
        }
    }
}

// implementation and comments adopted from
// https://github.com/delta-io/delta/blob/1c18c1d972e37d314711b3a485e6fb7c98fce96d/core/src/main/scala/org/apache/spark/sql/delta/OptimisticTransaction.scala#L1268
//
// For no-data-change transactions such as OPTIMIZE/Auto Compaction/ZorderBY, we can
// change the isolation level to SnapshotIsolation. SnapshotIsolation allows reduced conflict
// detection by skipping the
// [ConflictChecker::check_for_added_files_that_should_have_been_read_by_current_txn] check i.e.
// don't worry about concurrent appends.
//
// We can also use SnapshotIsolation for empty transactions. e.g. consider a commit:
// t0 - Initial state of table
// t1 - Q1, Q2 starts
// t2 - Q1 commits
// t3 - Q2 is empty and wants to commit.
// In this scenario, we can always allow Q2 to commit without worrying about new files
// generated by Q1.
//
// The final order which satisfies both Serializability and WriteSerializability is: Q2, Q1
// Note that Metadata only update transactions shouldn't be considered empty. If Q2 above has
// a Metadata update (say schema change/identity column high watermark update), then Q2 can't
// be moved above Q1 in the final SERIALIZABLE order. This is because if Q2 is moved above Q1,
// then Q1 should see the updates from Q2 - which actually didn't happen.
pub(super) fn can_downgrade_to_snapshot_isolation<'a>(
    actions: impl IntoIterator<Item = &'a Action>,
    operation: &DeltaOperation,
    isolation_level: &IsolationLevel,
) -> bool {
    let mut data_changed = false;
    let mut has_non_file_actions = false;
    for action in actions {
        match action {
            Action::Add(act) if act.data_change => data_changed = true,
            Action::Remove(rem) if rem.data_change => data_changed = true,
            _ => has_non_file_actions = true,
        }
    }

    if has_non_file_actions {
        // if Non-file-actions are present (e.g. METADATA etc.), then don't downgrade the isolation level.
        return false;
    }

    match isolation_level {
        IsolationLevel::Serializable => !data_changed,
        IsolationLevel::WriteSerializable => !data_changed && !operation.changes_data(),
        IsolationLevel::SnapshotIsolation => false, // this case should never happen, since spanpshot isolation cannot be configured on table
    }
}

#[cfg(test)]
#[allow(unused)]
mod tests {
    use std::collections::HashMap;

    #[cfg(feature = "datafusion")]
    use datafusion::logical_expr::{col, lit};
    use serde_json::json;

    use super::*;
    use crate::kernel::Action;
    use crate::test_utils::{ActionFactory, TestSchemas};

    fn simple_add(data_change: bool, min: &str, max: &str) -> Add {
        ActionFactory::add(
            TestSchemas::simple(),
            HashMap::from_iter([("value", (min, max))]),
            Default::default(),
            true,
        )
    }

    fn init_table_actions() -> Vec<Action> {
        vec![
            ActionFactory::protocol(None, None, None::<Vec<_>>, None::<Vec<_>>).into(),
            ActionFactory::metadata(TestSchemas::simple(), None::<Vec<&str>>, None).into(),
        ]
    }

    #[test]
    fn test_can_downgrade_to_snapshot_isolation() {
        let isolation = IsolationLevel::WriteSerializable;
        let operation = DeltaOperation::Optimize {
            predicate: None,
            target_size: 0,
        };
        let add =
            ActionFactory::add(TestSchemas::simple(), HashMap::new(), Vec::new(), true).into();
        let res = can_downgrade_to_snapshot_isolation(&[add], &operation, &isolation);
        assert!(!res)
    }

    // Check whether the test transaction conflict with the concurrent writes by executing the
    // given params in the following order:
    // - setup (including setting table isolation level
    // - reads
    // - concurrentWrites
    // - actions
    #[cfg(feature = "datafusion")]
    async fn execute_test(
        setup: Option<Vec<Action>>,
        reads: Option<Expr>,
        concurrent: Vec<Action>,
        actions: Vec<Action>,
        read_whole_table: bool,
    ) -> Result<(), CommitConflictError> {
        use crate::table::state::DeltaTableState;
        use object_store::path::Path;

        let setup_actions = setup.unwrap_or_else(init_table_actions);
        let state = DeltaTableState::from_actions(setup_actions).await.unwrap();
        let snapshot = state.snapshot();
        let transaction_info =
            TransactionInfo::new(snapshot.log_data(), reads, &actions, read_whole_table);
        let summary = WinningCommitSummary {
            actions: concurrent,
            commit_info: None,
        };
        let checker = ConflictChecker::new(transaction_info, summary, None);
        checker.check_conflicts()
    }

    // tests adopted from https://github.com/delta-io/delta/blob/24c025128612a4ae02d0ad958621f928cda9a3ec/core/src/test/scala/org/apache/spark/sql/delta/OptimisticTransactionSuite.scala#L40-L94
    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_concurrent_append_append() {
        // append file to table while a concurrent writer also appends a file
        let file1 = simple_add(true, "1", "10").into();
        let file2 = simple_add(true, "1", "10").into();

        let result = execute_test(None, None, vec![file1], vec![file2], false).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_disjoint_delete_read() {
        // the concurrent transaction deletes a file that the current transaction did NOT read
        let file_not_read = simple_add(true, "1", "10");
        let file_read = simple_add(true, "100", "10000").into();
        let mut setup_actions = init_table_actions();
        setup_actions.push(file_not_read.clone().into());
        setup_actions.push(file_read);
        let result = execute_test(
            Some(setup_actions),
            Some(col("value").gt(lit::<i32>(10))),
            vec![ActionFactory::remove(&file_not_read, true).into()],
            vec![],
            false,
        )
        .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_disjoint_add_read() {
        // concurrently add file, that the current transaction would not have read
        let file_added = simple_add(true, "1", "10").into();
        let file_read = simple_add(true, "100", "10000").into();
        let mut setup_actions = init_table_actions();
        setup_actions.push(file_read);
        let result = execute_test(
            Some(setup_actions),
            Some(col("value").gt(lit::<i32>(10))),
            vec![file_added],
            vec![],
            false,
        )
        .await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_concurrent_delete_delete() {
        // remove file from table that has previously been removed
        let removed_file = simple_add(true, "1", "10");
        let removed_file: Action = ActionFactory::remove(&removed_file, true).into();
        let result = execute_test(
            None,
            None,
            vec![removed_file.clone()],
            vec![removed_file],
            false,
        )
        .await;
        assert!(matches!(
            result,
            Err(CommitConflictError::ConcurrentDeleteDelete)
        ));
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_concurrent_add_conflicts_with_read_and_write() {
        // a file is concurrently added that should have been read by the current transaction
        let file_added = simple_add(true, "1", "10").into();
        let file_should_have_read = simple_add(true, "1", "10").into();
        let result = execute_test(
            None,
            Some(col("value").lt_eq(lit::<i32>(10))),
            vec![file_should_have_read],
            vec![file_added],
            false,
        )
        .await;
        assert!(matches!(result, Err(CommitConflictError::ConcurrentAppend)));
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_concurrent_delete_conflicts_with_read() {
        // transaction reads a file that is removed by concurrent transaction
        let file_read = simple_add(true, "1", "10");
        let mut setup_actions = init_table_actions();
        setup_actions.push(file_read.clone().into());
        let result = execute_test(
            Some(setup_actions),
            Some(col("value").lt_eq(lit::<i32>(10))),
            vec![ActionFactory::remove(&file_read, true).into()],
            vec![],
            false,
        )
        .await;
        assert!(matches!(
            result,
            Err(CommitConflictError::ConcurrentDeleteRead)
        ));
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_concurrent_metadata_change() {
        // concurrent transactions changes table metadata
        let result = execute_test(
            None,
            None,
            vec![ActionFactory::metadata(TestSchemas::simple(), None::<Vec<&str>>, None).into()],
            vec![],
            false,
        )
        .await;
        assert!(matches!(result, Err(CommitConflictError::MetadataChanged)));
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_concurrent_protocol_upgrade() {
        // current and concurrent transactions change the protocol version
        let result = execute_test(
            None,
            None,
            vec![ActionFactory::protocol(None, None, None::<Vec<_>>, None::<Vec<_>>).into()],
            vec![ActionFactory::protocol(None, None, None::<Vec<_>>, None::<Vec<_>>).into()],
            false,
        )
        .await;
        assert!(matches!(
            result,
            Err(CommitConflictError::ProtocolChanged(_))
        ));
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_read_whole_table_disallows_concurrent_append() {
        // `read_whole_table` should disallow any concurrent change, even if the change
        // is disjoint with the earlier filter
        let file_part1 = simple_add(true, "1", "10").into();
        let file_part2 = simple_add(true, "11", "100").into();
        let file_part3 = simple_add(true, "101", "1000").into();
        let mut setup_actions = init_table_actions();
        setup_actions.push(file_part1);
        let result = execute_test(
            Some(setup_actions),
            // filter matches neither existing nor added files
            Some(col("value").lt(lit::<i32>(0))),
            vec![file_part2],
            vec![file_part3],
            true,
        )
        .await;
        assert!(matches!(result, Err(CommitConflictError::ConcurrentAppend)));
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_read_whole_table_disallows_concurrent_remove() {
        // `read_whole_table` should disallow any concurrent remove actions
        let file_part1 = simple_add(true, "1", "10");
        let file_part2 = simple_add(true, "11", "100").into();
        let mut setup_actions = init_table_actions();
        setup_actions.push(file_part1.clone().into());
        let result = execute_test(
            Some(setup_actions),
            None,
            vec![ActionFactory::remove(&file_part1, true).into()],
            vec![file_part2],
            true,
        )
        .await;
        assert!(matches!(
            result,
            Err(CommitConflictError::ConcurrentDeleteRead)
        ));
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_compaction_remove_does_not_conflict_with_read() {
        // Concurrent compaction (data_change = false) should NOT conflict with reads
        // A file is removed by a compaction operation (data_change = false) while being read
        let file_read = simple_add(true, "1", "10");
        let mut setup_actions = init_table_actions();
        setup_actions.push(file_read.clone().into());

        // Create a remove action with data_change = false (simulating OPTIMIZE/compaction)
        let mut compaction_remove = ActionFactory::remove(&file_read, false);
        compaction_remove.data_change = false;

        let result = execute_test(
            Some(setup_actions),
            Some(col("value").lt_eq(lit::<i32>(10))),
            vec![compaction_remove.into()],
            vec![],
            false,
        )
        .await;
        // Should succeed because data_change = false means it's just a physical reorganization
        assert!(
            result.is_ok(),
            "Compaction with data_change=false should not conflict with reads"
        );
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_data_delete_conflicts_with_read() {
        // Delete with data_change = true should still conflict with reads
        let file_read = simple_add(true, "1", "10");
        let mut setup_actions = init_table_actions();
        setup_actions.push(file_read.clone().into());

        let result = execute_test(
            Some(setup_actions),
            Some(col("value").lt_eq(lit::<i32>(10))),
            vec![ActionFactory::remove(&file_read, true).into()],
            vec![],
            false,
        )
        .await;
        // Should fail because data_change = true means logical data was removed
        assert!(
            matches!(result, Err(CommitConflictError::ConcurrentDeleteRead)),
            "Delete with data_change=true should conflict with reads"
        );
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_compaction_does_not_conflict_with_whole_table_read() {
        // Concurrent compaction with read_whole_table and data_change = false
        let file_part1 = simple_add(true, "1", "10");
        let file_part2 = simple_add(true, "11", "100").into();
        let mut setup_actions = init_table_actions();
        setup_actions.push(file_part1.clone().into());

        let compaction_remove = ActionFactory::remove(&file_part1, false);

        let result = execute_test(
            Some(setup_actions),
            None,
            vec![compaction_remove.into()],
            vec![file_part2],
            true, // read_whole_table
        )
        .await;
        // Should succeed because data_change = false, even with read_whole_table
        assert!(
            result.is_ok(),
            "Compaction with data_change=false should not conflict even with read_whole_table"
        );
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_multiple_compaction_removes_do_not_conflict() {
        // Multiple files removed with data_change = false
        let file1 = simple_add(true, "1", "10");
        let file2 = simple_add(true, "11", "20");
        let mut setup_actions = init_table_actions();
        setup_actions.push(file1.clone().into());
        setup_actions.push(file2.clone().into());

        let mut remove1 = ActionFactory::remove(&file1, false);
        remove1.data_change = false;
        let mut remove2 = ActionFactory::remove(&file2, false);
        remove2.data_change = false;

        let result = execute_test(
            Some(setup_actions),
            Some(col("value").lt_eq(lit::<i32>(20))),
            vec![remove1.into(), remove2.into()],
            vec![],
            false,
        )
        .await;
        // Should succeed because both removes have data_change = false
        assert!(
            result.is_ok(),
            "Multiple compaction removes with data_change=false should not conflict"
        );
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_mixed_removes_conflict_if_any_data_change() {
        // Mixed removes - one with data_change = false, one with data_change = true
        let file1 = simple_add(true, "1", "10");
        let file2 = simple_add(true, "11", "20");
        let mut setup_actions = init_table_actions();
        setup_actions.push(file1.clone().into());
        setup_actions.push(file2.clone().into());

        let mut compaction_remove = ActionFactory::remove(&file1, false);
        compaction_remove.data_change = false;
        let data_remove = ActionFactory::remove(&file2, true); // data_change = true

        let result = execute_test(
            Some(setup_actions),
            Some(col("value").lt_eq(lit::<i32>(20))),
            vec![compaction_remove.into(), data_remove.into()],
            vec![],
            false,
        )
        .await;
        // Should fail because one of the removes has data_change = true
        assert!(
            matches!(result, Err(CommitConflictError::ConcurrentDeleteRead)),
            "Mixed removes should conflict if any have data_change=true"
        );
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_concurrent_compaction_double_delete_still_conflicts() {
        // Concurrent double delete with data_change = false
        // Both transactions try to remove the same file with data_change = false
        let removed_file = simple_add(true, "1", "10");
        let mut setup_actions = init_table_actions();
        setup_actions.push(removed_file.clone().into());

        let mut remove1 = ActionFactory::remove(&removed_file, false);
        remove1.data_change = false;
        let mut remove2 = ActionFactory::remove(&removed_file, false);
        remove2.data_change = false;

        let result = execute_test(
            Some(setup_actions),
            None,
            vec![remove1.into()],
            vec![remove2.into()],
            false,
        )
        .await;
        // Should still fail - even with data_change = false, can't delete the same file twice
        assert!(
            matches!(result, Err(CommitConflictError::ConcurrentDeleteDelete)),
            "Concurrent double delete should conflict even with data_change=false"
        );
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_disjoint_partitions_add_and_write() {
        // Test for: "add in part=2 / read from part=1,2 and write to part=1"
        // Transaction reads from partitions 1 and 2, concurrent txn adds to partition 2,
        // and current txn writes to partition 1.
        // This should succeed because the write is disjoint from the concurrent add.

        let file_part1_existing = simple_add(true, "1", "10");
        let file_part2_added = simple_add(true, "100", "200").into();
        let file_part1_new = simple_add(true, "5", "15").into();

        let mut setup_actions = init_table_actions();
        setup_actions.push(file_part1_existing.into());

        // Read from both partitions (value <= 200 covers both ranges)
        // Concurrent adds file with value 100-200
        // Current transaction adds file with value 5-15
        let result = execute_test(
            Some(setup_actions),
            Some(col("value").lt_eq(lit::<i32>(200))),
            vec![file_part2_added],
            vec![file_part1_new],
            false,
        )
        .await;

        // This should fail with ConcurrentAppend because the predicate matches the added file
        assert!(
            matches!(result, Err(CommitConflictError::ConcurrentAppend)),
            "Adding file matching read predicate should conflict"
        );
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_disjoint_partitions_read_write_different_ranges() {
        // Test disjoint partitions: read from one range, concurrent write to different range
        let file_part1 = simple_add(true, "1", "10");
        let file_part2_added = simple_add(true, "100", "200").into();
        let file_part1_new = simple_add(true, "5", "15").into();

        let mut setup_actions = init_table_actions();
        setup_actions.push(file_part1.into());

        // Read only from partition 1 (value <= 50)
        // Concurrent adds to partition 2 (value 100-200)
        // Current transaction adds to partition 1
        let result = execute_test(
            Some(setup_actions),
            Some(col("value").lt_eq(lit::<i32>(50))),
            vec![file_part2_added],
            vec![file_part1_new],
            false,
        )
        .await;

        // This should succeed because the concurrent add is outside the read predicate
        assert!(
            result.is_ok(),
            "Disjoint partition writes should not conflict"
        );
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_conflicting_app_transactions() {
        // Test for conflicting application transactions (e.g., duplicate streaming queries)
        let file1 = simple_add(true, "1", "10").into();

        let app_id = "streaming_query_1".to_string();
        let txn_action = Action::Txn(Transaction {
            app_id: app_id.clone(),
            version: 1,
            last_updated: None,
        });

        // Current transaction depends on app_id
        let current_actions = vec![txn_action.clone()];

        // Concurrent transaction also updates the same app_id
        let concurrent_actions = vec![txn_action, file1];

        let result = execute_test(None, None, concurrent_actions, current_actions, false).await;

        // Should fail because both transactions use the same app_id
        assert!(
            matches!(result, Err(CommitConflictError::ConcurrentTransaction)),
            "Conflicting app transactions should fail"
        );
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_non_conflicting_different_app_transactions() {
        // Test non-conflicting application transactions with different app_ids
        let file1 = simple_add(true, "1", "10").into();

        let app_id1 = "streaming_query_1".to_string();
        let app_id2 = "streaming_query_2".to_string();

        let txn_action1 = Action::Txn(Transaction {
            app_id: app_id1,
            version: 1,
            last_updated: None,
        });

        let txn_action2 = Action::Txn(Transaction {
            app_id: app_id2,
            version: 1,
            last_updated: None,
        });

        // Current transaction depends on app_id1
        let current_actions = vec![txn_action1];

        // Concurrent transaction updates app_id2
        let concurrent_actions = vec![txn_action2, file1];

        let result = execute_test(None, None, concurrent_actions, current_actions, false).await;

        // Should succeed because different app_ids don't conflict
        assert!(
            result.is_ok(),
            "Non-conflicting app transactions should succeed"
        );
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_replace_where_initial_empty_conflicts_on_concurrent_add() {
        // Empty predicate read, concurrent add within predicate, then current write -> conflicts
        let mut setup_actions = init_table_actions();
        setup_actions.push(simple_add(true, "1", "1").into());

        let result = execute_test(
            Some(setup_actions),
            Some(col("value").gt_eq(lit::<i32>(2))), // no files read
            vec![simple_add(true, "3", "3").into()], // concurrent add matches predicate
            vec![simple_add(true, "2", "2").into()],
            false,
        )
        .await;

        assert!(
            matches!(result, Err(CommitConflictError::ConcurrentAppend)),
            "ReplaceWhere-style empty read should conflict when a matching row is concurrently added"
        );
    }

    #[tokio::test]
    #[cfg(feature = "datafusion")]
    async fn test_replace_where_disjoint_empty_allows_commit() {
        // Empty predicate read, concurrent add outside predicate, then current write -> allowed
        let mut setup_actions = init_table_actions();
        setup_actions.push(simple_add(true, "1", "1").into());

        let result = execute_test(
            Some(setup_actions),
            Some(
                col("value")
                    .gt(lit::<i32>(1))
                    .and(col("value").lt_eq(lit::<i32>(3))),
            ), // empty read
            vec![simple_add(true, "5", "5").into()], // disjoint from read predicate
            vec![simple_add(true, "2", "2").into()],
            false,
        )
        .await;

        assert!(
            result.is_ok(),
            "Disjoint replaceWhere-style transactions with empty reads should succeed"
        );
    }
}