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 2023/11/14 06:18:22 UTC
(incubator-paimon) branch master updated: [flink] support the feature of ContinuousFileStoreSource using consumer-id. (#2221)
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 567241908 [flink] support the feature of ContinuousFileStoreSource using consumer-id. (#2221)
567241908 is described below
commit 567241908c5df476357a8810f1e1f033aabe8bd2
Author: liming.1018 <69...@users.noreply.github.com>
AuthorDate: Tue Nov 14 14:18:18 2023 +0800
[flink] support the feature of ContinuousFileStoreSource using consumer-id. (#2221)
---
docs/content/how-to/querying-tables.md | 12 +++
.../shortcodes/generated/core_configuration.html | 6 ++
.../main/java/org/apache/paimon/CoreOptions.java | 39 ++++++++
.../flink/source/ConsumerProgressCalculator.java | 111 +++++++++++++++++++++
.../source/ContinuousFileSplitEnumerator.java | 26 ++++-
.../paimon/flink/source/FileStoreSourceReader.java | 17 ++++
.../paimon/flink/source/FlinkSourceBuilder.java | 4 +-
.../flink/source/ReaderConsumeProgressEvent.java | 48 +++++++++
.../AlignedContinuousFileSplitEnumerator.java | 2 +-
.../source/assigners/AlignedSplitAssigner.java | 12 ++-
.../assigners/DynamicPartitionPruningAssigner.java | 6 ++
.../flink/source/assigners/FIFOSplitAssigner.java | 10 ++
.../source/assigners/PreAssignSplitAssigner.java | 11 ++
.../flink/source/assigners/SplitAssigner.java | 4 +
.../apache/paimon/flink/utils/TableScanUtils.java | 11 ++
.../source/ContinuousFileSplitEnumeratorTest.java | 91 ++++++++++++++++-
.../flink/source/FileStoreSourceReaderTest.java | 22 ++++
.../source/align/AlignedSourceReaderTest.java | 7 ++
18 files changed, 430 insertions(+), 9 deletions(-)
diff --git a/docs/content/how-to/querying-tables.md b/docs/content/how-to/querying-tables.md
index e2b9fc912..223c6c7e9 100644
--- a/docs/content/how-to/querying-tables.md
+++ b/docs/content/how-to/querying-tables.md
@@ -332,6 +332,18 @@ NOTE: The consumer will prevent expiration of the snapshot. You can specify 'con
lifetime of consumers.
{{< /hint >}}
+By default, the consumer uses `exactly-once` mode to record consumption progress, which strictly ensures that what is
+recorded in the consumer is the snapshot-id + 1 that all readers have exactly consumed. You can set `consumer.mode` to
+`at-least-once` to allow readers consume snapshots at different rates and record the slowest snapshot-id among all
+readers into the consumer. This mode can provide more capabilities, such as watermark alignment.
+
+{{< hint warning >}}
+1. When there is no watermark definition, the consumer in `at-least-once` mode cannot provide the ability to pass the
+watermark in the snapshot to the downstream.
+2. Since the implementation of `exactly-once` mode and `at-least-once` mode are completely different, the state of
+flink is incompatible and cannot be restored from the state when switching modes.
+{{< /hint >}}
+
You can reset a consumer with a given consumer ID and next snapshot ID and delete a consumer with a given consumer ID.
{{< hint info >}}
diff --git a/docs/layouts/shortcodes/generated/core_configuration.html b/docs/layouts/shortcodes/generated/core_configuration.html
index 13565db12..d6c533a7f 100644
--- a/docs/layouts/shortcodes/generated/core_configuration.html
+++ b/docs/layouts/shortcodes/generated/core_configuration.html
@@ -110,6 +110,12 @@ under the License.
<td>Duration</td>
<td>The expiration interval of consumer files. A consumer file will be expired if it's lifetime after last modification is over this value.</td>
</tr>
+ <tr>
+ <td><h5>consumer.mode</h5></td>
+ <td style="word-wrap: break-word;">exactly-once</td>
+ <td>Boolean</td>
+ <td>Specify the consumer consistency mode for table.<br /><br />Possible values:<ul><li>"exactly-once": Readers consume data at snapshot granularity, and strictly ensure that the snapshot-id recorded in the consumer is the snapshot-id + 1 that all readers have exactly consumed.</li><li>"at-least-once": Each reader consumes snapshots at a different rate, and the snapshot with the slowest consumption progress among all readers will be recorded in the consumer.</li></ul></td>
+ </tr>
<tr>
<td><h5>continuous.discovery-interval</h5></td>
<td style="word-wrap: break-word;">10 s</td>
diff --git a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
index 54abd1e22..55057e36e 100644
--- a/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/CoreOptions.java
@@ -709,6 +709,12 @@ public class CoreOptions implements Serializable {
"The expiration interval of consumer files. A consumer file will be expired if "
+ "it's lifetime after last modification is over this value.");
+ public static final ConfigOption<ConsumerMode> CONSUMER_CONSISTENCY_MODE =
+ key("consumer.mode")
+ .enumType(ConsumerMode.class)
+ .defaultValue(ConsumerMode.EXACTLY_ONCE)
+ .withDescription("Specify the consumer consistency mode for table.");
+
public static final ConfigOption<Long> DYNAMIC_BUCKET_TARGET_ROW_NUM =
key("dynamic-bucket.target-row-num")
.longType()
@@ -1307,6 +1313,10 @@ public class CoreOptions implements Serializable {
return options.get(CONSUMER_EXPIRATION_TIME);
}
+ public ConsumerMode consumerWithLegacyMode() {
+ return options.get(CONSUMER_CONSISTENCY_MODE);
+ }
+
public boolean partitionedTableInMetastore() {
return options.get(METASTORE_PARTITIONED_TABLE);
}
@@ -1951,4 +1961,33 @@ public class CoreOptions implements Serializable {
return text(description);
}
}
+
+ /** Specifies the log consistency mode for table. */
+ public enum ConsumerMode implements DescribedEnum {
+ EXACTLY_ONCE(
+ "exactly-once",
+ "Readers consume data at snapshot granularity, and strictly ensure that the snapshot-id recorded in the consumer is the snapshot-id + 1 that all readers have exactly consumed."),
+
+ AT_LEAST_ONCE(
+ "at-least-once",
+ "Each reader consumes snapshots at a different rate, and the snapshot with the slowest consumption progress among all readers will be recorded in the consumer.");
+
+ private final String value;
+ private final String description;
+
+ ConsumerMode(String value, String description) {
+ this.value = value;
+ this.description = description;
+ }
+
+ @Override
+ public String toString() {
+ return value;
+ }
+
+ @Override
+ public InlineElement getDescription() {
+ return text(description);
+ }
+ }
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ConsumerProgressCalculator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ConsumerProgressCalculator.java
new file mode 100644
index 000000000..f88428a2e
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ConsumerProgressCalculator.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.source;
+
+import org.apache.paimon.flink.utils.TableScanUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NavigableMap;
+import java.util.Optional;
+import java.util.OptionalLong;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.function.Function;
+
+/** Calculator for calculating consumer consumption progress. */
+public class ConsumerProgressCalculator {
+ private final TreeMap<Long, Long> minNextSnapshotPerCheckpoint;
+
+ private final Map<Integer, Long> assignedSnapshotPerReader;
+
+ private final Map<Integer, Long> consumingSnapshotPerReader;
+
+ public ConsumerProgressCalculator(int parallelism) {
+ this.minNextSnapshotPerCheckpoint = new TreeMap<>();
+ this.assignedSnapshotPerReader = new HashMap<>(parallelism);
+ this.consumingSnapshotPerReader = new HashMap<>(parallelism);
+ }
+
+ public void updateConsumeProgress(int subtaskId, ReaderConsumeProgressEvent event) {
+ consumingSnapshotPerReader.put(subtaskId, event.lastConsumeSnapshotId());
+ }
+
+ public void updateAssignInformation(int subtaskId, FileStoreSourceSplit split) {
+ TableScanUtils.getSnapshotId(split)
+ .ifPresent(snapshotId -> assignedSnapshotPerReader.put(subtaskId, snapshotId));
+ }
+
+ public void notifySnapshotState(
+ long checkpointId,
+ Set<Integer> readersAwaitingSplit,
+ Function<Integer, Long> unassignedCalculationFunction,
+ int parallelism) {
+ computeMinNextSnapshotId(readersAwaitingSplit, unassignedCalculationFunction, parallelism)
+ .ifPresent(
+ minNextSnapshotId ->
+ minNextSnapshotPerCheckpoint.put(checkpointId, minNextSnapshotId));
+ }
+
+ public OptionalLong notifyCheckpointComplete(long checkpointId) {
+ NavigableMap<Long, Long> nextSnapshots =
+ minNextSnapshotPerCheckpoint.headMap(checkpointId, true);
+ OptionalLong max = nextSnapshots.values().stream().mapToLong(Long::longValue).max();
+ nextSnapshots.clear();
+ return max;
+ }
+
+ /** Calculate the minimum snapshot currently being consumed by all readers. */
+ private Optional<Long> computeMinNextSnapshotId(
+ Set<Integer> readersAwaitingSplit,
+ Function<Integer, Long> unassignedCalculationFunction,
+ int parallelism) {
+ long globalMinSnapshotId = Long.MAX_VALUE;
+ for (int subtask = 0; subtask < parallelism; subtask++) {
+ // 1. if the reader is in the waiting list, it means that all allocated splits have
+ // been
+ // consumed, and the next snapshotId is calculated from splitAssigner.
+ //
+ // 2. if the reader is not in the waiting list, the larger value between the
+ // consumption
+ // progress reported by the reader and the most recently assigned snapshot id is
+ // used.
+ Long snapshotIdForSubtask;
+ if (readersAwaitingSplit.contains(subtask)) {
+ snapshotIdForSubtask = unassignedCalculationFunction.apply(subtask);
+ } else {
+ Long consumingSnapshotId = consumingSnapshotPerReader.get(subtask);
+ Long assignedSnapshotId = assignedSnapshotPerReader.get(subtask);
+ if (consumingSnapshotId != null && assignedSnapshotId != null) {
+ snapshotIdForSubtask = Math.max(consumingSnapshotId, assignedSnapshotId);
+ } else {
+ snapshotIdForSubtask =
+ consumingSnapshotId != null ? consumingSnapshotId : assignedSnapshotId;
+ }
+ }
+
+ if (snapshotIdForSubtask != null) {
+ globalMinSnapshotId = Math.min(globalMinSnapshotId, snapshotIdForSubtask);
+ } else {
+ return Optional.empty();
+ }
+ }
+ return Optional.of(globalMinSnapshotId);
+ }
+}
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 b5ba3d839..e8ca0f7ca 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
@@ -71,6 +71,8 @@ public class ContinuousFileSplitEnumerator
protected final SplitAssigner splitAssigner;
+ protected final ConsumerProgressCalculator consumerProgressCalculator;
+
@Nullable protected Long nextSnapshotId;
protected boolean finished = false;
@@ -93,6 +95,9 @@ public class ContinuousFileSplitEnumerator
this.scan = scan;
this.splitAssigner = createSplitAssigner(bucketMode);
addSplits(remainSplits);
+
+ this.consumerProgressCalculator =
+ new ConsumerProgressCalculator(context.currentParallelism());
}
@VisibleForTesting
@@ -140,7 +145,12 @@ public class ContinuousFileSplitEnumerator
@Override
public void handleSourceEvent(int subtaskId, SourceEvent sourceEvent) {
- LOG.error("Received unrecognized event: {}", sourceEvent);
+ if (sourceEvent instanceof ReaderConsumeProgressEvent) {
+ consumerProgressCalculator.updateConsumeProgress(
+ subtaskId, (ReaderConsumeProgressEvent) sourceEvent);
+ } else {
+ LOG.error("Received unrecognized event: {}", sourceEvent);
+ }
}
@Override
@@ -155,10 +165,23 @@ public class ContinuousFileSplitEnumerator
final PendingSplitsCheckpoint checkpoint =
new PendingSplitsCheckpoint(splits, nextSnapshotId);
+ consumerProgressCalculator.notifySnapshotState(
+ checkpointId,
+ readersAwaitingSplit,
+ subtask -> splitAssigner.getNextSnapshotId(subtask).orElse(nextSnapshotId),
+ context.currentParallelism());
+
LOG.debug("Source Checkpoint is {}", checkpoint);
return checkpoint;
}
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws Exception {
+ consumerProgressCalculator
+ .notifyCheckpointComplete(checkpointId)
+ .ifPresent(scan::notifyCheckpointComplete);
+ }
+
// ------------------------------------------------------------------------
// this need to be synchronized because scan object is not thread safe. handleSplitRequest and
@@ -218,6 +241,7 @@ public class ContinuousFileSplitEnumerator
List<FileStoreSourceSplit> splits = splitAssigner.getNext(task, null);
if (splits.size() > 0) {
assignment.put(task, splits);
+ consumerProgressCalculator.updateAssignInformation(task, splits.get(0));
}
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceReader.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceReader.java
index 155fbd5da..ed44ac67e 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceReader.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FileStoreSourceReader.java
@@ -20,6 +20,7 @@ package org.apache.paimon.flink.source;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
+import org.apache.paimon.flink.utils.TableScanUtils;
import org.apache.paimon.table.source.TableRead;
import org.apache.flink.api.connector.source.SourceReader;
@@ -33,6 +34,7 @@ import org.apache.flink.table.data.RowData;
import javax.annotation.Nullable;
import java.util.Map;
+import java.util.Optional;
/** A {@link SourceReader} that read records from {@link FileStoreSourceSplit}. */
public class FileStoreSourceReader
@@ -40,6 +42,7 @@ public class FileStoreSourceReader
RecordIterator<RowData>, RowData, FileStoreSourceSplit, FileStoreSourceSplitState> {
@Nullable private final IOManager ioManager;
+ private long lastConsumeSnapshotId = Long.MIN_VALUE;
public FileStoreSourceReader(
SourceReaderContext readerContext,
@@ -93,6 +96,20 @@ public class FileStoreSourceReader
if (getNumberOfCurrentlyAssignedSplits() == 0) {
context.sendSplitRequest();
}
+
+ long maxFinishedSplits =
+ finishedSplitIds.values().stream()
+ .map(splitState -> TableScanUtils.getSnapshotId(splitState.toSourceSplit()))
+ .filter(Optional::isPresent)
+ .mapToLong(Optional::get)
+ .max()
+ .orElse(Long.MIN_VALUE);
+
+ if (lastConsumeSnapshotId < maxFinishedSplits) {
+ lastConsumeSnapshotId = maxFinishedSplits;
+ context.sendSourceEventToCoordinator(
+ new ReaderConsumeProgressEvent(lastConsumeSnapshotId));
+ }
}
@Override
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
index bd9f4c7ac..215d2b6c6 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
@@ -235,7 +235,9 @@ public class FlinkSourceBuilder {
} else {
if (conf.get(FlinkConnectorOptions.SOURCE_CHECKPOINT_ALIGN_ENABLED)) {
return buildAlignedContinuousFileSource();
- } else if (conf.contains(CoreOptions.CONSUMER_ID)) {
+ } else if (conf.contains(CoreOptions.CONSUMER_ID)
+ && conf.get(CoreOptions.CONSUMER_CONSISTENCY_MODE)
+ == CoreOptions.ConsumerMode.EXACTLY_ONCE) {
return buildContinuousStreamOperator();
} else {
return buildContinuousFileSource();
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ReaderConsumeProgressEvent.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ReaderConsumeProgressEvent.java
new file mode 100644
index 000000000..f83d4b181
--- /dev/null
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/ReaderConsumeProgressEvent.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.source;
+
+import org.apache.flink.api.connector.source.SourceEvent;
+
+/**
+ * Event sent from {@link FileStoreSourceReader} to {@link ContinuousFileSplitEnumerator} to
+ * describe the current consumption progress.
+ */
+public class ReaderConsumeProgressEvent implements SourceEvent {
+
+ private static final long serialVersionUID = -1L;
+
+ private final long lastConsumeSnapshotId;
+
+ public ReaderConsumeProgressEvent(long lastConsumeSnapshotId) {
+ this.lastConsumeSnapshotId = lastConsumeSnapshotId;
+ }
+
+ public long lastConsumeSnapshotId() {
+ return lastConsumeSnapshotId;
+ }
+
+ @Override
+ public String toString() {
+ return "ReaderConsumeProgressEvent{"
+ + "lastConsumeSnapshotId="
+ + lastConsumeSnapshotId
+ + '}';
+ }
+}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumerator.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumerator.java
index 379a3ffda..62d0abae3 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumerator.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/align/AlignedContinuousFileSplitEnumerator.java
@@ -164,7 +164,7 @@ public class AlignedContinuousFileSplitEnumerator extends ContinuousFileSplitEnu
assignSplits();
}
Preconditions.checkArgument(alignedAssigner.isAligned());
- lastConsumedSnapshotId = alignedAssigner.minRemainingSnapshotId();
+ lastConsumedSnapshotId = alignedAssigner.getNextSnapshotId(0).orElse(null);
alignedAssigner.removeFirst();
currentCheckpointId = checkpointId;
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 bb03fb454..eba25a1af 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
@@ -33,6 +33,7 @@ import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
/**
* Splits are allocated at the granularity of snapshots. When the splits of the current snapshot are
@@ -98,6 +99,12 @@ public class AlignedSplitAssigner implements SplitAssigner {
return remainingSplits;
}
+ @Override
+ public Optional<Long> getNextSnapshotId(int subtask) {
+ PendingSnapshot head = pendingSplitAssignment.peek();
+ return Optional.ofNullable(head != null ? head.snapshotId : null);
+ }
+
public boolean isAligned() {
PendingSnapshot head = pendingSplitAssignment.peek();
return head != null && head.empty();
@@ -114,11 +121,6 @@ public class AlignedSplitAssigner implements SplitAssigner {
"The head pending splits is not empty. This is a bug, please file an issue.");
}
- public Long minRemainingSnapshotId() {
- PendingSnapshot head = pendingSplitAssignment.peek();
- return head != null ? head.snapshotId : null;
- }
-
private static class PendingSnapshot {
private final long snapshotId;
private final boolean isPlaceHolder;
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 7b997dae2..d43bc1b02 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
@@ -34,6 +34,7 @@ import javax.annotation.Nullable;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
import java.util.stream.Collectors;
/** Assigner to perform dynamic partition pruning by given {@link DynamicFilteringData}. */
@@ -103,6 +104,11 @@ public class DynamicPartitionPruningAssigner implements SplitAssigner {
: oriAssigner;
}
+ @Override
+ public Optional<Long> getNextSnapshotId(int subtask) {
+ return innerAssigner.getNextSnapshotId(subtask);
+ }
+
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 52cf28f21..7af226ab0 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
@@ -28,6 +28,9 @@ import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
+import java.util.Optional;
+
+import static org.apache.paimon.flink.utils.TableScanUtils.getSnapshotId;
/**
* Splits are assigned preemptively in the order requested by the task. Only one split is assigned
@@ -65,4 +68,11 @@ public class FIFOSplitAssigner implements SplitAssigner {
public Collection<FileStoreSourceSplit> remainingSplits() {
return new ArrayList<>(pendingSplitAssignment);
}
+
+ @Override
+ public Optional<Long> getNextSnapshotId(int subtask) {
+ return pendingSplitAssignment.isEmpty()
+ ? Optional.empty()
+ : getSnapshotId(pendingSplitAssignment.peekFirst());
+ }
}
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 63e24026f..118ed109e 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
@@ -32,8 +32,11 @@ import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
+import java.util.Optional;
import java.util.Queue;
+import static org.apache.paimon.flink.utils.TableScanUtils.getSnapshotId;
+
/**
* Pre-calculate which splits each task should process according to the weight, and then distribute
* the splits fairly.
@@ -104,4 +107,12 @@ public class PreAssignSplitAssigner implements SplitAssigner {
}
return assignment;
}
+
+ @Override
+ public Optional<Long> getNextSnapshotId(int subtask) {
+ LinkedList<FileStoreSourceSplit> pendingSplits = pendingSplitAssignment.get(subtask);
+ return (pendingSplits == null || pendingSplits.isEmpty())
+ ? Optional.empty()
+ : getSnapshotId(pendingSplits.peekFirst());
+ }
}
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 e89d9fa63..fecd59c88 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
@@ -24,6 +24,7 @@ import javax.annotation.Nullable;
import java.util.Collection;
import java.util.List;
+import java.util.Optional;
/**
* The {@code SplitAssigner} is responsible for deciding what splits should be processed next by
@@ -50,4 +51,7 @@ public interface SplitAssigner {
/** Gets the remaining splits that this assigner has pending. */
Collection<FileStoreSourceSplit> remainingSplits();
+
+ /** Gets the snapshot id of the next split. */
+ Optional<Long> getNextSnapshotId(int subtask);
}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java
index 05486972f..a5645302f 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/utils/TableScanUtils.java
@@ -19,10 +19,13 @@
package org.apache.paimon.flink.utils;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.flink.source.FileStoreSourceSplit;
import org.apache.paimon.table.Table;
+import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.TableScan;
import java.util.HashMap;
+import java.util.Optional;
/** Utility methods for {@link TableScan}, such as validating. */
public class TableScanUtils {
@@ -48,4 +51,12 @@ public class TableScanUtils {
}
}
}
+
+ /** Get snapshot id from {@link FileStoreSourceSplit}. */
+ public static Optional<Long> getSnapshotId(FileStoreSourceSplit split) {
+ if (split.split() instanceof DataSplit) {
+ return Optional.of(((DataSplit) split.split()).snapshotId());
+ }
+ return Optional.empty();
+ }
}
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 64b6463bf..dfc788403 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
@@ -37,6 +37,7 @@ import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
@@ -681,6 +682,86 @@ public class ContinuousFileSplitEnumeratorTest extends FileSplitEnumeratorTestBa
assertThat(toDataSplits(state.splits())).containsExactlyElementsOf(expectedResults.get(2L));
}
+ @Test
+ public void testEnumeratorWithConsumer() throws Exception {
+ final TestingAsyncSplitEnumeratorContext<FileStoreSourceSplit> context =
+ new TestingAsyncSplitEnumeratorContext<>(3);
+ for (int i = 0; i < 3; i++) {
+ context.registerReader(i, "test-host");
+ }
+
+ // prepare test data
+ TreeMap<Long, TableScan.Plan> dataSplits = new TreeMap<>();
+ for (int i = 1; i <= 2; i++) {
+ dataSplits.put(
+ (long) i,
+ new DataFilePlan(
+ Arrays.asList(
+ createDataSplit(i, 0, Collections.emptyList()),
+ createDataSplit(i, 2, Collections.emptyList()))));
+ }
+ MockScan scan = new MockScan(dataSplits);
+
+ final ContinuousFileSplitEnumerator enumerator =
+ new Builder()
+ .setSplitEnumeratorContext(context)
+ .setInitialSplits(Collections.emptyList())
+ .setDiscoveryInterval(1)
+ .setScan(scan)
+ .build();
+ enumerator.start();
+
+ long checkpointId = 1L;
+
+ // request for splits
+ for (int i = 0; i < 3; i++) {
+ enumerator.handleSplitRequest(i, "test-host");
+ }
+
+ // checkpoint is triggered for the first time and no snapshot is found
+ triggerCheckpointAndComplete(enumerator, checkpointId++);
+ assertThat(scan.getNextSnapshotIdForConsumer()).isNull();
+
+ // find a new snapshot and trigger for the second checkpoint, but no snapshot is consumed
+ scanNextSnapshot(context);
+ triggerCheckpointAndComplete(enumerator, checkpointId++);
+ assertThat(scan.getNextSnapshotIdForConsumer()).isEqualTo(1L);
+
+ // subtask-0 has consumed the snapshot-1 and trigger for the next checkpoint
+ enumerator.handleSourceEvent(0, new ReaderConsumeProgressEvent(1L));
+ triggerCheckpointAndComplete(enumerator, checkpointId++);
+ assertThat(scan.getNextSnapshotIdForConsumer()).isEqualTo(1L);
+
+ // subtask-2 has consumed the snapshot-1 and trigger for the next checkpoint
+ enumerator.handleSourceEvent(2, new ReaderConsumeProgressEvent(1L));
+ triggerCheckpointAndComplete(enumerator, checkpointId++);
+ assertThat(scan.getNextSnapshotIdForConsumer()).isEqualTo(1L);
+
+ // subtask-0 and subtask-2 request for the next splits but there are no new snapshot
+ enumerator.handleSplitRequest(0, "test-host");
+ enumerator.handleSplitRequest(2, "test-host");
+ triggerCheckpointAndComplete(enumerator, checkpointId++);
+ assertThat(scan.getNextSnapshotIdForConsumer()).isEqualTo(2L);
+
+ // find next snapshot and trigger for the next checkpoint, subtask-0 and subtask-2 has been
+ // assigned new snapshot
+ scanNextSnapshot(context);
+ triggerCheckpointAndComplete(enumerator, checkpointId++);
+ assertThat(scan.getNextSnapshotIdForConsumer()).isEqualTo(2L);
+ }
+
+ private void triggerCheckpointAndComplete(
+ ContinuousFileSplitEnumerator enumerator, long checkpointId) throws Exception {
+ enumerator.snapshotState(checkpointId);
+ enumerator.notifyCheckpointComplete(checkpointId);
+ }
+
+ private void scanNextSnapshot(
+ TestingAsyncSplitEnumeratorContext<FileStoreSourceSplit> context) {
+ context.workerExecutor.triggerPeriodicScheduledTasks();
+ context.triggerAlCoordinatorAction();
+ }
+
private static PendingSplitsCheckpoint checkpointWithoutException(
ContinuousFileSplitEnumerator enumerator, long checkpointId) {
try {
@@ -753,10 +834,12 @@ public class ContinuousFileSplitEnumeratorTest extends FileSplitEnumeratorTestBa
private final TreeMap<Long, Plan> results;
private @Nullable Long nextSnapshotId;
private boolean allowEnd = true;
+ private Long nextSnapshotIdForConsumer;
public MockScan(TreeMap<Long, Plan> results) {
this.results = results;
this.nextSnapshotId = null;
+ this.nextSnapshotIdForConsumer = null;
}
@Override
@@ -784,7 +867,9 @@ public class ContinuousFileSplitEnumeratorTest extends FileSplitEnumeratorTestBa
}
@Override
- public void notifyCheckpointComplete(@Nullable Long nextSnapshot) {}
+ public void notifyCheckpointComplete(@Nullable Long nextSnapshot) {
+ nextSnapshotIdForConsumer = nextSnapshot;
+ }
@Nullable
@Override
@@ -798,6 +883,10 @@ public class ContinuousFileSplitEnumeratorTest extends FileSplitEnumeratorTestBa
public void allowEnd(boolean allowEnd) {
this.allowEnd = allowEnd;
}
+
+ public Long getNextSnapshotIdForConsumer() {
+ return nextSnapshotIdForConsumer;
+ }
}
private static class TestingAsyncSplitEnumeratorContext<SplitT extends SourceSplit>
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceReaderTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceReaderTest.java
index b92d30f32..6db17aff6 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceReaderTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FileStoreSourceReaderTest.java
@@ -27,6 +27,7 @@ import org.apache.paimon.types.DataField;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
+import org.apache.flink.api.connector.source.SourceEvent;
import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
import org.apache.flink.table.data.RowData;
@@ -36,6 +37,7 @@ import org.junit.jupiter.api.io.TempDir;
import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
import static org.apache.paimon.flink.source.FileStoreSourceSplitSerializerTest.newSourceSplit;
import static org.apache.paimon.mergetree.compact.MergeTreeCompactManagerTest.row;
@@ -104,6 +106,26 @@ public class FileStoreSourceReaderTest {
assertThat(context.getNumSplitRequests()).isEqualTo(2);
}
+ @Test
+ public void testReaderOnSplitFinished() throws Exception {
+ final TestingReaderContext context = new TestingReaderContext();
+ final FileStoreSourceReader reader = createReader(context);
+
+ reader.start();
+ reader.addSplits(Collections.singletonList(createTestFileSplit("id1")));
+ TestingReaderOutput<RowData> output = new TestingReaderOutput<>();
+ while (reader.getNumberOfCurrentlyAssignedSplits() > 0) {
+ reader.pollNext(output);
+ Thread.sleep(10);
+ }
+
+ List<SourceEvent> sourceEvents = context.getSentEvents();
+ assertThat(sourceEvents.size()).isEqualTo(1);
+ assertThat(sourceEvents.get(0)).isExactlyInstanceOf(ReaderConsumeProgressEvent.class);
+ assertThat(((ReaderConsumeProgressEvent) sourceEvents.get(0)))
+ .matches(event -> event.lastConsumeSnapshotId() == 1L);
+ }
+
protected FileStoreSourceReader createReader(TestingReaderContext context) {
return new FileStoreSourceReader(
context,
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/align/AlignedSourceReaderTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/align/AlignedSourceReaderTest.java
index 0d38ae3cf..37e626069 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/align/AlignedSourceReaderTest.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/align/AlignedSourceReaderTest.java
@@ -26,6 +26,7 @@ import org.apache.flink.connector.base.source.reader.synchronization.FutureCompl
import org.apache.flink.connector.testutils.source.reader.TestingReaderContext;
import org.apache.flink.connector.testutils.source.reader.TestingReaderOutput;
import org.apache.flink.table.data.RowData;
+import org.junit.Ignore;
import org.junit.jupiter.api.Test;
import java.util.Arrays;
@@ -60,6 +61,12 @@ public class AlignedSourceReaderTest extends FileStoreSourceReaderTest {
assertThat(context.getNumSplitRequests()).isEqualTo(2);
}
+ @Override
+ @Ignore
+ public void testReaderOnSplitFinished() throws Exception {
+ // ignore
+ }
+
@Override
protected FileStoreSourceReader createReader(TestingReaderContext context) {
return new AlignedSourceReader(