#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use std::collections::{HashMap, HashSet};

use ducklake_catalog::TableColumnStats;
#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
use ducklake_catalog::{CatalogError, ColumnEntry, DuckLakeType, SchemaField};

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

#[derive(Debug, Clone)]
pub struct ColumnNullabilityCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub column_id: ducklake_catalog::ColumnId,
    pub nulls_allowed: bool,
    pub existing_values_validated: bool,
    pub commit_message: String,
}

#[derive(Debug, Clone)]
pub struct ColumnDefaultCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub column_id: ducklake_catalog::ColumnId,
    pub default_value: Option<String>,
    pub default_value_type: Option<String>,
    pub default_value_dialect: Option<String>,
    pub commit_message: String,
}

#[derive(Debug, Clone)]
pub struct ColumnTypeCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub column_id: ducklake_catalog::ColumnId,
    pub new_column_type: String,
    pub commit_message: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ColumnTypeTree {
    pub column_name: String,
    pub column_type: String,
    pub nulls_allowed: bool,
    pub children: Vec<ColumnTypeTree>,
}

#[derive(Debug, Clone)]
pub struct ColumnTypeTreeCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub column_id: ducklake_catalog::ColumnId,
    pub replacement: ColumnTypeTree,
    pub commit_message: String,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone)]
pub(crate) struct PlannedColumnTypeTreeRow {
    pub column_id: ducklake_catalog::ColumnId,
    pub column_order: i64,
    pub column_name: String,
    pub column_type: String,
    pub initial_default: Option<String>,
    pub default_value: Option<String>,
    pub nulls_allowed: bool,
    pub parent_column: Option<ducklake_catalog::ColumnId>,
    pub default_value_type: Option<String>,
    pub default_value_dialect: Option<String>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ColumnCloseRequest {
    pub operation: &'static str,
    pub table_id: ducklake_catalog::TableId,
    pub column_id: ducklake_catalog::ColumnId,
    pub new_snapshot: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct ColumnSubtreeCloseRequest {
    pub operation: &'static str,
    pub table_id: ducklake_catalog::TableId,
    pub column_id: ducklake_catalog::ColumnId,
    pub latest_snapshot: ducklake_catalog::SnapshotId,
    pub new_snapshot: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone)]
pub(crate) enum PlannedColumnTypeTreeChange {
    Noop {
        snapshot: ducklake_catalog::SnapshotId,
    },
    Commit {
        metadata: MetadataMutationPlan,
        close_columns: Vec<ColumnCloseRequest>,
        insert_columns: Vec<PlannedColumnTypeTreeRow>,
    },
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_column_type_tree_mutation(
    change: ColumnTypeTreeCommit,
    snapshot: CatalogSnapshotIds,
    root: Option<SchemaField>,
    active_columns: &[ColumnEntry],
) -> Result<PlannedColumnTypeTreeChange> {
    if change.column_id.0 <= 0 {
        return Err(StorageError::Backend(format!(
            "DuckLake column type change column id {} is not positive",
            change.column_id.0
        )));
    }
    if snapshot.snapshot_id != change.expected_snapshot.0 {
        return Err(StorageError::conflict(
            "alter-column-type",
            change.expected_snapshot,
            snapshot.snapshot_id,
        ));
    }
    let root = root.ok_or_else(|| {
        StorageError::Catalog(ducklake_catalog::CatalogError::NotFound(format!(
            "column {} in table {}",
            change.column_id.0, change.table_id.0
        )))
    })?;

    let target = normalize_replacement_root(&root, change.replacement)?;
    validate_target_tree(&target)?;
    validate_type_tree_compatibility(&root, &target)?;
    if column_type_tree_matches(&root, &target)? {
        return Ok(PlannedColumnTypeTreeChange::Noop {
            snapshot: change.expected_snapshot,
        });
    }

    let mut next_catalog_id = snapshot.next_catalog_id;
    let mut insert_columns = Vec::new();
    let current_by_id = active_columns
        .iter()
        .map(|column| (column.column_id, column))
        .collect::<HashMap<_, _>>();
    let root_parent_column = current_by_id
        .get(&root.field_id)
        .ok_or_else(|| {
            StorageError::Catalog(ducklake_catalog::CatalogError::NotFound(format!(
                "column {}",
                root.field_id.0
            )))
        })?
        .parent_column;
    collect_replacement_rows(
        &root,
        &target,
        root_parent_column,
        root.column_order,
        &current_by_id,
        &mut next_catalog_id,
        &mut insert_columns,
    )?;

    let mut close_column_ids = Vec::new();
    collect_schema_field_ids(&root, &mut close_column_ids);
    close_column_ids.sort_unstable();
    close_column_ids.dedup();

    let metadata = snapshot.schema_metadata_mutation(
        "alter-column-type",
        change.expected_snapshot,
        change.table_id,
        next_catalog_id,
        change.commit_message,
    )?;
    let close_columns = close_column_ids
        .into_iter()
        .map(|column_id| ColumnCloseRequest {
            operation: "ALTER COLUMN TYPE",
            table_id: metadata.table_id,
            column_id,
            new_snapshot: metadata.new_snapshot,
        })
        .collect();
    Ok(PlannedColumnTypeTreeChange::Commit {
        metadata,
        close_columns,
        insert_columns,
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn normalize_replacement_root(
    root: &SchemaField,
    mut target: ColumnTypeTree,
) -> Result<ColumnTypeTree> {
    if target.column_name.trim().is_empty() {
        target.column_name = root.name.clone();
    }
    Ok(target)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn validate_target_tree(target: &ColumnTypeTree) -> Result<()> {
    let ducklake_type = DuckLakeType::parse(&target.column_type).map_err(|err| {
        StorageError::Backend(format!(
            "DuckLake target column type {:?} is not supported: {err}",
            target.column_type
        ))
    })?;
    if target.column_name.trim().is_empty() {
        return Err(StorageError::Backend(
            "DuckLake target column name is empty".into(),
        ));
    }
    match ducklake_type {
        DuckLakeType::Struct => {
            if target.children.is_empty() {
                return Err(StorageError::Backend(format!(
                    "DuckLake struct column {} has no child fields",
                    target.column_name
                )));
            }
        }
        DuckLakeType::List => {
            if target.children.len() != 1 {
                return Err(StorageError::Backend(format!(
                    "DuckLake list column {} must have exactly one child field",
                    target.column_name
                )));
            }
        }
        DuckLakeType::Map => {
            if target.children.len() != 2 {
                return Err(StorageError::Backend(format!(
                    "DuckLake map column {} must have key and value child fields",
                    target.column_name
                )));
            }
        }
        _ => {
            if !target.children.is_empty() {
                return Err(StorageError::Backend(format!(
                    "DuckLake scalar column {} cannot have child fields",
                    target.column_name
                )));
            }
        }
    }

    let mut sibling_names = HashSet::with_capacity(target.children.len());
    for child in &target.children {
        if !sibling_names.insert(child.column_name.to_ascii_lowercase()) {
            return Err(StorageError::Backend(format!(
                "DuckLake target column name {} is duplicated",
                child.column_name
            )));
        }
        validate_target_tree(child)?;
    }
    Ok(())
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn validate_type_tree_compatibility(root: &SchemaField, target: &ColumnTypeTree) -> Result<()> {
    let current_type = DuckLakeType::parse(&root.type_name).map_err(|err| {
        StorageError::Backend(format!(
            "DuckLake existing column type {:?} is not supported: {err}",
            root.type_name
        ))
    })?;
    let target_type = DuckLakeType::parse(&target.column_type).map_err(|err| {
        StorageError::Backend(format!(
            "DuckLake target column type {:?} is not supported: {err}",
            target.column_type
        ))
    })?;
    match (&current_type, &target_type) {
        (DuckLakeType::Struct, DuckLakeType::Struct)
        | (DuckLakeType::List, DuckLakeType::List)
        | (DuckLakeType::Map, DuckLakeType::Map) => {}
        _ if root.children.is_empty() && target.children.is_empty() => {
            if current_type != target_type && !target_type.is_valid_promotion_from(&current_type) {
                return Err(StorageError::Backend(format!(
                    "Cannot change type of column \"{}\" from {} to {}: only widening type promotions are supported",
                    root.name,
                    current_type.canonical_name(),
                    target_type.canonical_name()
                )));
            }
        }
        _ => {
            return Err(StorageError::Unsupported(format!(
                "DuckLake complex type replacement from {} to {} is not supported",
                current_type.canonical_name(),
                target_type.canonical_name()
            )));
        }
    }

    for child in &target.children {
        if let Some(existing) = match_child_field(&root.children, &child.column_name)? {
            validate_type_tree_compatibility(existing, child)?;
        }
    }
    Ok(())
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn column_type_tree_matches(root: &SchemaField, target: &ColumnTypeTree) -> Result<bool> {
    let current_type = DuckLakeType::parse(&root.type_name).map_err(|err| {
        StorageError::Backend(format!(
            "DuckLake existing column type {:?} is not supported: {err}",
            root.type_name
        ))
    })?;
    let target_type = DuckLakeType::parse(&target.column_type).map_err(|err| {
        StorageError::Backend(format!(
            "DuckLake target column type {:?} is not supported: {err}",
            target.column_type
        ))
    })?;
    if root.name != target.column_name
        || current_type.canonical_name() != target_type.canonical_name()
        || root.children.len() != target.children.len()
    {
        return Ok(false);
    }
    for (current, target) in root.children.iter().zip(&target.children) {
        if !column_type_tree_matches(current, target)? {
            return Ok(false);
        }
    }
    Ok(true)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn collect_replacement_rows(
    current: &SchemaField,
    target: &ColumnTypeTree,
    parent_column: Option<ducklake_catalog::ColumnId>,
    column_order: i64,
    current_by_id: &HashMap<ducklake_catalog::ColumnId, &ColumnEntry>,
    next_catalog_id: &mut i64,
    out: &mut Vec<PlannedColumnTypeTreeRow>,
) -> Result<ducklake_catalog::ColumnId> {
    let target_type = DuckLakeType::parse(&target.column_type).map_err(|err| {
        StorageError::Backend(format!(
            "DuckLake target column type {:?} is not supported: {err}",
            target.column_type
        ))
    })?;
    let current_entry = current_by_id.get(&current.field_id).ok_or_else(|| {
        StorageError::Catalog(ducklake_catalog::CatalogError::NotFound(format!(
            "column {}",
            current.field_id.0
        )))
    })?;
    let row = PlannedColumnTypeTreeRow {
        column_id: current.field_id,
        column_order,
        column_name: target.column_name.clone(),
        column_type: target_type.canonical_name(),
        initial_default: current_entry.initial_default.clone(),
        default_value: current_entry.default_value.clone(),
        nulls_allowed: current_entry.nulls_allowed,
        parent_column,
        default_value_type: current_entry.default_value_type.clone(),
        default_value_dialect: current_entry.default_value_dialect.clone(),
    };
    let column_id = row.column_id;
    out.push(row);

    for (child_order, target_child) in target.children.iter().enumerate() {
        let child_order = i64::try_from(child_order).map_err(|_| {
            StorageError::Backend("DuckLake target child order is out of range".into())
        })?;
        if let Some(existing_child) =
            match_child_field(&current.children, &target_child.column_name)?
        {
            collect_replacement_rows(
                existing_child,
                target_child,
                Some(column_id),
                child_order,
                current_by_id,
                next_catalog_id,
                out,
            )?;
        } else {
            collect_new_replacement_rows(
                target_child,
                Some(column_id),
                child_order,
                next_catalog_id,
                out,
            )?;
        }
    }
    Ok(column_id)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn collect_new_replacement_rows(
    target: &ColumnTypeTree,
    parent_column: Option<ducklake_catalog::ColumnId>,
    column_order: i64,
    next_catalog_id: &mut i64,
    out: &mut Vec<PlannedColumnTypeTreeRow>,
) -> Result<ducklake_catalog::ColumnId> {
    let target_type = DuckLakeType::parse(&target.column_type).map_err(|err| {
        StorageError::Backend(format!(
            "DuckLake target column type {:?} is not supported: {err}",
            target.column_type
        ))
    })?;
    let column_id = ducklake_catalog::ColumnId(*next_catalog_id);
    *next_catalog_id = next_catalog_id
        .checked_add(1)
        .ok_or_else(|| StorageError::Backend("DuckLake column id overflow".into()))?;
    out.push(PlannedColumnTypeTreeRow {
        column_id,
        column_order,
        column_name: target.column_name.clone(),
        column_type: target_type.canonical_name(),
        initial_default: None,
        default_value: Some("NULL".to_string()),
        nulls_allowed: target.nulls_allowed,
        parent_column,
        default_value_type: Some("literal".to_string()),
        default_value_dialect: Some("duckdb".to_string()),
    });

    for (child_order, child) in target.children.iter().enumerate() {
        let child_order = i64::try_from(child_order).map_err(|_| {
            StorageError::Backend("DuckLake target child order is out of range".into())
        })?;
        collect_new_replacement_rows(child, Some(column_id), child_order, next_catalog_id, out)?;
    }
    Ok(column_id)
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn match_child_field<'a>(
    children: &'a [SchemaField],
    column_name: &str,
) -> Result<Option<&'a SchemaField>> {
    if let Some(field) = children.iter().find(|field| field.name == column_name) {
        return Ok(Some(field));
    }
    let mut matches = children
        .iter()
        .filter(|field| field.name.eq_ignore_ascii_case(column_name));
    let Some(field) = matches.next() else {
        return Ok(None);
    };
    if matches.next().is_some() {
        return Err(StorageError::Backend(format!(
            "DuckLake column name {column_name:?} is ambiguous"
        )));
    }
    Ok(Some(field))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn collect_schema_field_ids(field: &SchemaField, out: &mut Vec<ducklake_catalog::ColumnId>) {
    out.push(field.field_id);
    for child in &field.children {
        collect_schema_field_ids(child, out);
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone)]
pub(crate) enum PlannedColumnTypeChange {
    Noop {
        snapshot: ducklake_catalog::SnapshotId,
    },
    Commit {
        replacement: PlannedColumnReplacement,
    },
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_column_type_mutation(
    change: ColumnTypeCommit,
    snapshot: CatalogSnapshotIds,
    column: Option<ColumnEntry>,
) -> Result<PlannedColumnTypeChange> {
    if change.column_id.0 <= 0 {
        return Err(StorageError::Backend(format!(
            "DuckLake column type change column id {} is not positive",
            change.column_id.0
        )));
    }
    if snapshot.snapshot_id != change.expected_snapshot.0 {
        return Err(StorageError::conflict(
            "alter-column-type",
            change.expected_snapshot,
            snapshot.snapshot_id,
        ));
    }
    let column = column.ok_or_else(|| {
        StorageError::Catalog(ducklake_catalog::CatalogError::NotFound(format!(
            "column {} in table {}",
            change.column_id.0, change.table_id.0
        )))
    })?;
    let current_type = DuckLakeType::parse(&column.column_type).map_err(|err| {
        StorageError::Backend(format!(
            "DuckLake existing column type {:?} is not supported: {err}",
            column.column_type
        ))
    })?;
    let new_type = DuckLakeType::parse(&change.new_column_type).map_err(|err| {
        StorageError::Backend(format!(
            "DuckLake target column type {:?} is not supported: {err}",
            change.new_column_type
        ))
    })?;
    if current_type == new_type {
        return Ok(PlannedColumnTypeChange::Noop {
            snapshot: change.expected_snapshot,
        });
    }
    if !new_type.is_valid_promotion_from(&current_type) {
        return Err(StorageError::Backend(format!(
            "Cannot change type of column \"{}\" from {} to {}: only widening type promotions are supported",
            column.column_name,
            current_type.canonical_name(),
            new_type.canonical_name()
        )));
    }

    let metadata = snapshot.schema_metadata_mutation(
        "alter-column-type",
        change.expected_snapshot,
        change.table_id,
        snapshot.next_catalog_id,
        change.commit_message,
    )?;
    let mut column = replacement_column(column, &metadata);
    column.column_type = new_type.canonical_name();
    Ok(PlannedColumnTypeChange::Commit {
        replacement: PlannedColumnReplacement { metadata, column },
    })
}

#[derive(Debug, Clone)]
pub struct RenameColumnCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub column_id: ducklake_catalog::ColumnId,
    pub new_column_name: String,
    pub commit_message: String,
}

#[derive(Debug, Clone)]
pub struct DropColumnCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub column_id: ducklake_catalog::ColumnId,
    pub commit_message: String,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy)]
pub(crate) struct CatalogSnapshotIds {
    pub snapshot_id: i64,
    pub schema_version: i64,
    pub next_catalog_id: i64,
    pub next_file_id: i64,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct MetadataMutationPlan {
    pub table_id: ducklake_catalog::TableId,
    pub new_snapshot: i64,
    pub new_schema_version: i64,
    pub next_catalog_id: i64,
    pub next_file_id: i64,
    pub changes_made: String,
    pub commit_message: String,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl CatalogSnapshotIds {
    pub(crate) fn schema_metadata_mutation(
        &self,
        operation: &'static str,
        expected_snapshot: ducklake_catalog::SnapshotId,
        table_id: ducklake_catalog::TableId,
        next_catalog_id: i64,
        commit_message: String,
    ) -> Result<MetadataMutationPlan> {
        ensure_expected_snapshot(operation, expected_snapshot, self.snapshot_id)?;
        let new_snapshot = self.snapshot_id.checked_add(1).ok_or_else(|| {
            StorageError::Backend("DuckLake metadata snapshot id overflow".into())
        })?;
        let new_schema_version = self.schema_version.checked_add(1).ok_or_else(|| {
            StorageError::Backend("DuckLake metadata schema version overflow".into())
        })?;
        Ok(MetadataMutationPlan {
            table_id,
            new_snapshot,
            new_schema_version,
            next_catalog_id,
            next_file_id: self.next_file_id,
            changes_made: format!("altered_table:{}", table_id.0),
            commit_message,
        })
    }

    pub(crate) fn table_metadata_mutation(
        &self,
        operation: &'static str,
        expected_snapshot: ducklake_catalog::SnapshotId,
        table_id: ducklake_catalog::TableId,
        next_catalog_id: i64,
        commit_message: String,
    ) -> Result<MetadataMutationPlan> {
        ensure_expected_snapshot(operation, expected_snapshot, self.snapshot_id)?;
        let new_snapshot = self.snapshot_id.checked_add(1).ok_or_else(|| {
            StorageError::Backend("DuckLake metadata snapshot id overflow".into())
        })?;
        Ok(MetadataMutationPlan {
            table_id,
            new_snapshot,
            new_schema_version: self.schema_version,
            next_catalog_id,
            next_file_id: self.next_file_id,
            changes_made: format!("altered_table:{}", table_id.0),
            commit_message,
        })
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone)]
pub(crate) struct PlannedColumnReplacement {
    pub metadata: MetadataMutationPlan,
    pub column: ColumnEntry,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl PlannedColumnReplacement {
    pub(crate) fn close_request(&self, operation: &'static str) -> ColumnCloseRequest {
        ColumnCloseRequest {
            operation,
            table_id: self.metadata.table_id,
            column_id: self.column.column_id,
            new_snapshot: self.metadata.new_snapshot,
        }
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone)]
pub(crate) struct PlannedColumnNullabilityChange {
    pub replacement: PlannedColumnReplacement,
    pub requires_existing_value_validation: bool,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone)]
pub(crate) struct PlannedRenameColumnStart {
    pub(crate) metadata: MetadataMutationPlan,
    pub(crate) column: ColumnEntry,
    pub(crate) new_column_name: String,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy)]
pub(crate) struct PlannedRenameColumnConflictLookup<'a> {
    pub table_id: ducklake_catalog::TableId,
    pub column_id: ducklake_catalog::ColumnId,
    pub parent_column: Option<ducklake_catalog::ColumnId>,
    pub new_column_name: &'a str,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy)]
pub(crate) struct PlannedRenameColumnSortRewrite<'a> {
    pub table_id: ducklake_catalog::TableId,
    pub latest_snapshot: ducklake_catalog::SnapshotId,
    pub new_snapshot: i64,
    pub next_catalog_id: i64,
    pub old_column_name: &'a str,
    pub new_column_name: &'a str,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl PlannedRenameColumnStart {
    pub(crate) fn conflict_lookup(&self) -> PlannedRenameColumnConflictLookup<'_> {
        PlannedRenameColumnConflictLookup {
            table_id: self.metadata.table_id,
            column_id: self.column.column_id,
            parent_column: self.column.parent_column,
            new_column_name: &self.new_column_name,
        }
    }

    pub(crate) fn sort_rewrite_request(
        &self,
        latest_snapshot: ducklake_catalog::SnapshotId,
        next_catalog_id: i64,
    ) -> Option<PlannedRenameColumnSortRewrite<'_>> {
        self.column
            .parent_column
            .is_none()
            .then_some(PlannedRenameColumnSortRewrite {
                table_id: self.metadata.table_id,
                latest_snapshot,
                new_snapshot: self.metadata.new_snapshot,
                next_catalog_id,
                old_column_name: &self.column.column_name,
                new_column_name: &self.new_column_name,
            })
    }

    pub(crate) fn finish(
        mut self,
        conflicting_column_name: Option<String>,
        next_catalog_id_after_sort_rewrite: i64,
    ) -> Result<PlannedColumnReplacement> {
        if let Some(conflict) = conflicting_column_name {
            return Err(StorageError::Backend(format!(
                "DuckLake column {} already exists in table {}",
                conflict, self.metadata.table_id.0
            )));
        }
        self.metadata.next_catalog_id = next_catalog_id_after_sort_rewrite;
        self.column.column_name = self.new_column_name;
        self.column.validity = ducklake_catalog::Validity {
            begin: ducklake_catalog::SnapshotId(self.metadata.new_snapshot),
            end: None,
        };
        Ok(PlannedColumnReplacement {
            metadata: self.metadata,
            column: self.column,
        })
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn require_column(
    column: Option<ColumnEntry>,
    column_id: ducklake_catalog::ColumnId,
    table_id: ducklake_catalog::TableId,
) -> Result<ColumnEntry> {
    column.ok_or_else(|| {
        StorageError::Catalog(ducklake_catalog::CatalogError::NotFound(format!(
            "column {} in table {}",
            column_id.0, table_id.0
        )))
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
fn replacement_column(mut column: ColumnEntry, metadata: &MetadataMutationPlan) -> ColumnEntry {
    column.validity = ducklake_catalog::Validity {
        begin: ducklake_catalog::SnapshotId(metadata.new_snapshot),
        end: None,
    };
    column
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_column_default_mutation(
    change: ColumnDefaultCommit,
    snapshot: CatalogSnapshotIds,
    column: Option<ColumnEntry>,
) -> Result<PlannedColumnReplacement> {
    if change.column_id.0 <= 0 {
        return Err(StorageError::Backend(format!(
            "DuckLake column default change column id {} is not positive",
            change.column_id.0
        )));
    }
    let metadata = snapshot.schema_metadata_mutation(
        "alter-column-default",
        change.expected_snapshot,
        change.table_id,
        snapshot.next_catalog_id,
        change.commit_message,
    )?;
    let mut column = replacement_column(
        require_column(column, change.column_id, change.table_id)?,
        &metadata,
    );
    column.default_value = change.default_value;
    column.default_value_type = change.default_value_type;
    column.default_value_dialect = change.default_value_dialect;
    Ok(PlannedColumnReplacement { metadata, column })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_column_nullability_mutation(
    change: ColumnNullabilityCommit,
    snapshot: CatalogSnapshotIds,
    column: Option<ColumnEntry>,
) -> Result<PlannedColumnNullabilityChange> {
    if change.column_id.0 <= 0 {
        return Err(StorageError::Backend(format!(
            "DuckLake column nullability change column id {} is not positive",
            change.column_id.0
        )));
    }
    let metadata = snapshot.schema_metadata_mutation(
        "alter-column-nullability",
        change.expected_snapshot,
        change.table_id,
        snapshot.next_catalog_id,
        change.commit_message,
    )?;
    let mut column = require_column(column, change.column_id, change.table_id)?;
    if column.nulls_allowed == change.nulls_allowed {
        let message = if change.nulls_allowed {
            format!(
                "Cannot DROP NULL on column {} - it has no NOT NULL constraint defined",
                column.column_name
            )
        } else {
            format!(
                "Cannot SET NOT NULL on column {} - it already has a NOT NULL constraint",
                column.column_name
            )
        };
        return Err(StorageError::Backend(message));
    }
    let requires_existing_value_validation =
        !change.nulls_allowed && !change.existing_values_validated;
    column = replacement_column(column, &metadata);
    column.nulls_allowed = change.nulls_allowed;
    Ok(PlannedColumnNullabilityChange {
        replacement: PlannedColumnReplacement { metadata, column },
        requires_existing_value_validation,
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_rename_column_mutation_start(
    rename: RenameColumnCommit,
    snapshot: CatalogSnapshotIds,
    column: Option<ColumnEntry>,
) -> Result<PlannedRenameColumnStart> {
    if rename.column_id.0 <= 0 {
        return Err(StorageError::Backend(format!(
            "DuckLake rename-column id {} is not positive",
            rename.column_id.0
        )));
    }
    if rename.new_column_name.trim().is_empty() {
        return Err(StorageError::Backend(
            "DuckLake rename-column target name is empty".into(),
        ));
    }
    let metadata = snapshot.schema_metadata_mutation(
        "rename-column",
        rename.expected_snapshot,
        rename.table_id,
        snapshot.next_catalog_id,
        rename.commit_message,
    )?;
    let column = require_column(column, rename.column_id, rename.table_id)?;
    Ok(PlannedRenameColumnStart {
        metadata,
        column,
        new_column_name: rename.new_column_name,
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone)]
pub(crate) struct DropColumnFacts {
    pub column: Option<ColumnEntry>,
    pub top_level_column_count: i64,
    pub referenced_by_partition: bool,
    pub referenced_by_sort: bool,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone, Copy)]
pub(crate) struct DropColumnSortReferenceLookup<'a> {
    pub table_id: ducklake_catalog::TableId,
    pub column_name: &'a str,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn drop_column_sort_reference_lookup(
    table_id: ducklake_catalog::TableId,
    column: Option<&ColumnEntry>,
) -> Option<DropColumnSortReferenceLookup<'_>> {
    column.map(|column| DropColumnSortReferenceLookup {
        table_id,
        column_name: &column.column_name,
    })
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
#[derive(Debug, Clone)]
pub(crate) struct PlannedDropColumn {
    pub metadata: MetadataMutationPlan,
    pub column_id: ducklake_catalog::ColumnId,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
impl PlannedDropColumn {
    pub(crate) fn close_request(
        &self,
        latest_snapshot: ducklake_catalog::SnapshotId,
    ) -> ColumnSubtreeCloseRequest {
        ColumnSubtreeCloseRequest {
            operation: "DROP COLUMN",
            table_id: self.metadata.table_id,
            column_id: self.column_id,
            latest_snapshot,
            new_snapshot: self.metadata.new_snapshot,
        }
    }
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_drop_column_mutation(
    drop: DropColumnCommit,
    snapshot: CatalogSnapshotIds,
    facts: DropColumnFacts,
) -> Result<PlannedDropColumn> {
    if drop.column_id.0 <= 0 {
        return Err(StorageError::Backend(format!(
            "DuckLake drop-column id {} is not positive",
            drop.column_id.0
        )));
    }
    if snapshot.snapshot_id != drop.expected_snapshot.0 {
        return Err(StorageError::conflict(
            "drop-column",
            drop.expected_snapshot,
            snapshot.snapshot_id,
        ));
    }
    let column = facts.column.ok_or_else(|| {
        StorageError::Catalog(ducklake_catalog::CatalogError::NotFound(format!(
            "column {} in table {}",
            drop.column_id.0, drop.table_id.0
        )))
    })?;
    if column.parent_column.is_none() && facts.top_level_column_count <= 1 {
        return Err(StorageError::Backend(
            "Cannot drop column: table only has one column remaining".into(),
        ));
    }
    if facts.referenced_by_partition {
        return Err(StorageError::Backend(format!(
            "Cannot drop column \"{}\" - the table is partitioned by this column",
            column.column_name
        )));
    }
    if facts.referenced_by_sort {
        return Err(StorageError::Backend(format!(
            "Cannot drop column \"{}\" - the table is sorted by this column",
            column.column_name
        )));
    }

    let metadata = snapshot.schema_metadata_mutation(
        "drop-column",
        drop.expected_snapshot,
        drop.table_id,
        snapshot.next_catalog_id,
        drop.commit_message,
    )?;
    Ok(PlannedDropColumn {
        metadata,
        column_id: drop.column_id,
    })
}

#[derive(Debug, Clone)]
pub struct AddColumnCommit {
    pub table_id: ducklake_catalog::TableId,
    pub expected_snapshot: ducklake_catalog::SnapshotId,
    pub columns: Vec<AddColumn>,
    pub table_column_stats: Vec<TableColumnStats>,
    pub commit_message: String,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AddColumn {
    pub column_id: ducklake_catalog::ColumnId,
    pub column_order: i64,
    pub column_name: String,
    pub column_type: String,
    pub initial_default: Option<String>,
    pub default_value: Option<String>,
    pub nulls_allowed: bool,
    pub parent_column: Option<ducklake_catalog::ColumnId>,
    pub default_value_type: Option<String>,
    pub default_value_dialect: Option<String>,
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn validate_add_column_commit(add: &AddColumnCommit) -> Result<()> {
    if add.columns.is_empty() {
        return Err(StorageError::Backend(
            "DuckLake ADD COLUMN commit received no columns".into(),
        ));
    }

    let mut seen_column_ids = HashSet::with_capacity(add.columns.len());
    let mut seen_sibling_names = HashSet::with_capacity(add.columns.len());
    for column in &add.columns {
        if column.column_id.0 <= 0 {
            return Err(StorageError::Backend(format!(
                "DuckLake ADD COLUMN id {} is not positive",
                column.column_id.0
            )));
        }
        if column.column_name.trim().is_empty() {
            return Err(StorageError::Backend(
                "DuckLake ADD COLUMN name is empty".into(),
            ));
        }
        if column.column_type.trim().is_empty() {
            return Err(StorageError::Backend(format!(
                "DuckLake ADD COLUMN {} type is empty",
                column.column_name
            )));
        }
        if !seen_column_ids.insert(column.column_id) {
            return Err(StorageError::Backend(format!(
                "DuckLake ADD COLUMN id {} is duplicated",
                column.column_id.0
            )));
        }
        if !seen_sibling_names.insert((
            column.parent_column.map(|parent| parent.0),
            column.column_name.to_ascii_lowercase(),
        )) {
            return Err(StorageError::Backend(format!(
                "DuckLake ADD COLUMN name {} is duplicated",
                column.column_name
            )));
        }
    }
    for stats in &add.table_column_stats {
        if stats.table_id != add.table_id {
            return Err(StorageError::Backend(format!(
                "DuckLake ADD COLUMN stats table {} does not match table {}",
                stats.table_id.0, add.table_id.0
            )));
        }
        if !seen_column_ids.contains(&stats.column_id) {
            return Err(StorageError::Backend(format!(
                "DuckLake ADD COLUMN stats reference non-added column {}",
                stats.column_id.0
            )));
        }
    }
    Ok(())
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn validate_add_column_catalog_state(
    add: &AddColumnCommit,
    next_catalog_id: i64,
    active_rows: &[(
        ducklake_catalog::ColumnId,
        String,
        Option<ducklake_catalog::ColumnId>,
    )],
) -> Result<i64> {
    let mut seen_column_ids = HashSet::with_capacity(add.columns.len());
    for column in &add.columns {
        seen_column_ids.insert(column.column_id);
    }

    for (idx, column) in add.columns.iter().enumerate() {
        let expected_column_id = next_catalog_id
            + i64::try_from(idx).map_err(|_| {
                StorageError::Backend("DuckLake ADD COLUMN index is out of range".into())
            })?;
        if column.column_id.0 != expected_column_id {
            return Err(StorageError::Backend(format!(
                "DuckLake next_catalog_id sequence expected ADD COLUMN id {}, found {}",
                expected_column_id, column.column_id.0
            )));
        }
    }

    let active_ids = active_rows
        .iter()
        .map(|(column_id, _, _)| *column_id)
        .collect::<HashSet<_>>();
    let mut active_sibling_names = active_rows
        .iter()
        .map(|(_, name, parent_column)| {
            (
                parent_column.map(|parent| parent.0),
                name.to_ascii_lowercase(),
            )
        })
        .collect::<HashSet<_>>();
    for column in &add.columns {
        if active_ids.contains(&column.column_id) {
            return Err(StorageError::Backend(format!(
                "DuckLake ADD COLUMN id {} already exists in table {}",
                column.column_id.0, add.table_id.0
            )));
        }
        if let Some(parent_column) = column.parent_column
            && !active_ids.contains(&parent_column)
            && !seen_column_ids.contains(&parent_column)
        {
            return Err(StorageError::Catalog(CatalogError::NotFound(format!(
                "parent column {} in table {}",
                parent_column.0, add.table_id.0
            ))));
        }
        if !active_sibling_names.insert((
            column.parent_column.map(|parent| parent.0),
            column.column_name.to_ascii_lowercase(),
        )) {
            return Err(StorageError::Backend(format!(
                "DuckLake ADD COLUMN name {} already exists in table {}",
                column.column_name, add.table_id.0
            )));
        }
    }

    i64::try_from(add.columns.len())
        .map_err(|_| StorageError::Backend("DuckLake ADD COLUMN count is out of range".into()))
}

#[cfg(any(
    feature = "sqlite",
    feature = "duckdb",
    feature = "postgres",
    feature = "quack"
))]
pub(crate) fn plan_add_columns_commit(
    add: &AddColumnCommit,
    snapshot: CatalogSnapshotIds,
    table_exists: bool,
    active_rows: &[(
        ducklake_catalog::ColumnId,
        String,
        Option<ducklake_catalog::ColumnId>,
    )],
    table_column_stats_exists: bool,
) -> Result<MetadataMutationPlan> {
    validate_add_column_commit(add)?;
    ensure_expected_snapshot("add-column", add.expected_snapshot, snapshot.snapshot_id)?;
    if !table_exists {
        return Err(StorageError::Catalog(CatalogError::NotFound(format!(
            "table {}",
            add.table_id.0
        ))));
    }
    let allocated_columns =
        validate_add_column_catalog_state(add, snapshot.next_catalog_id, active_rows)?;
    if !add.table_column_stats.is_empty() && !table_column_stats_exists {
        return Err(StorageError::Backend(
            "DuckLake ADD COLUMN stats require ducklake_table_column_stats".into(),
        ));
    }

    let new_snapshot = snapshot
        .snapshot_id
        .checked_add(1)
        .ok_or_else(|| StorageError::Backend("DuckLake snapshot id overflow".into()))?;
    let new_schema_version = snapshot
        .schema_version
        .checked_add(1)
        .ok_or_else(|| StorageError::Backend("DuckLake schema version overflow".into()))?;
    let next_catalog_id = snapshot
        .next_catalog_id
        .checked_add(allocated_columns)
        .ok_or_else(|| {
            StorageError::Backend("DuckLake ADD COLUMN next_catalog_id overflow".into())
        })?;
    Ok(MetadataMutationPlan {
        table_id: add.table_id,
        new_snapshot,
        new_schema_version,
        next_catalog_id,
        next_file_id: snapshot.next_file_id,
        changes_made: format!("altered_table:{}", add.table_id.0),
        commit_message: add.commit_message.clone(),
    })
}
