feat: derive ALTER TABLE isBlindAppend from per-op policy

Replaces the hardcoded isBlindAppend=false with a per-op fold over
SchemaOperation::is_blind_append(). All current ops (add/drop/rename/
drop-not-null) are blind-append, matching delta-spark. The flag is
emitted by apply_schema_operations via SchemaEvolutionResult.is_blind_append
alongside the operation string fold, no second iteration.

Author: william-ch-databricks

kernel/src/transaction/alter_table.rs

@@ -39,6 +39,7 @@ impl AlterTableTransaction {
         effective_table_config: TableConfiguration,
         committer: Box<dyn Committer>,
         operation: String,
+        is_blind_append: bool,
     ) -> DeltaResult<Self> {
         let span = tracing::info_span!(
             "txn",
@@ -66,10 +67,7 @@ impl AlterTableTransaction {
             data_change: false,
             shared_write_state: OnceLock::new(),
             engine_commit_info: None,
-            // TODO(#2446): match delta-spark's per-op isBlindAppend policy
-            // (ADD/DROP/DROP NOT NULL -> true, SET NOT NULL -> false). Hardcoded false for
-            // now: safe, but misses the true-case optimization delta-spark applies.
-            is_blind_append: false,
+            is_blind_append,
             dv_matched_files: vec![],
             physical_clustering_columns: None,
             _state: PhantomData,

kernel/src/transaction/builder/alter_table.rs

@@ -224,6 +224,7 @@ impl<S: Buildable> AlterTableTransactionBuilder<S> {
             schema: evolved_schema,
             new_max_column_id,
             operation,
+            is_blind_append,
         } = apply_schema_operations(
             schema,
             self.operations,
@@ -255,6 +256,7 @@ impl<S: Buildable> AlterTableTransactionBuilder<S> {
             evolved_table_config,
             committer,
             operation,
+            is_blind_append,
         )
     }
 }

kernel/src/transaction/mod.rs

@@ -197,6 +197,20 @@ pub trait SupportsDataFiles {}
 impl SupportsDataFiles for ExistingTable {}
 impl SupportsDataFiles for CreateTable {}
 
+/// Per-state constants consumed by validation. Implemented by every transaction state marker
+/// (`ExistingTable`, `CreateTable`, `AlterTable`); each supplies its own values, and validation
+/// reads `S::<CONST>` to branch on the current state.
+pub trait TransactionKind {
+    /// Whether the transaction produces only metadata changes (no data files). Metadata-only
+    /// commits have no data-file invariants to check during blind-append validation.
+    const IS_METADATA_ONLY: bool = false;
+}
+impl TransactionKind for ExistingTable {}
+impl TransactionKind for CreateTable {}
+impl TransactionKind for AlterTable {
+    const IS_METADATA_ONLY: bool = true;
+}
+
 /// A transaction represents an in-progress write to a table. After creating a transaction, changes
 /// to the table may be staged via the transaction methods before calling `commit` to commit the
 /// changes to the table.
@@ -340,7 +354,10 @@ impl<S> Transaction<S> {
         ),
         err
     )]
-    pub fn commit(self, engine: &dyn Engine) -> DeltaResult<CommitResult<S>> {
+    pub fn commit(self, engine: &dyn Engine) -> DeltaResult<CommitResult<S>>
+    where
+        S: TransactionKind,
+    {
         info!(
             num_add_files = self.add_files_metadata.len(),
             num_remove_files = self.remove_files_metadata.len(),
@@ -665,7 +682,10 @@ impl<S> Transaction<S> {
     /// Note: Domain metadata additions/removals are allowed; blind append only constrains
     /// data-file operations and read predicates. Conflict resolution determines whether
     /// metadata changes are problematic.
-    fn validate_blind_append_semantics(&self) -> DeltaResult<()> {
+    fn validate_blind_append_semantics(&self) -> DeltaResult<()>
+    where
+        S: TransactionKind,
+    {
         if !self.is_blind_append {
             return Ok(());
         }
@@ -675,6 +695,10 @@ impl<S> Transaction<S> {
                 "Blind append is not supported for create-table transactions",
             )
         );
+        // Metadata-only transactions (e.g. ALTER TABLE) have no data-file operations to check
+        if S::IS_METADATA_ONLY {
+            return Ok(());
+        }
         require!(
             !self.add_files_metadata.is_empty(),
             Error::invalid_transaction_state("Blind append requires at least one added data file")

kernel/src/transaction/schema_evolution.rs

@@ -47,6 +47,17 @@ impl SchemaOperation {
             SchemaOperation::SetNullable { .. } => "CHANGE COLUMNS",
         }
     }
+
+    /// Whether this operation is eligible to be committed as a blind append. A commit is a
+    /// blind append only if every operation in it is eligible.
+    pub(crate) fn is_blind_append(&self) -> bool {
+        match self {
+            SchemaOperation::AddColumn { .. } => true,
+            SchemaOperation::DropColumn { .. } => true,
+            SchemaOperation::RenameColumn { .. } => true,
+            SchemaOperation::SetNullable { .. } => true,
+        }
+    }
 }
 
 // Helper to transform a nested column. For each component in `remaining`, locates the
@@ -185,6 +196,9 @@ pub(crate) struct SchemaEvolutionResult {
     /// `"ADD COLUMNS"`, `"ADD COLUMNS + CHANGE COLUMNS"`). Used as the commit's
     /// `operation` field.
     pub operation: String,
+    /// Whether the commit is eligible to be marked `isBlindAppend=true`. True only if every
+    /// operation in the batch is eligible (see [`SchemaOperation::is_blind_append`]).
+    pub is_blind_append: bool,
 }
 
 /// Applies a sequence of schema operations to the given schema, returning the evolved schema.
@@ -223,12 +237,15 @@ pub(crate) fn apply_schema_operations(
     // Distinct operation names in first-occurrence order; joined into the commit's
     // operation string at the end.
     let mut op_names: Vec<&'static str> = Vec::new();
+    // Cumulative blind-append eligibility across the batch; false as soon as any op isn't blind.
+    let mut is_blind_append = true;
 
     for op in operations {
         let name = op.operation_name();
         if !op_names.contains(&name) {
             op_names.push(name);
         }
+        is_blind_append &= op.is_blind_append();
         match op {
             // Protocol feature checks for the field's data type (e.g. `timestampNtz`) happen
             // later when the caller builds a new TableConfiguration from the evolved schema --
@@ -488,6 +505,7 @@ pub(crate) fn apply_schema_operations(
         schema: evolved_schema.into(),
         new_max_column_id,
         operation: op_names.join(" + "),
+        is_blind_append,
     })
 }
 

kernel/tests/alter_table.rs

@@ -916,23 +916,34 @@ async fn rename_clustering_column_fails() -> Result<(), Box<dyn std::error::Erro
 // OPERATION STRING tests
 // ============================================================================
 
-// Reads `commitInfo.operation` from a commit JSON written to the local filesystem.
-fn read_commit_operation(table_path: &str, version: u64) -> String {
+// Reads the `commitInfo` object from a commit JSON written to the local filesystem.
+fn read_commit_info(table_path: &str, version: u64) -> serde_json::Value {
     let log_file = format!("{table_path}/_delta_log/{version:020}.json");
     let contents = std::fs::read_to_string(&log_file).expect("read log file");
     for line in contents.lines() {
         let action: serde_json::Value = serde_json::from_str(line).expect("parse action");
         if let Some(ci) = action.get("commitInfo") {
-            return ci
-                .get("operation")
-                .and_then(|v| v.as_str())
-                .expect("operation field")
-                .to_string();
+            return ci.clone();
         }
     }
     panic!("no commitInfo in commit {version}");
 }
 
+fn read_commit_operation(table_path: &str, version: u64) -> String {
+    read_commit_info(table_path, version)
+        .get("operation")
+        .and_then(|v| v.as_str())
+        .expect("operation field")
+        .to_string()
+}
+
+fn read_commit_is_blind_append(table_path: &str, version: u64) -> bool {
+    read_commit_info(table_path, version)
+        .get("isBlindAppend")
+        .and_then(|v| v.as_bool())
+        .expect("isBlindAppend field")
+}
+
 #[tokio::test]
 async fn alter_commit_operation_add_columns() -> DeltaResult<()> {
     let (_temp_dir, table_path, engine) = test_table_setup()?;
@@ -981,3 +992,34 @@ async fn alter_commit_operation_mixed_ops_joined_with_plus() -> DeltaResult<()>
     );
     Ok(())
 }
+
+#[tokio::test]
+async fn alter_commit_is_blind_append_true_for_add_columns() -> DeltaResult<()> {
+    let (_temp_dir, table_path, engine) = test_table_setup()?;
+    let snapshot =
+        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &[])?;
+    snapshot
+        .alter_table()
+        .add_column(StructField::nullable("email", DataType::STRING))
+        .build(engine.as_ref(), committer())?
+        .commit(engine.as_ref())?
+        .unwrap_committed();
+    assert!(read_commit_is_blind_append(&table_path, 1));
+    Ok(())
+}
+
+#[tokio::test]
+async fn alter_commit_is_blind_append_true_for_mixed_blind_ops() -> DeltaResult<()> {
+    let (_temp_dir, table_path, engine) = test_table_setup()?;
+    let snapshot =
+        create_table_and_load_snapshot(&table_path, simple_schema(), engine.as_ref(), &[])?;
+    snapshot
+        .alter_table()
+        .add_column(StructField::nullable("email", DataType::STRING))
+        .set_nullable(column_name!("id"))
+        .build(engine.as_ref(), committer())?
+        .commit(engine.as_ref())?
+        .unwrap_committed();
+    assert!(read_commit_is_blind_append(&table_path, 1));
+    Ok(())
+}