delta_kernel 0.29.0

//! Builder for ALTER TABLE (schema evolution) transactions.
//!
//! This module contains [`AlterTableTransactionBuilder`], which uses a type-state pattern to
//! enforce valid operation chaining at compile time.
//!
//! # Type States
//!
//! - [`Ready`]: Initial state. Operations are available, but `build()` is not (at least one
//!   operation is required).
//! - [`Modifying`]: After any chainable schema operation. More ops can be chained, and `build()` is
//!   available. See [`AlterTableTransactionBuilder<Modifying>`] for ops.
//!
//! # Transitions
//!
//! Each `impl` block below is gated by a state bound and documents which operations that
//! state enables. Chainable schema operations live on `impl<S: Chainable>` and transition
//! the builder to a chainable state; `build()` lives on states that are buildable.
//!
//! ```ignore
//! // Allowed: at least one op queued before build().
//! snapshot.alter_table().add_column(field).build(engine, committer)?;
//!
//! // Not allowed: build() is not defined on Ready (no ops queued).
//! snapshot.alter_table().build(engine, committer)?;  // compile error
//! ```

use std::sync::Arc;

use delta_kernel_derive::internal_api;

use crate::committer::Committer;
use crate::expressions::ColumnName;
use crate::schema::StructField;
use crate::snapshot::SnapshotRef;
use crate::table_features::{Operation, TableFeature};
use crate::transaction::alter_table::AlterTableTransaction;
use crate::transaction::schema_evolution::{evolve_table_config, SchemaOperation};
use crate::utils::PhantomType;
use crate::{DeltaResult, Engine, Error};

/// Initial state: `build()` is not yet available (at least one operation is required).
/// See [`Chainable`] for the operations available on this state.
pub struct Ready;

/// State after at least one operation has been added. `build()` is available.
/// See [`Chainable`] for the operations available on this state.
pub struct Modifying;

/// Marker trait for builder states that accept chainable schema operations. Grouping states
/// under one bound lets each op (like `add_column`) live on a single `impl<S: Chainable>`
/// block -- chainable states share the body rather than duplicating it per state.
///
/// Sealed: external types cannot implement this, keeping the set of chainable states closed.
pub trait Chainable: sealed::Sealed {}
impl Chainable for Ready {}
impl Chainable for Modifying {}

mod sealed {
    pub trait Sealed {}
    impl Sealed for super::Ready {}
    impl Sealed for super::Modifying {}
}

/// Builder for constructing an [`AlterTableTransaction`] with schema evolution operations.
///
/// Uses a type-state pattern (`S`) to enforce at compile time:
/// - At least one schema operation must be queued before `build()` is callable.
/// - Only operations valid for the current state can be chained. This will disallow incompatible
///   chaining.
pub struct AlterTableTransactionBuilder<S = Ready> {
    snapshot: SnapshotRef,
    operations: Vec<SchemaOperation>,
    correlation_id: Option<Arc<str>>,
    // PhantomType marker for builder state (Ready or Modifying).
    // Zero-sized; only affects which methods are available at compile time.
    _state: PhantomType<S>,
}

impl<S> AlterTableTransactionBuilder<S> {
    // Reconstructs the builder with a different PhantomType marker, changing which methods
    // are available at compile time (e.g. Ready -> Modifying enables `build()`). All real
    // fields are moved as-is; only the zero-sized type state changes.
    //
    // `T` (distinct from the struct's `S`) lets the caller pick the target state:
    // `self.transition::<Modifying>()` returns `AlterTableTransactionBuilder<Modifying>`.
    fn transition<T>(self) -> AlterTableTransactionBuilder<T> {
        AlterTableTransactionBuilder {
            snapshot: self.snapshot,
            operations: self.operations,
            correlation_id: self.correlation_id,
            _state: PhantomType::default(),
        }
    }

    /// Attach an opaque, caller-supplied correlation id for joining the alter-table commit's metric
    /// events to the caller's own request or operation id. An empty id is treated as unset.
    pub fn with_correlation_id(mut self, correlation_id: impl Into<Arc<str>>) -> Self {
        self.correlation_id = Some(correlation_id.into()).filter(|id| !id.is_empty());
        self
    }
}

impl AlterTableTransactionBuilder<Ready> {
    /// Create a new builder from a snapshot.
    pub(crate) fn new(snapshot: SnapshotRef) -> Self {
        AlterTableTransactionBuilder {
            snapshot,
            operations: Vec::new(),
            correlation_id: None,
            _state: PhantomType::default(),
        }
    }
}

impl<S: Chainable> AlterTableTransactionBuilder<S> {
    /// Add a new top-level column to the table schema.
    ///
    /// The field must not already exist in the schema (case-insensitive). The field must be
    /// nullable because existing data files do not contain this column and will read NULL for it.
    /// On column-mapping tables, Kernel assigns or preserves column-mapping IDs and physical names
    /// for the added field.
    ///
    /// These constraints are validated during [`build()`](AlterTableTransactionBuilder::build).
    pub fn add_column(mut self, field: StructField) -> AlterTableTransactionBuilder<Modifying> {
        self.operations
            .push(SchemaOperation::add_column(None, field));
        self.transition()
    }

    /// Change a column's nullability from NOT NULL to nullable. If the column is already
    /// nullable, the op is a no-op but still generates a commit.
    ///
    /// Note: this matches Spark's behavior.
    pub fn set_nullable(mut self, column: ColumnName) -> AlterTableTransactionBuilder<Modifying> {
        self.operations
            .push(SchemaOperation::SetNullable { column });
        self.transition()
    }

    /// Add a new column or nested field to the table schema.
    ///
    /// `parent` identifies the struct that will contain `field`. An empty parent targets the
    /// table's root schema; segments may traverse nested structs, array elements, map keys, and
    /// map values.
    ///
    /// The added field must be nullable (existing data files lack the column and will read NULL),
    /// must not be a metadata column and must not collide case-insensitively with a
    /// sibling in the target struct. `parent` must resolve to a struct.
    ///
    /// With column mapping enabled, existing IDs and physical names are preserved and missing
    /// annotations are assigned.
    ///
    /// These constraints are validated during [`build()`](AlterTableTransactionBuilder::build).
    #[internal_api]
    pub(crate) fn add_column_at(
        mut self,
        parent: ColumnName,
        field: StructField,
    ) -> AlterTableTransactionBuilder<Modifying> {
        self.operations
            .push(SchemaOperation::add_column(parent, field));
        self.transition()
    }
}

impl AlterTableTransactionBuilder<Modifying> {
    /// Validate and apply schema operations, then build the [`AlterTableTransaction`].
    ///
    /// This method:
    /// 1. Validates the table supports writes
    /// 2. Applies each operation sequentially against the evolving schema
    /// 3. Constructs new Metadata action with evolved schema
    /// 4. Builds the evolved table configuration
    /// 5. Creates the transaction
    ///
    /// # Errors
    ///
    /// - The table enables `icebergCompatV2`, `icebergCompatV3`, or `allowColumnDefaults`, which
    ///   ALTER TABLE does not yet support
    /// - Any individual operation fails validation (see per-method errors above)
    /// - CDF is enabled and the evolved schema contains a top-level column reserved for CDF
    /// - Table does not support writes (unsupported features)
    /// - The evolved schema requires protocol features not enabled on the table (e.g. adding a
    ///   `timestampNtz` column without the `timestampNtz` feature)
    pub fn build(
        self,
        _engine: &dyn Engine,
        committer: Box<dyn Committer>,
    ) -> DeltaResult<AlterTableTransaction> {
        let table_config = self.snapshot.table_configuration();
        // kernel doesn't currently support altering tables with these features
        let unsupported_iceberg_compat =
            [TableFeature::IcebergCompatV2, TableFeature::IcebergCompatV3]
                .into_iter()
                .find(|feature| table_config.is_feature_enabled(feature));
        if let Some(feature) = unsupported_iceberg_compat {
            return Err(Error::unsupported(format!(
                "ALTER TABLE is not yet supported on tables with {feature} enabled"
            )));
        }
        // TODO(#2630): Support ALTER TABLE on tables with column defaults.
        if table_config.is_feature_enabled(&TableFeature::AllowColumnDefaults) {
            return Err(Error::unsupported(
                "ALTER TABLE is not yet supported on tables with allowColumnDefaults enabled",
            ));
        }
        // Rejects writes to tables kernel can't safely commit to: writer version out of
        // kernel's supported range, unsupported writer features, or schemas with SQL-expression
        // invariants. Runs on the pre-alter snapshot; future ALTER variants that change the
        // protocol must also re-check this on the evolved `TableConfiguration`.
        table_config.ensure_operation_supported(Operation::Write)?;

        let evolved_table_config = evolve_table_config(table_config, self.operations)?;

        AlterTableTransaction::try_new_alter_table(
            self.snapshot,
            evolved_table_config,
            committer,
            self.correlation_id,
        )
    }
}