(paimon) branch master updated: [hotfix] Fix dedicated compaction (#5146)

[lzljs3620320]

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

lzljs3620320 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 8f7625e7d7 [hotfix] Fix dedicated compaction (#5146)
8f7625e7d7 is described below

commit 8f7625e7d7d0cce555ad6b0335b07fb35b5a34bf
Author: Yubin Li <lixin58...@163.com>
AuthorDate: Tue Feb 25 12:18:32 2025 +0800

    [hotfix] Fix dedicated compaction (#5146)
---
 .../org/apache/paimon/flink/source/CompactorSourceBuilder.java    | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
index 564f9d529a..1f6b97edd2 100644
--- 
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
+++ 
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/CompactorSourceBuilder.java
@@ -86,18 +86,18 @@ public class CompactorSourceBuilder {
     }
 
     private Source<RowData, ?, ?> buildSource(CompactBucketsTable 
compactBucketsTable) {
+        compactBucketsTable =
+                compactBucketsTable.copy(
+                        isContinuous ? streamingCompactOptions() : 
batchCompactOptions());
         ReadBuilder readBuilder =
                 
compactBucketsTable.newReadBuilder().withFilter(partitionPredicate);
-        if (compactBucketsTable.coreOptions().manifestDeleteFileDropStats()) {
+        if 
(CoreOptions.fromMap(table.options()).manifestDeleteFileDropStats()) {
             readBuilder = readBuilder.dropStats();
         }
         if (isContinuous) {
-            compactBucketsTable = 
compactBucketsTable.copy(streamingCompactOptions());
             return new ContinuousFileStoreSource(readBuilder, 
compactBucketsTable.options(), null);
         } else {
-            compactBucketsTable = 
compactBucketsTable.copy(batchCompactOptions());
             Options options = 
compactBucketsTable.coreOptions().toConfiguration();
-
             return new StaticFileStoreSource(
                     readBuilder,
                     null,