(paimon) branch master updated: [spark] Reject ALTER TABLE REPLACE COLUMNS to avoid silent data corruption (#8246)

[lz...@apache.org]

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

JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/master by this push:
     new 0e8ce76645 [spark] Reject ALTER TABLE REPLACE COLUMNS to avoid silent data corruption (#8246)
0e8ce76645 is described below

commit 0e8ce76645e4a660a7ed19850707b521fb567dd1
Author: huangxiaoping <17...@qq.com>
AuthorDate: Tue Jun 23 14:11:41 2026 +0800

    [spark] Reject ALTER TABLE REPLACE COLUMNS to avoid silent data corruption (#8246)
    
    Spark translates `ALTER TABLE ... REPLACE COLUMNS` into a batch that
    drops every existing column and re-adds the new set (a combination of
    `DeleteColumn` + `AddColumn`). For Paimon this is unsafe: re-adding
    columns assigns brand-new field ids while existing data files keep the
    old ids, so same-named columns are treated as new columns and read back
    as `null` — a silent data corruption.
    
    This PR detects that change pattern in `SparkCatalog.alterTable` and
    throws an `UnsupportedOperationException` with a clear message pointing
    users to `RENAME COLUMN` / `ALTER COLUMN TYPE` / `DROP COLUMN` / `ADD
    COLUMN` instead.
    
    The detection matches exclusively on `DeleteColumn` + `AddColumn` so a
    legitimate mixed batch (e.g. a programmatic DROP + RENAME) is not
    mistaken for a replace.
---
 .../paimon/spark/catalyst/analysis/PaimonAnalysis.scala | 10 ++++++++++
 .../apache/paimon/spark/SparkSchemaEvolutionITCase.java | 17 +++++++++++++++++
 2 files changed, 27 insertions(+)

diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
index c26eeedd3f..87bdfeffba 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
@@ -81,6 +81,16 @@ class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
         if d.resolved =>
       PaimonDropPartitions.validate(table, parts.asResolvedPartitionSpecs)
       d
+
+    case r: ReplaceColumns if r.resolved && isPaimonTable(r.table) =>
+      // Spark rewrites REPLACE COLUMNS into a batch that drops every existing column and re-adds
+      // the new set. Re-adding columns assigns brand-new field ids while existing data files keep
+      // the old ids, so same-named columns are read back as null, silently corrupting data. Reject
+      // it here, before the change batch reaches the catalog where it is indistinguishable from an
+      // ordinary drop+add.
+      throw new UnsupportedOperationException(
+        "ALTER TABLE ... REPLACE COLUMNS is not supported for Paimon tables. " +
+          "Please use RENAME COLUMN, ALTER COLUMN TYPE, DROP COLUMN, and ADD COLUMN instead.")
   }
 
   private def writeOptions(v2WriteCommand: V2WriteCommand): Map[String, String] = {
diff --git a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkSchemaEvolutionITCase.java b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkSchemaEvolutionITCase.java
index 7afe3c76cf..4e15b0880b 100644
--- a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkSchemaEvolutionITCase.java
+++ b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkSchemaEvolutionITCase.java
@@ -248,6 +248,23 @@ public class SparkSchemaEvolutionITCase extends SparkReadTestBase {
                 .contains(showCreateString("testDropColumns", "a INT NOT NULL"));
     }
 
+    @Test
+    public void testReplaceColumnsUnsupported() {
+        createTable("testReplaceColumnsUnsupported");
+
+        assertThatThrownBy(
+                        () ->
+                                spark.sql(
+                                        "ALTER TABLE testReplaceColumnsUnsupported REPLACE COLUMNS "
+                                                + "(a BIGINT, bb STRING, c STRING)"))
+                .satisfies(
+                        anyCauseMatches(
+                                UnsupportedOperationException.class,
+                                "ALTER TABLE ... REPLACE COLUMNS is not supported for Paimon tables. "
+                                        + "Please use RENAME COLUMN, ALTER COLUMN TYPE, DROP COLUMN, "
+                                        + "and ADD COLUMN instead."));
+    }
+
     @Test
     public void testDropPartitionKey() {
         spark.sql(