[spark] Reject ALTER TABLE REPLACE COLUMNS to avoid silent data corruption

Spark translates REPLACE COLUMNS into a DeleteColumn + AddColumn batch.
Re-adding columns assigns new field ids while existing data files keep the
old ids, so same-named columns are read back as null. Detect this pattern
and throw UnsupportedOperationException instead.

Author: huangxiaopingRD

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/SparkCatalog.java

@@ -350,9 +350,16 @@ public SparkTable loadTable(Identifier ident, long timestamp) throws NoSuchTable
     @Override
     public org.apache.spark.sql.connector.catalog.Table alterTable(
             Identifier ident, TableChange... changes) throws NoSuchTableException {
-        List<SchemaChange> schemaChanges =
-                Arrays.stream(changes).map(this::toSchemaChange).collect(Collectors.toList());
         try {
+            if (isReplaceColumns(changes)) {
+                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.");
+            }
+
+            List<SchemaChange> schemaChanges =
+                    Arrays.stream(changes).map(this::toSchemaChange).collect(Collectors.toList());
             catalog.alterTable(toIdentifier(ident, catalogName), schemaChanges, false);
             return loadTable(ident);
         } catch (Catalog.TableNotExistException e) {
@@ -362,6 +369,36 @@ public org.apache.spark.sql.connector.catalog.Table alterTable(
         }
     }
 
+    /**
+     * Detects whether the given changes originate from an {@code ALTER TABLE ... REPLACE COLUMNS}
+     * statement.
+     *
+     * <p>Spark translates {@code REPLACE COLUMNS} into a batch that drops every existing column and
+     * re-adds the new set, i.e. a combination of {@link TableChange.DeleteColumn} and {@link
+     * TableChange.AddColumn} only. Other column changes such as rename or type update are never
+     * produced by {@code REPLACE COLUMNS}, so we match exclusively on these two types to avoid
+     * mistaking a legitimate mixed batch (e.g. a programmatic DROP + RENAME) for a replace.
+     *
+     * <p>This operation must be rejected because re-adding columns assigns brand-new field ids
+     * while existing data files keep the old ids; same-named columns would then be treated as new
+     * columns and read back as null, silently corrupting data.
+     */
+    private boolean isReplaceColumns(TableChange[] changes) {
+        boolean hasDeleteColumn = false;
+        boolean hasAddColumn = false;
+        for (TableChange change : changes) {
+            if (change instanceof TableChange.DeleteColumn) {
+                hasDeleteColumn = true;
+            } else if (change instanceof TableChange.AddColumn) {
+                hasAddColumn = true;
+            } else {
+                return false;
+            }
+        }
+
+        return hasDeleteColumn && hasAddColumn;
+    }
+
     @Override
     public org.apache.spark.sql.connector.catalog.Table createTable(
             Identifier ident,

paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkSchemaEvolutionITCase.java

@@ -248,6 +248,23 @@ public void testDropColumns() {
                 .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(