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 2024/03/15 02:24:28 UTC
(incubator-paimon) branch master updated: [flink] splitAssigner adds a thread-safe interface for obtaining the number of remaining splits. (#3011)
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/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 994849cdf [flink] splitAssigner adds a thread-safe interface for obtaining the number of remaining splits. (#3011)
994849cdf is described below
commit 994849cdff59b7acc449454bc29eceac2662d9cf
Author: liming.1018 <69...@users.noreply.github.com>
AuthorDate: Fri Mar 15 10:24:23 2024 +0800
[flink] splitAssigner adds a thread-safe interface for obtaining the number of remaining splits. (#3011)
---
.../flink/source/ContinuousFileSplitEnumerator.java | 2 +-
.../flink/source/assigners/AlignedSplitAssigner.java | 19 ++++++++++++++++---
.../assigners/DynamicPartitionPruningAssigner.java | 5 +++++
.../flink/source/assigners/FIFOSplitAssigner.java | 5 +++++
.../source/assigners/PreAssignSplitAssigner.java | 12 ++++++++++++
.../paimon/flink/source/assigners/SplitAssigner.java | 6 ++++++
.../source/ContinuousFileSplitEnumeratorTest.java | 7 +++++++
7 files changed, 52 insertions(+), 4 deletions(-)
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumerator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumerator.java
index 9096f2814..72b3742b3 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumerator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumerator.java
@@ -192,7 +192,7 @@ public class ContinuousFileSplitEnumerator
// context.callAsync will invoke this. This method runs in workerExecutorThreadPool in
// parallelism.
protected synchronized Optional<PlanWithNextSnapshotId> scanNextSnapshot() {
- if (splitAssigner.remainingSplits().size() >= splitMaxNum) {
+ if (splitAssigner.numberOfRemainingSplits() >= splitMaxNum) {
return Optional.empty();
}
TableScan.Plan plan = scan.plan();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/AlignedSplitAssigner.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/AlignedSplitAssigner.java
index eba25a1af..648758f83 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/AlignedSplitAssigner.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/AlignedSplitAssigner.java
@@ -34,6 +34,7 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
/**
* Splits are allocated at the granularity of snapshots. When the splits of the current snapshot are
@@ -43,8 +44,11 @@ public class AlignedSplitAssigner implements SplitAssigner {
private final Deque<PendingSnapshot> pendingSplitAssignment;
+ private final AtomicInteger numberOfPendingSplits;
+
public AlignedSplitAssigner() {
this.pendingSplitAssignment = new LinkedList<>();
+ this.numberOfPendingSplits = new AtomicInteger(0);
}
@Override
@@ -52,10 +56,12 @@ public class AlignedSplitAssigner implements SplitAssigner {
PendingSnapshot head = pendingSplitAssignment.peek();
if (head != null && !head.isPlaceHolder) {
List<FileStoreSourceSplit> subtaskSplits = head.remove(subtask);
- return subtaskSplits != null ? subtaskSplits : Collections.emptyList();
- } else {
- return Collections.emptyList();
+ if (subtaskSplits != null) {
+ numberOfPendingSplits.getAndAdd(-subtaskSplits.size());
+ return subtaskSplits;
+ }
}
+ return Collections.emptyList();
}
@Override
@@ -70,6 +76,7 @@ public class AlignedSplitAssigner implements SplitAssigner {
} else {
last.add(subtask, splits);
}
+ numberOfPendingSplits.incrementAndGet();
}
@Override
@@ -88,6 +95,7 @@ public class AlignedSplitAssigner implements SplitAssigner {
} else {
head.addAll(suggestedTask, splits);
}
+ numberOfPendingSplits.getAndAdd(splits.size());
}
@Override
@@ -105,6 +113,11 @@ public class AlignedSplitAssigner implements SplitAssigner {
return Optional.ofNullable(head != null ? head.snapshotId : null);
}
+ @Override
+ public int numberOfRemainingSplits() {
+ return numberOfPendingSplits.get();
+ }
+
public boolean isAligned() {
PendingSnapshot head = pendingSplitAssignment.peek();
return head != null && head.empty();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/DynamicPartitionPruningAssigner.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/DynamicPartitionPruningAssigner.java
index d43bc1b02..9221f1f27 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/DynamicPartitionPruningAssigner.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/DynamicPartitionPruningAssigner.java
@@ -109,6 +109,11 @@ public class DynamicPartitionPruningAssigner implements SplitAssigner {
return innerAssigner.getNextSnapshotId(subtask);
}
+ @Override
+ public int numberOfRemainingSplits() {
+ return innerAssigner.numberOfRemainingSplits();
+ }
+
private boolean filter(FileStoreSourceSplit sourceSplit) {
DataSplit dataSplit = (DataSplit) sourceSplit.split();
BinaryRow partition = dataSplit.partition();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/FIFOSplitAssigner.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/FIFOSplitAssigner.java
index 7af226ab0..a2f0b983c 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/FIFOSplitAssigner.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/FIFOSplitAssigner.java
@@ -75,4 +75,9 @@ public class FIFOSplitAssigner implements SplitAssigner {
? Optional.empty()
: getSnapshotId(pendingSplitAssignment.peekFirst());
}
+
+ @Override
+ public int numberOfRemainingSplits() {
+ return pendingSplitAssignment.size();
+ }
}
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 118ed109e..400a2e5c5 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
@@ -34,6 +34,7 @@ import java.util.ListIterator;
import java.util.Map;
import java.util.Optional;
import java.util.Queue;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.paimon.flink.utils.TableScanUtils.getSnapshotId;
@@ -48,6 +49,8 @@ public class PreAssignSplitAssigner implements SplitAssigner {
private final Map<Integer, LinkedList<FileStoreSourceSplit>> pendingSplitAssignment;
+ private final AtomicInteger numberOfPendingSplits;
+
public PreAssignSplitAssigner(
int splitBatchSize,
SplitEnumeratorContext<FileStoreSourceSplit> context,
@@ -55,6 +58,7 @@ public class PreAssignSplitAssigner implements SplitAssigner {
this.splitBatchSize = splitBatchSize;
this.pendingSplitAssignment =
createBatchFairSplitAssignment(splits, context.currentParallelism());
+ this.numberOfPendingSplits = new AtomicInteger(splits.size());
}
@Override
@@ -67,12 +71,14 @@ public class PreAssignSplitAssigner implements SplitAssigner {
while (taskSplits != null && !taskSplits.isEmpty() && assignment.size() < splitBatchSize) {
assignment.add(taskSplits.poll());
}
+ numberOfPendingSplits.getAndAdd(-assignment.size());
return assignment;
}
@Override
public void addSplit(int suggestedTask, FileStoreSourceSplit split) {
pendingSplitAssignment.computeIfAbsent(suggestedTask, k -> new LinkedList<>()).add(split);
+ numberOfPendingSplits.incrementAndGet();
}
@Override
@@ -83,6 +89,7 @@ public class PreAssignSplitAssigner implements SplitAssigner {
while (iterator.hasPrevious()) {
remainingSplits.addFirst(iterator.previous());
}
+ numberOfPendingSplits.getAndAdd(splits.size());
}
@Override
@@ -115,4 +122,9 @@ public class PreAssignSplitAssigner implements SplitAssigner {
? Optional.empty()
: getSnapshotId(pendingSplits.peekFirst());
}
+
+ @Override
+ public int numberOfRemainingSplits() {
+ return numberOfPendingSplits.get();
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/SplitAssigner.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/SplitAssigner.java
index fecd59c88..2ce3af5d7 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/SplitAssigner.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/assigners/SplitAssigner.java
@@ -54,4 +54,10 @@ public interface SplitAssigner {
/** Gets the snapshot id of the next split. */
Optional<Long> getNextSnapshotId(int subtask);
+
+ /**
+ * Gets the current number of remaining splits. This method should be guaranteed to be
+ * thread-safe.
+ */
+ int numberOfRemainingSplits();
}
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java
index 1fd76482e..1a3f42424 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/ContinuousFileSplitEnumeratorTest.java
@@ -792,6 +792,13 @@ public class ContinuousFileSplitEnumeratorTest extends FileSplitEnumeratorTestBa
context.triggerAllActions();
Assertions.assertThat(enumerator.splitAssigner.remainingSplits().size()).isEqualTo(16 * 2);
+ Assertions.assertThat(enumerator.splitAssigner.numberOfRemainingSplits()).isEqualTo(16 * 2);
+
+ enumerator.handleSplitRequest(0, "test");
+ enumerator.handleSplitRequest(1, "test");
+
+ Assertions.assertThat(enumerator.splitAssigner.remainingSplits().size()).isEqualTo(15 * 2);
+ Assertions.assertThat(enumerator.splitAssigner.numberOfRemainingSplits()).isEqualTo(15 * 2);
}
private void triggerCheckpointAndComplete(