[spark] Throw exception when to sort compact data evolution table

Author: JingsongLi

paimon-core/src/main/java/org/apache/paimon/schema/SchemaValidation.java

@@ -599,6 +599,9 @@ private static void validateRowTracking(TableSchema schema, CoreOptions options)
             checkArgument(
                     !options.deletionVectorsEnabled(),
                     "Data evolution config must disabled with deletion-vectors.enabled");
+            checkArgument(
+                    !options.clusteringIncrementalEnabled(),
+                    "Data evolution config must disabled with clustering.incremental");
         }
 
         Pair<RowType, RowType> normalAndBlobType = BlobType.splitBlob(schema.logicalRowType());

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactAction.java

@@ -144,7 +144,6 @@ protected boolean buildImpl() throws Exception {
         if (fileStoreTable.coreOptions().bucket() == BucketMode.POSTPONE_BUCKET) {
             buildForPostponeBucketCompaction(env, fileStoreTable, isStreaming);
         } else if (fileStoreTable.bucketMode() == BucketMode.BUCKET_UNAWARE) {
-
             if (fileStoreTable.coreOptions().dataEvolutionEnabled()) {
                 buildForDataEvolutionTableCompact(env, fileStoreTable, isStreaming);
             } else if (fileStoreTable.coreOptions().clusteringIncrementalEnabled()) {

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/SortCompactAction.java

@@ -56,7 +56,6 @@ public SortCompactAction(
             Map<String, String> catalogConfig,
             Map<String, String> tableConf) {
         super(database, tableName, catalogConfig, tableConf);
-
         table = table.copy(Collections.singletonMap(CoreOptions.WRITE_ONLY.key(), "true"));
     }
 
@@ -77,6 +76,10 @@ public void build() throws Exception {
         }
         FileStoreTable fileStoreTable = (FileStoreTable) table;
 
+        if (fileStoreTable.coreOptions().dataEvolutionEnabled()) {
+            throw new UnsupportedOperationException("Data Evolution table cannot be sorted!");
+        }
+
         if (fileStoreTable.bucketMode() != BucketMode.BUCKET_UNAWARE
                 && fileStoreTable.bucketMode() != BucketMode.HASH_DYNAMIC) {
             throw new IllegalArgumentException("Sort Compact only supports bucket=-1 yet.");

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactActionITCase.java

@@ -849,7 +849,7 @@ public void testDataEvolutionTableCompact() throws Exception {
 
         FileStoreTable table =
                 prepareTable(
-                        Arrays.asList("k"),
+                        Collections.singletonList("k"),
                         Collections.emptyList(),
                         Collections.emptyList(),
                         tableOptions);

paimon-spark/paimon-spark-common/src/main/java/org/apache/paimon/spark/procedure/CompactProcedure.java

@@ -608,6 +608,9 @@ private void sortCompactUnAwareBucketTable(
             List<String> sortColumns,
             DataSourceV2Relation relation,
             @Nullable PartitionPredicate partitionPredicate) {
+        if (table.coreOptions().dataEvolutionEnabled()) {
+            throw new UnsupportedOperationException("Data Evolution table cannot be sorted!");
+        }
         SnapshotReader snapshotReader = table.newSnapshotReader();
         if (partitionPredicate != null) {
             snapshotReader.withPartitionFilter(partitionPredicate);

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/RowTrackingTestBase.scala

@@ -651,6 +651,20 @@ abstract class RowTrackingTestBase extends PaimonSparkTestBase {
     }
   }
 
+  test("Data Evolution: compact sort throw exception") {
+    withTable("s", "t") {
+      sql(
+        s"CREATE TABLE t (id INT, b INT, c INT) TBLPROPERTIES ('row-tracking.enabled' = 'true', 'data-evolution.enabled' = 'true')")
+      sql("INSERT INTO t VALUES (1, 1, 1)")
+      assert(
+        intercept[Exception](
+          sql("CALL sys.compact(table => 't', order_strategy => 'order', order_by => 'id')")
+            .collect()).getMessage
+          .contains("Data Evolution table cannot be sorted!"))
+
+    }
+  }
+
   test("Data Evolution: test global indexed column update action -- throw error") {
     withTable("T") {
       spark.sql("""