[spark] Fix sort compact with partition filter (#6371)

Author: Zouxxyy

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

@@ -263,7 +263,7 @@ private boolean execute(
         }
         boolean fullCompact = compactStrategy.equalsIgnoreCase(FULL);
         RowType partitionType = table.schema().logicalPartitionType();
-        Predicate filter =
+        Predicate partitionFilter =
                 condition == null
                         ? null
                         : ExpressionUtils.convertConditionToPaimonPredicate(
@@ -273,7 +273,7 @@ private boolean execute(
                                         false)
                                 .getOrElse(null);
         PartitionPredicate partitionPredicate =
-                PartitionPredicate.fromPredicate(partitionType, filter);
+                PartitionPredicate.fromPredicate(partitionType, partitionFilter);
 
         if (orderType.equals(OrderType.NONE)) {
             JavaSparkContext javaSparkContext = new JavaSparkContext(spark().sparkContext());
@@ -302,7 +302,8 @@ private boolean execute(
         } else {
             switch (bucketMode) {
                 case BUCKET_UNAWARE:
-                    sortCompactUnAwareBucketTable(table, orderType, sortColumns, relation, filter);
+                    sortCompactUnAwareBucketTable(
+                            table, orderType, sortColumns, relation, partitionFilter);
                     break;
                 default:
                     throw new UnsupportedOperationException(
@@ -521,10 +522,10 @@ private void sortCompactUnAwareBucketTable(
             OrderType orderType,
             List<String> sortColumns,
             DataSourceV2Relation relation,
-            @Nullable Predicate filter) {
+            @Nullable Predicate partitionFilter) {
         SnapshotReader snapshotReader = table.newSnapshotReader();
-        if (filter != null) {
-            snapshotReader.withFilter(filter);
+        if (partitionFilter != null) {
+            snapshotReader.withPartitionFilter(partitionFilter);
         }
         Map<BinaryRow, DataSplit[]> packedSplits = packForSort(snapshotReader.read().dataSplits());
         TableSorter sorter = TableSorter.getSorter(table, orderType, sortColumns);

paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/procedure/CompactProcedureTestBase.scala

@@ -343,6 +343,19 @@ abstract class CompactProcedureTestBase extends PaimonSparkTestBase with StreamT
     }
   }
 
+  test("Paimon Procedure: sort compact with partition filter") {
+    withTable("t") {
+      sql("CREATE TABLE t (a INT, pt INT) PARTITIONED BY (pt)")
+      sql("INSERT INTO t VALUES (1, 1)")
+      sql("INSERT INTO t VALUES (2, 1)")
+      sql(
+        "CALL sys.compact(table => 't', order_strategy => 'order', where => 'pt = 1', order_by => 'a')")
+      val table = loadTable("t")
+      assert(table.latestSnapshot().get().commitKind.equals(CommitKind.OVERWRITE))
+      checkAnswer(sql("SELECT * FROM t ORDER BY a"), Seq(Row(1, 1), Row(2, 1)))
+    }
+  }
+
   test("Paimon Procedure: compact for pk") {
     failAfter(streamingTimeout) {
       withTempDir {