You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/10/05 12:45:38 UTC
(paimon) branch master updated: [flink] Reject a non-positive scan.split-enumerator.batch-size (#10369)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi 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 ca1fdc5724 [flink] Reject a non-positive scan.split-enumerator.batch-size (#10369)
ca1fdc5724 is described below
commit ca1fdc57248d3a77b7ac9ce6d4ca884c272e01b7
Author: jackylee <qc...@gmail.com>
AuthorDate: Mon Oct 5 20:45:32 2026 +0800
[flink] Reject a non-positive scan.split-enumerator.batch-size (#10369)
---
.../flink/source/assigners/PreAssignSplitAssigner.java | 6 ++++++
.../flink/source/assigners/PreAssignSplitAssignerTest.java | 14 ++++++++++++++
2 files changed, 20 insertions(+)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/PreAssignSplitAssigner.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/PreAssignSplitAssigner.java
index dbf1b65d6c..46e9fa20c7 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/PreAssignSplitAssigner.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/PreAssignSplitAssigner.java
@@ -18,6 +18,7 @@
package org.apache.paimon.flink.source.assigners;
+import org.apache.paimon.flink.FlinkConnectorOptions;
import org.apache.paimon.flink.source.DynamicPartitionFilteringInfo;
import org.apache.paimon.flink.source.FileStoreSourceSplit;
import org.apache.paimon.utils.BinPacking;
@@ -44,6 +45,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import static org.apache.paimon.flink.utils.TableScanUtils.getSnapshotId;
+import static org.apache.paimon.utils.Preconditions.checkArgument;
/**
* Pre-calculate which splits each task should process according to the weight or given
@@ -142,6 +144,10 @@ public class PreAssignSplitAssigner implements SplitAssigner {
Collection<FileStoreSourceSplit> splits,
SerializableFunction<FileStoreSourceSplit, Long> weightFunc,
@Nullable SerializableFunction<FileStoreSourceSplit, ?> groupFunc) {
+ checkArgument(
+ splitBatchSize > 0,
+ "%s must be positive.",
+ FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_BATCH_SIZE.key());
this.splitBatchSize = splitBatchSize;
this.parallelism = parallelism;
this.splits = splits;
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/assigners/PreAssignSplitAssignerTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/assigners/PreAssignSplitAssignerTest.java
index ea1e54873b..8ee140f972 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/assigners/PreAssignSplitAssignerTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/assigners/PreAssignSplitAssignerTest.java
@@ -47,6 +47,7 @@ import java.util.stream.Collectors;
import static org.apache.paimon.io.DataFileTestUtils.row;
import static org.apache.paimon.utils.Preconditions.checkArgument;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests {@link PreAssignSplitAssigner} pruning over {@link QueryAuthSplit}. */
public class PreAssignSplitAssignerTest {
@@ -124,6 +125,19 @@ public class PreAssignSplitAssignerTest {
assertThat(assigner.getNext(0, null)).isEmpty();
}
+ @Test
+ public void testNonPositiveSplitBatchSizeRejected() {
+ // A non-positive scan.split-enumerator.batch-size makes PreAssignSplitAssigner#getNext's
+ // "assignment.size() < splitBatchSize" loop never run, so the enumerator hands out no
+ // splits and a bounded read silently returns zero rows. Reject it at construction.
+ Collection<FileStoreSourceSplit> splits = Collections.singletonList(partitionSplit(1));
+ for (int invalid : new int[] {0, -1}) {
+ assertThatThrownBy(() -> new PreAssignSplitAssigner(invalid, 1, splits))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("scan.split-enumerator.batch-size must be positive.");
+ }
+ }
+
private static SplitAssigner pruningAssigner(Collection<FileStoreSourceSplit> splits) {
DynamicPartitionFilteringInfo filteringInfo =
new DynamicPartitionFilteringInfo(