Compaction Operator parallelism in paimon MUST be one (#3758)

This closes #3758.
apache · Jul 24, 2024 · 094e925 · 094e925
1 parent 10a982e
commit 094e925
Showing 1 changed file with 14 additions and 7 deletions.
diff --git a/...flink-common/src/main/java/org/apache/paimon/flink/source/BucketUnawareCompactSource.java b/...flink-common/src/main/java/org/apache/paimon/flink/source/BucketUnawareCompactSource.java
@@ -24,6 +24,7 @@
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.table.FileStoreTable;
 import org.apache.paimon.table.source.EndOfScanException;
+import org.apache.paimon.utils.Preconditions;
 
 import org.apache.flink.api.connector.source.Boundedness;
 import org.apache.flink.configuration.Configuration;
@@ -74,6 +75,9 @@ public BucketUnawareCompactSource(
     @Override
     public void open(Configuration parameters) throws Exception {
         compactionCoordinator = new AppendOnlyTableCompactionCoordinator(table, streaming, filter);
+        Preconditions.checkArgument(
+                this.getRuntimeContext().getNumberOfParallelSubtasks() == 1,
+                "Compaction Operator parallelism in paimon MUST be one.");
     }
 
     @Override
@@ -120,12 +124,15 @@ public static DataStreamSource<AppendOnlyCompactionTask> buildSource(
             String tableIdentifier) {
         final StreamSource<AppendOnlyCompactionTask, BucketUnawareCompactSource> sourceOperator =
                 new StreamSource<>(source);
-        return new DataStreamSource<>(
-                env,
-                new CompactionTaskTypeInfo(),
-                sourceOperator,
-                false,
-                COMPACTION_COORDINATOR_NAME + " : " + tableIdentifier,
-                streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED);
+        return (DataStreamSource<AppendOnlyCompactionTask>)
+                new DataStreamSource<>(
+                                env,
+                                new CompactionTaskTypeInfo(),
+                                sourceOperator,
+                                false,
+                                COMPACTION_COORDINATOR_NAME + " : " + tableIdentifier,
+                                streaming ? Boundedness.CONTINUOUS_UNBOUNDED : Boundedness.BOUNDED)
+                        .setParallelism(1)
+                        .setMaxParallelism(1);
     }
 }