use arrow::record_batch::RecordBatch;
use ducklake_catalog::{
    DeleteFormat, FileColumnStats, FilePartitionValue, FileVariantStats, TableSchema,
};

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use crate::{Result, StorageError};

#[derive(Debug, Clone)]
pub struct AppendFile {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub data_file_id: ducklake_catalog::DataFileId,
    pub begin_snapshot: Option<ducklake_catalog::SnapshotId>,
    pub path: String,
    pub path_is_relative: bool,
    pub record_count: i64,
    pub file_size_bytes: i64,
    pub footer_size: Option<i64>,
    pub row_id_start: Option<i64>,
    /// True when the Parquet file carries `_ducklake_internal_row_id`.
    ///
    /// Normal append files leave this false so the catalog backend allocates a
    /// contiguous `row_id_start` range. Rewrite/update-style files set this to
    /// true so preserved row ids are read from the physical file and
    /// `ducklake_table_stats.next_row_id` is not advanced.
    pub explicit_row_ids: bool,
    pub column_stats: Vec<FileColumnStats>,
    pub variant_stats: Vec<FileVariantStats>,
    pub partition_id: Option<ducklake_catalog::PartitionId>,
    pub partition_values: Vec<FilePartitionValue>,
    pub encryption_key: Option<String>,
    pub mapping_id: Option<ducklake_catalog::MappingId>,
    pub partial_max: Option<ducklake_catalog::SnapshotId>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn append_file_row_id_start(
    file: &AppendFile,
    next_row_id: &mut Option<i64>,
    operation: &str,
) -> Result<(Option<i64>, i64)> {
    if file.explicit_row_ids {
        return Ok((file.row_id_start, 0));
    }

    let row_id_start = *next_row_id;
    if let Some(next_row_id) = next_row_id {
        *next_row_id = next_row_id
            .checked_add(file.record_count)
            .ok_or_else(|| StorageError::Backend(format!("{operation} row id overflow")))?;
    }
    Ok((row_id_start, file.record_count))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct AppendFilesCommitPlan {
    pub(crate) table_id: ducklake_catalog::TableId,
    pub(crate) expected_snapshot: ducklake_catalog::SnapshotId,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct AppendFilesSnapshotPlan {
    pub(crate) new_snapshot: ducklake_catalog::SnapshotId,
    pub(crate) new_next_file_id: i64,
    pub(crate) total_record_count: i64,
    pub(crate) total_allocated_row_id_count: i64,
    pub(crate) total_file_size_bytes: i64,
    pub(crate) files: Vec<AppendFilePlan>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct AppendFilePlan {
    pub(crate) row_id_start: Option<i64>,
    pub(crate) begin_snapshot: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_append_files_commit(files: &[AppendFile]) -> Result<AppendFilesCommitPlan> {
    let first = files
        .first()
        .ok_or_else(|| StorageError::Backend("DuckLake append commit received no files".into()))?;
    let table_id = first.table_id;
    let expected_snapshot = first.expected_snapshot;
    for file in files {
        if file.table_id != table_id {
            return Err(StorageError::Backend(
                "DuckLake append files span multiple tables".into(),
            ));
        }
        if file.expected_snapshot != expected_snapshot {
            return Err(StorageError::Backend(
                "DuckLake append files have inconsistent expected snapshots".into(),
            ));
        }
    }
    Ok(AppendFilesCommitPlan {
        table_id,
        expected_snapshot,
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl AppendFilesCommitPlan {
    pub(crate) fn plan_snapshot(
        &self,
        latest_snapshot: i64,
        next_file_id: i64,
        next_row_id: Option<i64>,
        files: &[AppendFile],
    ) -> Result<AppendFilesSnapshotPlan> {
        if latest_snapshot != self.expected_snapshot.0 {
            return Err(StorageError::conflict(
                "append",
                self.expected_snapshot,
                latest_snapshot,
            ));
        }
        for (idx, file) in files.iter().enumerate() {
            let expected_file_id = next_file_id
                + i64::try_from(idx).map_err(|_| {
                    StorageError::Backend("DuckLake append file index is out of range".into())
                })?;
            if expected_file_id != file.data_file_id.0 {
                return Err(StorageError::Backend(format!(
                    "DuckLake next_file_id sequence expected append file id {}, found {}",
                    expected_file_id, file.data_file_id.0
                )));
            }
        }

        let new_snapshot = latest_snapshot + 1;
        let mut next_row_id = next_row_id;
        let mut total_record_count = 0_i64;
        let mut total_allocated_row_id_count = 0_i64;
        let mut total_file_size_bytes = 0_i64;
        let mut planned_files = Vec::with_capacity(files.len());
        for file in files {
            let begin_snapshot = file
                .begin_snapshot
                .map(|snapshot| snapshot.0)
                .unwrap_or(new_snapshot);
            if begin_snapshot > new_snapshot {
                return Err(StorageError::Backend(format!(
                    "DuckLake append begin_snapshot {begin_snapshot} is greater than the new snapshot {new_snapshot}"
                )));
            }
            let (row_id_start, allocated_row_id_count) =
                append_file_row_id_start(file, &mut next_row_id, "DuckLake append")?;
            total_allocated_row_id_count = total_allocated_row_id_count
                .checked_add(allocated_row_id_count)
                .ok_or_else(|| {
                    StorageError::Backend("DuckLake append row count overflow".into())
                })?;
            total_record_count = total_record_count
                .checked_add(file.record_count)
                .ok_or_else(|| {
                    StorageError::Backend("DuckLake append row count overflow".into())
                })?;
            total_file_size_bytes = total_file_size_bytes
                .checked_add(file.file_size_bytes)
                .ok_or_else(|| {
                    StorageError::Backend("DuckLake append file size overflow".into())
                })?;
            planned_files.push(AppendFilePlan {
                row_id_start,
                begin_snapshot,
            });
        }

        let new_next_file_id = next_file_id
            + i64::try_from(files.len()).map_err(|_| {
                StorageError::Backend("DuckLake append file count is out of range".into())
            })?;
        Ok(AppendFilesSnapshotPlan {
            new_snapshot: ducklake_catalog::SnapshotId(new_snapshot),
            new_next_file_id,
            total_record_count,
            total_allocated_row_id_count,
            total_file_size_bytes,
            files: planned_files,
        })
    }
}

#[derive(Debug, Clone)]
pub struct DeleteFileCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub delete_file_id: ducklake_catalog::DeleteFileId,
    pub begin_snapshot: Option<ducklake_catalog::SnapshotId>,
    pub data_file_id: ducklake_catalog::DataFileId,
    pub path: String,
    pub path_is_relative: bool,
    pub format: DeleteFormat,
    pub delete_count: i64,
    pub rows_deleted: i64,
    pub file_size_bytes: i64,
    pub footer_size: Option<i64>,
    pub partial_max: Option<ducklake_catalog::SnapshotId>,
    pub overwritten_delete_file_id: Option<ducklake_catalog::DeleteFileId>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct DeleteFilesCommitPlan {
    pub(crate) table_id: ducklake_catalog::TableId,
    pub(crate) expected_snapshot: ducklake_catalog::SnapshotId,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct DeleteFilesSnapshotPlan {
    pub(crate) new_snapshot: ducklake_catalog::SnapshotId,
    pub(crate) new_next_file_id: i64,
    pub(crate) total_delete_count: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_delete_files_commit(
    files: &[DeleteFileCommit],
) -> Result<DeleteFilesCommitPlan> {
    let first = files
        .first()
        .ok_or_else(|| StorageError::Backend("DuckLake delete commit received no files".into()))?;
    let table_id = first.table_id;
    let expected_snapshot = first.expected_snapshot;
    for file in files {
        if file.table_id != table_id {
            return Err(StorageError::Backend(
                "DuckLake delete files span multiple tables".into(),
            ));
        }
        if file.expected_snapshot != expected_snapshot {
            return Err(StorageError::Backend(
                "DuckLake delete files have inconsistent expected snapshots".into(),
            ));
        }
    }
    Ok(DeleteFilesCommitPlan {
        table_id,
        expected_snapshot,
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl DeleteFilesCommitPlan {
    pub(crate) fn plan_snapshot(
        &self,
        latest_snapshot: i64,
        next_file_id: i64,
        files: &[DeleteFileCommit],
    ) -> Result<DeleteFilesSnapshotPlan> {
        if latest_snapshot != self.expected_snapshot.0 {
            return Err(StorageError::conflict(
                "delete",
                self.expected_snapshot,
                latest_snapshot,
            ));
        }
        for (idx, file) in files.iter().enumerate() {
            let expected_file_id = next_file_id
                + i64::try_from(idx).map_err(|_| {
                    StorageError::Backend("DuckLake delete file index is out of range".into())
                })?;
            if expected_file_id != file.delete_file_id.0 {
                return Err(StorageError::Backend(format!(
                    "DuckLake next_file_id sequence expected delete file id {}, found {}",
                    expected_file_id, file.delete_file_id.0
                )));
            }
        }

        let total_delete_count = files.iter().try_fold(0_i64, |acc, file| {
            acc.checked_add(file.rows_deleted)
                .ok_or_else(|| StorageError::Backend("DuckLake delete row count overflow".into()))
        })?;
        let new_next_file_id = next_file_id
            .checked_add(i64::try_from(files.len()).map_err(|_| {
                StorageError::Backend("DuckLake delete file count is out of range".into())
            })?)
            .ok_or_else(|| StorageError::Backend("DuckLake delete file id overflow".into()))?;
        Ok(DeleteFilesSnapshotPlan {
            new_snapshot: ducklake_catalog::SnapshotId(latest_snapshot + 1),
            new_next_file_id,
            total_delete_count,
        })
    }
}

#[derive(Debug, Clone)]
pub struct InlinedFileDeletionCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub data_file_id: ducklake_catalog::DataFileId,
    pub row_ids: Vec<i64>,
    pub rows_deleted: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl InlinedFileDeletionCommit {
    pub(crate) fn validated_row_ids(&self) -> Result<Vec<i64>> {
        if self.row_ids.is_empty() {
            return Err(StorageError::Backend(
                "DuckLake table-change inlined file deletion has no rows".into(),
            ));
        }
        if self.rows_deleted <= 0 {
            return Err(StorageError::Backend(
                "DuckLake table-change inlined file deletion row count must be positive".into(),
            ));
        }
        let expected = usize::try_from(self.rows_deleted).map_err(|_| {
            StorageError::Backend(
                "DuckLake table-change inlined file deletion row count is out of range".into(),
            )
        })?;
        if self.row_ids.len() != expected {
            return Err(StorageError::Backend(
                "DuckLake table-change inlined file deletion row count does not match row ids"
                    .into(),
            ));
        }
        let mut row_ids = self.row_ids.clone();
        if row_ids.iter().any(|row_id| *row_id < 0) {
            return Err(StorageError::Backend(
                "DuckLake table-change inlined file deletion row id is negative".into(),
            ));
        }
        row_ids.sort_unstable();
        let mut deduped = row_ids.clone();
        deduped.dedup();
        if deduped.len() != row_ids.len() {
            return Err(StorageError::Backend(
                "DuckLake table-change inlined file deletion row ids contain duplicates".into(),
            ));
        }
        Ok(row_ids)
    }
}

#[derive(Debug, Clone)]
pub struct InlinedDataDeletionCommit {
    pub table_name: String,
    pub row_ids: Vec<i64>,
    pub rows_deleted: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl InlinedDataDeletionCommit {
    pub(crate) fn validated_row_ids(&self) -> Result<Vec<i64>> {
        if self.table_name.trim().is_empty() {
            return Err(StorageError::Backend(
                "DuckLake inlined-data deletion table name is empty".into(),
            ));
        }
        if self.rows_deleted < 0 {
            return Err(StorageError::Backend(
                "DuckLake inlined-data deletion row count is negative".into(),
            ));
        }
        let expected = usize::try_from(self.rows_deleted).map_err(|_| {
            StorageError::Backend("DuckLake inlined-data deletion row count is out of range".into())
        })?;
        if self.row_ids.len() != expected {
            return Err(StorageError::Backend(
                "DuckLake inlined-data deletion row count does not match row ids".into(),
            ));
        }
        let mut row_ids = self.row_ids.clone();
        if row_ids.iter().any(|id| *id < 0) {
            return Err(StorageError::Backend(
                "DuckLake inlined-data deletion row id is negative".into(),
            ));
        }
        row_ids.sort_unstable();
        let deduped_len = {
            let mut deduped = row_ids.clone();
            deduped.dedup();
            deduped.len()
        };
        if deduped_len != row_ids.len() {
            return Err(StorageError::Backend(
                "DuckLake inlined-data deletion row ids contain duplicates".into(),
            ));
        }
        Ok(row_ids)
    }
}

#[derive(Debug, Clone)]
pub struct FlushedInlinedDataCommit {
    pub table_name: String,
    pub row_ids: Vec<i64>,
    pub rows_flushed: i64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct FlushedInlinedFileDeletion {
    pub data_file_id: ducklake_catalog::DataFileId,
    pub row_id: i64,
    pub begin_snapshot: ducklake_catalog::SnapshotId,
}

#[derive(Debug, Clone)]
pub struct FlushedInlinedFileDeletionCommit {
    pub rows: Vec<FlushedInlinedFileDeletion>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl FlushedInlinedDataCommit {
    pub(crate) fn validated_row_ids(&self) -> Result<Vec<i64>> {
        if self.table_name.trim().is_empty() {
            return Err(StorageError::Backend(
                "DuckLake flushed inlined-data table name is empty".into(),
            ));
        }
        if self.rows_flushed <= 0 {
            return Err(StorageError::Backend(
                "DuckLake flushed inlined-data row count must be positive".into(),
            ));
        }
        let expected = usize::try_from(self.rows_flushed).map_err(|_| {
            StorageError::Backend("DuckLake flushed inlined-data row count is out of range".into())
        })?;
        if self.row_ids.len() != expected {
            return Err(StorageError::Backend(
                "DuckLake flushed inlined-data row count does not match row ids".into(),
            ));
        }
        let mut row_ids = self.row_ids.clone();
        if row_ids.iter().any(|id| *id < 0) {
            return Err(StorageError::Backend(
                "DuckLake flushed inlined-data row id is negative".into(),
            ));
        }
        row_ids.sort_unstable();
        let mut deduped = row_ids.clone();
        deduped.dedup();
        if deduped.len() != row_ids.len() {
            return Err(StorageError::Backend(
                "DuckLake flushed inlined-data row ids contain duplicates".into(),
            ));
        }
        Ok(row_ids)
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl FlushedInlinedFileDeletionCommit {
    pub(crate) fn validated_rows(&self) -> Result<Vec<FlushedInlinedFileDeletion>> {
        if self.rows.is_empty() {
            return Err(StorageError::Backend(
                "DuckLake flushed inlined-file deletion commit has no rows".into(),
            ));
        }
        for row in &self.rows {
            if row.data_file_id.0 < 0 {
                return Err(StorageError::Backend(format!(
                    "DuckLake flushed inlined-file deletion data file id {} is negative",
                    row.data_file_id.0
                )));
            }
            if row.row_id < 0 {
                return Err(StorageError::Backend(
                    "DuckLake flushed inlined-file deletion row id is negative".into(),
                ));
            }
            if row.begin_snapshot.0 <= 0 {
                return Err(StorageError::Backend(format!(
                    "DuckLake flushed inlined-file deletion begin snapshot {} is not positive",
                    row.begin_snapshot.0
                )));
            }
        }
        let mut rows = self.rows.clone();
        rows.sort_unstable_by_key(|row| (row.data_file_id.0, row.row_id, row.begin_snapshot.0));
        let mut deduped = rows.clone();
        deduped.dedup();
        if deduped.len() != rows.len() {
            return Err(StorageError::Backend(
                "DuckLake flushed inlined-file deletion rows contain duplicates".into(),
            ));
        }
        Ok(rows)
    }
}

#[derive(Debug, Clone)]
pub struct InlinedDataAppendCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub schema_version: i64,
    pub schema: TableSchema,
    pub batches: Vec<RecordBatch>,
    pub column_stats: Vec<FileColumnStats>,
    pub rows_appended: i64,
    pub changes_made: String,
    pub commit_message: String,
}

#[derive(Debug, Clone)]
pub struct InlinedDataAppend {
    pub schema_version: i64,
    pub schema: TableSchema,
    pub batches: Vec<RecordBatch>,
    pub column_stats: Vec<FileColumnStats>,
    pub row_ids: Option<Vec<i64>>,
    pub rows_appended: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn validate_inlined_append_rows(
    rows_appended: i64,
    batches: &[RecordBatch],
    preserved_row_ids: Option<&[i64]>,
    context: &str,
) -> Result<()> {
    if rows_appended <= 0 {
        return Err(StorageError::Backend(format!("{context} received no rows")));
    }
    let actual_rows = batches.iter().try_fold(0_i64, |acc, batch| {
        let rows = i64::try_from(batch.num_rows())
            .map_err(|_| StorageError::Backend(format!("{context} row count overflow")))?;
        acc.checked_add(rows)
            .ok_or_else(|| StorageError::Backend(format!("{context} row count overflow")))
    })?;
    if actual_rows != rows_appended {
        return Err(StorageError::Backend(format!(
            "{context} row count does not match batches"
        )));
    }
    if let Some(row_ids) = preserved_row_ids {
        let expected = usize::try_from(rows_appended)
            .map_err(|_| StorageError::Backend(format!("{context} row count is out of range")))?;
        if row_ids.len() != expected {
            return Err(StorageError::Backend(format!(
                "{context} row-id count does not match rows"
            )));
        }
        if row_ids.iter().any(|row_id| *row_id < 0) {
            return Err(StorageError::Backend(format!(
                "{context} preserved row ids contain negative values"
            )));
        }
    }
    Ok(())
}

#[derive(Debug, Clone)]
pub struct InlinedDataBatch {
    pub batch: RecordBatch,
    pub row_ids: Option<Vec<i64>>,
    pub begin_snapshots: Option<Vec<ducklake_catalog::SnapshotId>>,
}

impl InlinedDataBatch {
    pub fn without_metadata(batch: RecordBatch) -> Self {
        Self {
            batch,
            row_ids: None,
            begin_snapshots: None,
        }
    }

    pub fn with_metadata(
        batch: RecordBatch,
        row_ids: Vec<i64>,
        begin_snapshots: Vec<ducklake_catalog::SnapshotId>,
    ) -> Self {
        Self {
            batch,
            row_ids: Some(row_ids),
            begin_snapshots: Some(begin_snapshots),
        }
    }
}

#[derive(Debug, Clone)]
pub struct TableChangeCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub append_files: Vec<AppendFile>,
    pub delete_files: Vec<DeleteFileCommit>,
    pub inlined_file_deletions: Vec<InlinedFileDeletionCommit>,
    pub inlined_data_appends: Vec<InlinedDataAppend>,
    pub inlined_data_deletions: Vec<InlinedDataDeletionCommit>,
    pub flushed_inlined_data: Vec<FlushedInlinedDataCommit>,
    pub flushed_inlined_file_deletions: Vec<FlushedInlinedFileDeletionCommit>,
    pub changes_made: String,
    pub commit_message: String,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct TableChangeCommitPlan {
    pub(crate) table_id: ducklake_catalog::TableId,
    pub(crate) expected_snapshot: ducklake_catalog::SnapshotId,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct TableChangeSnapshotPlan {
    pub(crate) new_snapshot: ducklake_catalog::SnapshotId,
    pub(crate) new_next_file_id: i64,
    pub(crate) next_row_id_after_appends: Option<i64>,
    pub(crate) total_appended_record_count: i64,
    pub(crate) total_allocated_row_id_count: i64,
    pub(crate) total_appended_file_size_bytes: i64,
    pub(crate) total_declared_deleted_rows: i64,
    pub(crate) append_files: Vec<AppendFilePlan>,
    pub(crate) inlined_data_appends: Vec<InlinedDataAppendPlan>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct InlinedDataAppendPlan {
    pub(crate) row_id_start: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_table_change_commit(
    changes: &TableChangeCommit,
) -> Result<TableChangeCommitPlan> {
    if changes.append_files.is_empty()
        && changes.delete_files.is_empty()
        && changes.inlined_file_deletions.is_empty()
        && changes.inlined_data_appends.is_empty()
        && changes.inlined_data_deletions.is_empty()
        && changes.flushed_inlined_data.is_empty()
        && changes.flushed_inlined_file_deletions.is_empty()
    {
        return Err(StorageError::Backend(
            "DuckLake table-change commit received no changes".into(),
        ));
    }
    for file in &changes.append_files {
        if file.table_id != changes.table_id {
            return Err(StorageError::Backend(
                "DuckLake table-change append files span multiple tables".into(),
            ));
        }
        if file.expected_snapshot != changes.expected_snapshot {
            return Err(StorageError::Backend(
                "DuckLake table-change append files have inconsistent expected snapshots".into(),
            ));
        }
    }
    for file in &changes.delete_files {
        if file.table_id != changes.table_id {
            return Err(StorageError::Backend(
                "DuckLake table-change delete files span multiple tables".into(),
            ));
        }
        if file.expected_snapshot != changes.expected_snapshot {
            return Err(StorageError::Backend(
                "DuckLake table-change delete files have inconsistent expected snapshots".into(),
            ));
        }
    }
    for deletion in &changes.inlined_file_deletions {
        if deletion.table_id != changes.table_id {
            return Err(StorageError::Backend(
                "DuckLake table-change inlined file deletions span multiple tables".into(),
            ));
        }
        if deletion.expected_snapshot != changes.expected_snapshot {
            return Err(StorageError::Backend(
                "DuckLake table-change inlined file deletions have inconsistent expected snapshots"
                    .into(),
            ));
        }
        let _ = deletion.validated_row_ids()?;
    }
    for flushed in &changes.flushed_inlined_data {
        let _ = flushed.validated_row_ids()?;
    }
    for flushed in &changes.flushed_inlined_file_deletions {
        let _ = flushed.validated_rows()?;
    }
    for append in &changes.inlined_data_appends {
        validate_inlined_append_rows(
            append.rows_appended,
            &append.batches,
            append.row_ids.as_deref(),
            "DuckLake table-change inlined append",
        )?;
    }
    Ok(TableChangeCommitPlan {
        table_id: changes.table_id,
        expected_snapshot: changes.expected_snapshot,
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl TableChangeCommitPlan {
    pub(crate) fn plan_snapshot(
        &self,
        latest_snapshot: i64,
        next_file_id: i64,
        next_row_id: Option<i64>,
        changes: &TableChangeCommit,
    ) -> Result<TableChangeSnapshotPlan> {
        if latest_snapshot != self.expected_snapshot.0 {
            return Err(StorageError::conflict(
                "table-change",
                self.expected_snapshot,
                latest_snapshot,
            ));
        }

        let mut file_ids = changes
            .append_files
            .iter()
            .map(|file| file.data_file_id.0)
            .chain(
                changes
                    .delete_files
                    .iter()
                    .map(|file| file.delete_file_id.0),
            )
            .collect::<Vec<_>>();
        file_ids.sort_unstable();
        for (idx, file_id) in file_ids.iter().enumerate() {
            let expected_file_id = next_file_id
                + i64::try_from(idx).map_err(|_| {
                    StorageError::Backend("DuckLake table-change file index is out of range".into())
                })?;
            if expected_file_id != *file_id {
                return Err(StorageError::Backend(format!(
                    "DuckLake next_file_id sequence expected file id {}, found {}",
                    expected_file_id, file_id
                )));
            }
        }

        let new_snapshot = latest_snapshot + 1;
        let mut next_row_id = next_row_id;
        let mut total_appended_record_count = 0_i64;
        let mut total_allocated_row_id_count = 0_i64;
        let mut total_appended_file_size_bytes = 0_i64;
        let mut total_declared_deleted_rows = 0_i64;
        let mut append_files = Vec::with_capacity(changes.append_files.len());
        let mut inlined_data_appends = Vec::with_capacity(changes.inlined_data_appends.len());
        for file in &changes.append_files {
            let begin_snapshot = file
                .begin_snapshot
                .map(|snapshot| snapshot.0)
                .unwrap_or(new_snapshot);
            let (row_id_start, allocated_row_id_count) =
                append_file_row_id_start(file, &mut next_row_id, "DuckLake table-change")?;
            total_allocated_row_id_count = total_allocated_row_id_count
                .checked_add(allocated_row_id_count)
                .ok_or_else(|| {
                    StorageError::Backend("DuckLake table-change row count overflow".into())
                })?;
            total_appended_record_count = total_appended_record_count
                .checked_add(file.record_count)
                .ok_or_else(|| {
                    StorageError::Backend("DuckLake table-change row count overflow".into())
                })?;
            total_appended_file_size_bytes = total_appended_file_size_bytes
                .checked_add(file.file_size_bytes)
                .ok_or_else(|| {
                    StorageError::Backend("DuckLake table-change file size overflow".into())
                })?;
            append_files.push(AppendFilePlan {
                row_id_start,
                begin_snapshot,
            });
        }
        for append in &changes.inlined_data_appends {
            let row_id_start = next_row_id.unwrap_or(0);
            total_appended_record_count = total_appended_record_count
                .checked_add(append.rows_appended)
                .ok_or_else(|| {
                    StorageError::Backend("DuckLake table-change row count overflow".into())
                })?;
            if append.row_ids.is_none() {
                total_allocated_row_id_count = total_allocated_row_id_count
                    .checked_add(append.rows_appended)
                    .ok_or_else(|| {
                        StorageError::Backend("DuckLake table-change row count overflow".into())
                    })?;
                let next = row_id_start
                    .checked_add(append.rows_appended)
                    .ok_or_else(|| {
                        StorageError::Backend("DuckLake table-change row_id overflow".into())
                    })?;
                next_row_id = Some(next);
            }
            inlined_data_appends.push(InlinedDataAppendPlan { row_id_start });
        }
        for file in &changes.delete_files {
            total_declared_deleted_rows = total_declared_deleted_rows
                .checked_add(file.rows_deleted)
                .ok_or_else(|| {
                    StorageError::Backend("DuckLake table-change delete count overflow".into())
                })?;
        }
        for deletion in &changes.inlined_file_deletions {
            total_declared_deleted_rows = total_declared_deleted_rows
                .checked_add(deletion.rows_deleted)
                .ok_or_else(|| {
                    StorageError::Backend("DuckLake table-change delete count overflow".into())
                })?;
        }

        let allocated_file_ids = if file_ids.is_empty() && !changes.inlined_data_appends.is_empty()
        {
            1
        } else {
            i64::try_from(file_ids.len()).map_err(|_| {
                StorageError::Backend("DuckLake table-change file count is out of range".into())
            })?
        };
        let new_next_file_id = next_file_id
            .checked_add(allocated_file_ids)
            .ok_or_else(|| {
                StorageError::Backend("DuckLake table-change file id overflow".into())
            })?;
        Ok(TableChangeSnapshotPlan {
            new_snapshot: ducklake_catalog::SnapshotId(new_snapshot),
            new_next_file_id,
            next_row_id_after_appends: next_row_id,
            total_appended_record_count,
            total_allocated_row_id_count,
            total_appended_file_size_bytes,
            total_declared_deleted_rows,
            append_files,
            inlined_data_appends,
        })
    }
}
