(iceberg) branch main updated: Spark 4.1: Control merge schema evolution by table property (#15825)

[am...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

amoghj pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git


The following commit(s) were added to refs/heads/main by this push:
     new 05d7ece42a Spark 4.1: Control merge schema evolution by table property (#15825)
05d7ece42a is described below

commit 05d7ece42acccc041f711dcac0dbb03a3c127396
Author: Szehon Ho <sz...@gmail.com>
AuthorDate: Wed Apr 1 09:19:04 2026 -0700

    Spark 4.1: Control merge schema evolution by table property (#15825)
    
    * Spark: Control merge schema evolution by table property
    
    Add a new table property write.spark.auto-schema-evolution (default true)
    that controls whether the AUTOMATIC_SCHEMA_EVOLUTION capability is
    reported to Spark. When set to false, Spark's MERGE WITH SCHEMA
    EVOLUTION no longer evolves the target table schema.
    
    Also add a guard in SparkWriteBuilder to reject mergeSchema write option
    when the property is disabled.
    
    * Remove unnecessary validation from SparkWriteBuilder
    
    The capability removal in SparkTable is sufficient to control schema
    evolution. The mergeSchema write option path already requires
    accept-any-schema, making a second gate redundant.
    
    * Address review comments
    
    - Rename property to write.spark.auto-schema-evolution.enabled
    - Rename caps to tableCapabilities in computeCapabilities
    - Add explicit = in ALTER TABLE SET TBLPROPERTIES test SQL
---
 .../java/org/apache/iceberg/TableProperties.java   |  4 ++
 .../spark/extensions/TestMergeSchemaEvolution.java | 45 ++++++++++++++++++++++
 .../apache/iceberg/spark/source/SparkTable.java    | 32 +++++++++++----
 3 files changed, 73 insertions(+), 8 deletions(-)

diff --git a/core/src/main/java/org/apache/iceberg/TableProperties.java b/core/src/main/java/org/apache/iceberg/TableProperties.java
index 1f778984af..71991f633d 100644
--- a/core/src/main/java/org/apache/iceberg/TableProperties.java
+++ b/core/src/main/java/org/apache/iceberg/TableProperties.java
@@ -346,6 +346,10 @@ public class TableProperties {
   public static final String SPARK_WRITE_ACCEPT_ANY_SCHEMA = "write.spark.accept-any-schema";
   public static final boolean SPARK_WRITE_ACCEPT_ANY_SCHEMA_DEFAULT = false;
 
+  public static final String SPARK_WRITE_AUTO_SCHEMA_EVOLUTION =
+      "write.spark.auto-schema-evolution.enabled";
+  public static final boolean SPARK_WRITE_AUTO_SCHEMA_EVOLUTION_DEFAULT = true;
+
   public static final String SPARK_WRITE_ADVISORY_PARTITION_SIZE_BYTES =
       "write.spark.advisory-partition-size-bytes";
 
diff --git a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeSchemaEvolution.java b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeSchemaEvolution.java
index d760d56b7a..782321b588 100644
--- a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeSchemaEvolution.java
+++ b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeSchemaEvolution.java
@@ -19,10 +19,12 @@
 package org.apache.iceberg.spark.extensions;
 
 import static org.apache.spark.sql.functions.col;
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assumptions.assumeThat;
 
 import java.util.Map;
 import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.junit.jupiter.api.AfterEach;
@@ -260,6 +262,49 @@ public class TestMergeSchemaEvolution extends SparkRowLevelOperationsTestBase {
         sql("SELECT id, value FROM %s ORDER BY id", selectTarget()));
   }
 
+  @TestTemplate
+  public void testMergeWithSchemaEvolutionDisabledByTableProperty() {
+    assumeThat(branch).as("Schema evolution does not work for branches currently").isNull();
+
+    createAndInitTable(
+        "id INT, dep STRING",
+        "{ \"id\": 1, \"dep\": \"hr\" }\n" + "{ \"id\": 2, \"dep\": \"software\" }");
+
+    sql(
+        "ALTER TABLE %s SET TBLPROPERTIES ('%s' = 'false')",
+        tableName, TableProperties.SPARK_WRITE_AUTO_SCHEMA_EVOLUTION);
+
+    createOrReplaceView(
+        "source",
+        "id INT, dep STRING, salary INT",
+        "{ \"id\": 1, \"dep\": \"hr\", \"salary\": 100 }\n"
+            + "{ \"id\": 3, \"dep\": \"finance\", \"salary\": 300 }");
+
+    sql(
+        "MERGE WITH SCHEMA EVOLUTION INTO %s AS t USING source AS s "
+            + "ON t.id == s.id "
+            + "WHEN MATCHED THEN "
+            + "  UPDATE SET * "
+            + "WHEN NOT MATCHED THEN "
+            + "  INSERT *",
+        commitTarget());
+
+    // Schema should NOT be evolved - 'salary' column should not be added
+    assertThat(sql("SELECT * FROM %s", selectTarget()).get(0).length)
+        .as("Table should still have only 2 columns (id, dep)")
+        .isEqualTo(2);
+
+    ImmutableList<Object[]> expectedRows =
+        ImmutableList.of(
+            row(1, "hr"), // updated without salary
+            row(2, "software"), // kept
+            row(3, "finance")); // new without salary
+    assertEquals(
+        "Should have expected rows without schema evolution",
+        expectedRows,
+        sql("SELECT id, dep FROM %s ORDER BY id", selectTarget()));
+  }
+
   @Override
   protected Map<String, String> extraTableProperties() {
     return Map.of();
diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
index ae3c0ce0c8..07db8c4ed3 100644
--- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
+++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
@@ -83,20 +83,14 @@ public class SparkTable extends BaseSparkTable
 
   private static final Logger LOG = LoggerFactory.getLogger(SparkTable.class);
 
-  private static final Set<TableCapability> CAPABILITIES =
+  private static final Set<TableCapability> BASE_CAPABILITIES =
       ImmutableSet.of(
-          TableCapability.AUTOMATIC_SCHEMA_EVOLUTION,
           TableCapability.BATCH_READ,
           TableCapability.BATCH_WRITE,
           TableCapability.MICRO_BATCH_READ,
           TableCapability.STREAMING_WRITE,
           TableCapability.OVERWRITE_BY_FILTER,
           TableCapability.OVERWRITE_DYNAMIC);
-  private static final Set<TableCapability> CAPABILITIES_WITH_ACCEPT_ANY_SCHEMA =
-      ImmutableSet.<TableCapability>builder()
-          .addAll(CAPABILITIES)
-          .add(TableCapability.ACCEPT_ANY_SCHEMA)
-          .build();
 
   private final Schema schema; // effective schema (not necessarily current table schema)
   private final Snapshot snapshot; // always set unless table is empty
@@ -133,7 +127,7 @@ public class SparkTable extends BaseSparkTable
     this.snapshot = snapshot;
     this.branch = branch;
     this.timeTravel = timeTravel;
-    this.capabilities = acceptAnySchema(table) ? CAPABILITIES_WITH_ACCEPT_ANY_SCHEMA : CAPABILITIES;
+    this.capabilities = computeCapabilities(table);
   }
 
   public SparkTable copyWithBranch(String newBranch) {
@@ -353,6 +347,21 @@ public class SparkTable extends BaseSparkTable
     return new SparkTable(table, snapshotId, timeTravel);
   }
 
+  private static Set<TableCapability> computeCapabilities(Table table) {
+    ImmutableSet.Builder<TableCapability> tableCapabilities = ImmutableSet.builder();
+    tableCapabilities.addAll(BASE_CAPABILITIES);
+
+    if (autoSchemaEvolution(table)) {
+      tableCapabilities.add(TableCapability.AUTOMATIC_SCHEMA_EVOLUTION);
+    }
+
+    if (acceptAnySchema(table)) {
+      tableCapabilities.add(TableCapability.ACCEPT_ANY_SCHEMA);
+    }
+
+    return tableCapabilities.build();
+  }
+
   private static boolean acceptAnySchema(Table table) {
     return PropertyUtil.propertyAsBoolean(
         table.properties(),
@@ -360,6 +369,13 @@ public class SparkTable extends BaseSparkTable
         TableProperties.SPARK_WRITE_ACCEPT_ANY_SCHEMA_DEFAULT);
   }
 
+  private static boolean autoSchemaEvolution(Table table) {
+    return PropertyUtil.propertyAsBoolean(
+        table.properties(),
+        TableProperties.SPARK_WRITE_AUTO_SCHEMA_EVOLUTION,
+        TableProperties.SPARK_WRITE_AUTO_SCHEMA_EVOLUTION_DEFAULT);
+  }
+
   // returns latest snapshot for branch or current snapshot if branch is yet to be created
   private static Snapshot determineLatestSnapshot(Table table, String branch) {
     if (branch != null && table.refs().containsKey(branch)) {