This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 0afc21a4b [client] Fix tiering hang on first_row merge engine empty
batches (#3242)
0afc21a4b is described below
commit 0afc21a4bd094229c5943e3bb05a0db3286cc6d2
Author: Kaixuan Duan <[email protected]>
AuthorDate: Fri Jun 5 10:16:16 2026 +0800
[client] Fix tiering hang on first_row merge engine empty batches (#3242)
---
.../scanner/log/AbstractLogFetchCollector.java | 15 ++-
.../table/scanner/log/ArrowLogFetchCollector.java | 6 +-
.../table/scanner/log/LogFetchCollector.java | 6 +-
.../client/table/scanner/log/LogScannerImpl.java | 9 +-
.../client/table/scanner/log/ScanRecords.java | 33 +++++-
.../table/scanner/log/LogFetchCollectorTest.java | 3 +
.../client/table/scanner/log/LogFetcherITCase.java | 8 +-
.../client/table/scanner/log/ScanRecordsTest.java | 44 ++++++++
.../flink/tiering/source/TieringSplitReader.java | 120 +++++++++++++--------
.../tiering/source/TieringSplitReaderTest.java | 90 ++++++++++++++++
10 files changed, 271 insertions(+), 63 deletions(-)
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogFetchCollector.java
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogFetchCollector.java
index f4daf890e..e058e0e2d 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogFetchCollector.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogFetchCollector.java
@@ -65,7 +65,7 @@ abstract class AbstractLogFetchCollector<T, R> {
/**
* Return the fetched log records, empty the record buffer and update the
consumed position.
*
- * <p>NOTE: returning empty records guarantees the consumed position are
NOT updated.
+ * <p>NOTE: empty record lists may still advance the consumed position.
*
* @return The fetched records per partition
* @throws FetchException If there is OffsetOutOfRange error in
fetchResponse and the
@@ -73,6 +73,7 @@ abstract class AbstractLogFetchCollector<T, R> {
*/
public R collectFetch(final LogFetchBuffer logFetchBuffer) {
Map<TableBucket, List<T>> fetched = new HashMap<>();
+ Map<TableBucket, Long> consumedUpToOffsets = new HashMap<>();
int recordsRemaining = maxPollRecords;
try {
@@ -108,8 +109,11 @@ abstract class AbstractLogFetchCollector<T, R> {
logFetchBuffer.poll();
} else {
List<T> records = fetchRecords(nextInLineFetch,
recordsRemaining);
+ TableBucket tableBucket = nextInLineFetch.tableBucket;
+ // Always record the advanced next fetch offset for this
bucket, even when
+ // the materialized record list is empty.
+ consumedUpToOffsets.put(tableBucket,
nextInLineFetch.nextFetchOffset());
if (!records.isEmpty()) {
- TableBucket tableBucket = nextInLineFetch.tableBucket;
List<T> currentRecords = fetched.get(tableBucket);
if (currentRecords == null) {
fetched.put(tableBucket, records);
@@ -126,6 +130,8 @@ abstract class AbstractLogFetchCollector<T, R> {
}
recordsRemaining -= recordCount(records);
+ } else {
+ fetched.putIfAbsent(tableBucket,
Collections.emptyList());
}
}
}
@@ -140,7 +146,7 @@ abstract class AbstractLogFetchCollector<T, R> {
throw e;
}
- return toResult(fetched);
+ return toResult(fetched, consumedUpToOffsets);
}
/** Initialize a {@link CompletedFetch} object. */
@@ -293,7 +299,8 @@ abstract class AbstractLogFetchCollector<T, R> {
protected abstract int recordCount(List<T> fetchedRecords);
- protected abstract R toResult(Map<TableBucket, List<T>> fetchedRecords);
+ protected abstract R toResult(
+ Map<TableBucket, List<T>> fetchedRecords, Map<TableBucket, Long>
consumedUpToOffsets);
/**
* Release resources held by fetched records on failure. The default
implementation is a no-op,
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ArrowLogFetchCollector.java
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ArrowLogFetchCollector.java
index 39b6aac02..4c6f7eab9 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ArrowLogFetchCollector.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ArrowLogFetchCollector.java
@@ -62,7 +62,11 @@ public class ArrowLogFetchCollector
}
@Override
- protected ArrowScanRecords toResult(Map<TableBucket, List<ArrowBatchData>>
fetchedRecords) {
+ protected ArrowScanRecords toResult(
+ Map<TableBucket, List<ArrowBatchData>> fetchedRecords,
+ Map<TableBucket, Long> consumedUpToOffsets) {
+ // Arrow scan paths don't need consumedUpToOffsets (issue #2371 is
specific to
+ // row-based tiering), so it's discarded here rather than carried in
ArrowScanRecords.
return new ArrowScanRecords(fetchedRecords);
}
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
index bcf47c07f..799e74d2e 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
@@ -67,7 +67,9 @@ public class LogFetchCollector extends
AbstractLogFetchCollector<ScanRecord, Sca
}
@Override
- protected ScanRecords toResult(Map<TableBucket, List<ScanRecord>>
fetchedRecords) {
- return new ScanRecords(fetchedRecords);
+ protected ScanRecords toResult(
+ Map<TableBucket, List<ScanRecord>> fetchedRecords,
+ Map<TableBucket, Long> consumedUpToOffsets) {
+ return new ScanRecords(fetchedRecords, consumedUpToOffsets);
}
}
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerImpl.java
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerImpl.java
index e7178f066..9e9316069 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerImpl.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerImpl.java
@@ -141,7 +141,11 @@ public class LogScannerImpl implements LogScanner {
@Override
public ScanRecords poll(Duration timeout) {
- return doPoll(timeout, this::pollForFetches, ScanRecords::isEmpty, ()
-> ScanRecords.EMPTY);
+ return doPoll(
+ timeout,
+ this::pollForFetches,
+ scanRecords -> scanRecords.buckets().isEmpty(),
+ () -> ScanRecords.EMPTY);
}
/**
@@ -250,7 +254,8 @@ public class LogScannerImpl implements LogScanner {
private ScanRecords pollForFetches() {
ScanRecords scanRecords = logFetcher.collectFetch();
- if (!scanRecords.isEmpty()) {
+ // Check buckets() (includes progress-only buckets).
+ if (!scanRecords.buckets().isEmpty()) {
return scanRecords;
}
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java
index 9d58c22b4..eb3e157d1 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java
@@ -22,6 +22,8 @@ import org.apache.fluss.client.table.scanner.ScanRecord;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.utils.AbstractIterator;
+import javax.annotation.Nullable;
+
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
@@ -41,8 +43,18 @@ public class ScanRecords implements Iterable<ScanRecord> {
private final Map<TableBucket, List<ScanRecord>> records;
+ /** The exclusive upper bound of consumed offsets per polled bucket in
this round. */
+ private final Map<TableBucket, Long> consumedUpToOffsets;
+
public ScanRecords(Map<TableBucket, List<ScanRecord>> records) {
+ this(records, Collections.emptyMap());
+ }
+
+ public ScanRecords(
+ Map<TableBucket, List<ScanRecord>> records,
+ Map<TableBucket, Long> consumedUpToOffsets) {
this.records = records;
+ this.consumedUpToOffsets = consumedUpToOffsets;
}
/**
@@ -59,15 +71,25 @@ public class ScanRecords implements Iterable<ScanRecord> {
}
/**
- * Get the bucket ids which have records contained in this record set.
- *
- * @return the set of partitions with data in this record set (maybe empty
if no data was
- * returned)
+ * Get the bucket ids that were polled in this round, including buckets
whose record list is
+ * empty but whose log offset still advanced.
*/
public Set<TableBucket> buckets() {
return Collections.unmodifiableSet(records.keySet());
}
+ /**
+ * Get the exclusive upper bound of offsets consumed for the given bucket
in this poll round.
+ *
+ * @param bucket the bucket to query
+ * @return the exclusive upper bound offset, or {@code null} if the bucket
was not polled in
+ * this round
+ */
+ @Nullable
+ public Long consumedUpToOffset(TableBucket bucket) {
+ return consumedUpToOffsets.get(bucket);
+ }
+
/** The number of records for all buckets. */
public int count() {
int count = 0;
@@ -77,8 +99,9 @@ public class ScanRecords implements Iterable<ScanRecord> {
return count;
}
+ /** Returns {@code true} if this {@code ScanRecords} contains no
materialized records. */
public boolean isEmpty() {
- return records.isEmpty();
+ return count() == 0;
}
@Override
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
index e769ebf4e..3b4d47fc1 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java
@@ -258,6 +258,9 @@ public class LogFetchCollectorTest {
assertThat(scanRecords.records(tb)).isEmpty();
assertThat(logScannerStatus.getBucketOffset(tb)).isEqualTo(20L);
assertThat(completedFetch.isConsumed()).isTrue();
+ // Empty record list, but bucket exposed via buckets() with an
advanced consumedUpToOffset.
+ assertThat(scanRecords.buckets()).contains(tb);
+ assertThat(scanRecords.consumedUpToOffset(tb)).isEqualTo(20L);
}
private DefaultCompletedFetch makeCompletedFetch(
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherITCase.java
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherITCase.java
index 50addfcfe..147be7922 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherITCase.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherITCase.java
@@ -149,7 +149,10 @@ public class LogFetcherITCase extends
ClientToServerITCaseBase {
assertThat(logFetcher.getCompletedFetchesSize()).isEqualTo(2);
});
ScanRecords records = logFetcher.collectFetch();
- assertThat(records.buckets().size()).isEqualTo(1);
+ // Both polled buckets are exposed; tb1 was polled but produced no
records.
+ TableBucket tb1 = new TableBucket(tableId, bucketId1);
+ assertThat(records.buckets()).containsExactlyInAnyOrder(tb0, tb1);
+ assertThat(records.records(tb1)).isEmpty();
List<ScanRecord> scanRecords = records.records(tb0);
assertThat(scanRecords.stream().map(ScanRecord::getRow).collect(Collectors.toList()))
.isEqualTo(expectedRows);
@@ -195,7 +198,8 @@ public class LogFetcherITCase extends
ClientToServerITCaseBase {
assertThat(newSchemaLogFetcher.getCompletedFetchesSize()).isEqualTo(2);
});
records = newSchemaLogFetcher.collectFetch();
- assertThat(records.buckets().size()).isEqualTo(1);
+ assertThat(records.buckets()).containsExactlyInAnyOrder(tb0, tb1);
+ assertThat(records.records(tb1)).isEmpty();
assertThat(records.records(tb0)).hasSize(20);
scanRecords = records.records(tb0);
assertThat(scanRecords.stream().map(ScanRecord::getRow).collect(Collectors.toList()))
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/ScanRecordsTest.java
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/ScanRecordsTest.java
index db4a32667..49c6e9c42 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/ScanRecordsTest.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/ScanRecordsTest.java
@@ -25,6 +25,8 @@ import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
@@ -57,4 +59,46 @@ public class ScanRecordsTest {
}
assertThat(c).isEqualTo(4);
}
+
+ /**
+ * Verifies buckets(), isEmpty(), and consumedUpToOffset() semantics for
progress-only polls.
+ */
+ @Test
+ void bucketsAndIsEmptySemantics() {
+ TableBucket tb = new TableBucket(0L, 0);
+
+ // No records and no progress: both isEmpty() and buckets() must be
empty.
+ ScanRecords trulyEmpty = ScanRecords.EMPTY;
+ assertThat(trulyEmpty.isEmpty()).isTrue();
+ assertThat(trulyEmpty.buckets()).isEmpty();
+
+ // Progress-only round: isEmpty() stays true (no materialized records),
+ // but buckets() exposes the advanced buckets and consumedUpToOffset
carries the offset.
+ TableBucket emptyBucket = new TableBucket(0L, 1);
+ Map<TableBucket, List<ScanRecord>> progressRecords = new HashMap<>();
+ progressRecords.put(tb, Collections.emptyList());
+ progressRecords.put(emptyBucket, Collections.emptyList());
+ Map<TableBucket, Long> progressOffsets = new HashMap<>();
+ progressOffsets.put(tb, 42L);
+ progressOffsets.put(emptyBucket, 10L);
+ ScanRecords progressOnly = new ScanRecords(progressRecords,
progressOffsets);
+ assertThat(progressOnly.isEmpty()).isTrue();
+ assertThat(progressOnly.buckets()).containsExactlyInAnyOrder(tb,
emptyBucket);
+ assertThat(progressOnly.records(emptyBucket)).isEmpty();
+ assertThat(progressOnly.consumedUpToOffset(tb)).isEqualTo(42L);
+
assertThat(progressOnly.consumedUpToOffset(emptyBucket)).isEqualTo(10L);
+ assertThat(progressOnly.consumedUpToOffset(new TableBucket(0L,
99))).isNull();
+
+ // Materialized records present: isEmpty() flips to false;
+ // the legacy single-arg constructor has no consumedUpToOffset.
+ Map<TableBucket, List<ScanRecord>> matRecords = new HashMap<>();
+ matRecords.put(
+ tb,
+ Collections.singletonList(
+ new ScanRecord(0L, 1000L, ChangeType.INSERT, row(1,
"a"))));
+ ScanRecords withRecords = new ScanRecords(matRecords);
+ assertThat(withRecords.isEmpty()).isFalse();
+ assertThat(withRecords.buckets()).containsExactly(tb);
+ assertThat(withRecords.consumedUpToOffset(tb)).isNull();
+ }
}
diff --git
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java
index d59787e15..335f5114c 100644
---
a/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java
+++
b/fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java
@@ -355,63 +355,89 @@ public class TieringSplitReader<WriteResult>
Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults =
new HashMap<>();
Map<TableBucket, String> finishedSplitIds = new HashMap<>();
+ // Iterate every polled bucket, including those that only advanced
their offset.
for (TableBucket bucket : scanRecords.buckets()) {
- List<ScanRecord> bucketScanRecords = scanRecords.records(bucket);
- if (bucketScanRecords.isEmpty()) {
- continue;
- }
- // no any stopping offset, just skip handle the records for the
bucket
Long stoppingOffset = currentTableStoppingOffsets.get(bucket);
if (stoppingOffset == null) {
continue;
}
+
+ List<ScanRecord> records = scanRecords.records(bucket);
LakeWriter<WriteResult> lakeWriter = null;
- for (ScanRecord record : bucketScanRecords) {
- // if record is less than stopping offset
- if (record.logOffset() < stoppingOffset) {
- if (lakeWriter == null) {
- lakeWriter =
- getOrCreateLakeWriter(
- bucket,
-
currentTableSplitsByBucket.get(bucket).getPartitionName());
- }
- lakeWriter.write(record);
- if (record.getSizeInBytes() > 0) {
-
tieringMetrics.recordBytesRead(record.getSizeInBytes());
- }
+ ScanRecord lastRecord = null;
+
+ for (ScanRecord record : records) {
+ lastRecord = record;
+
+ // The scanner may return records beyond this split's
exclusive stopping offset.
+ // Those records belong to the next split and must not be
tiered here.
+ if (record.logOffset() >= stoppingOffset) {
+ continue;
}
+
+ if (lakeWriter == null) {
+ lakeWriter =
+ getOrCreateLakeWriter(
+ bucket,
+
currentTableSplitsByBucket.get(bucket).getPartitionName());
+ }
+ lakeWriter.write(record);
+ if (record.getSizeInBytes() > 0) {
+ tieringMetrics.recordBytesRead(record.getSizeInBytes());
+ }
+ }
+
+ // consumedUpToOffset is an exclusive upper bound: all offsets
before it have been
+ // consumed by the scanner in this poll round. It may advance even
when records is
+ // empty, for example when FIRST_ROW filters duplicate upserts
into empty WAL batches.
+ Long consumedUpToOffset = scanRecords.consumedUpToOffset(bucket);
+ checkState(
+ consumedUpToOffset != null,
+ "Missing consumed-up-to offset for polled bucket %s.",
+ bucket);
+
+ // The split owns offsets before stoppingOffset only. If the
scanner consumed past
+ // the split boundary, cap the tiered progress at stoppingOffset
so the next split
+ // still owns later data.
+ long tieredLogEndOffset = Math.min(consumedUpToOffset,
stoppingOffset);
+ long tieredTimestamp;
+ if (lastRecord != null) {
+ tieredTimestamp = lastRecord.timestamp();
+ } else {
+ LogOffsetAndTimestamp latest =
currentTableTieredOffsetAndTimestamp.get(bucket);
+ tieredTimestamp = latest != null ? latest.timestamp :
UNKNOWN_BUCKET_TIMESTAMP;
}
- ScanRecord lastRecord =
bucketScanRecords.get(bucketScanRecords.size() - 1);
currentTableTieredOffsetAndTimestamp.put(
- bucket,
- new LogOffsetAndTimestamp(lastRecord.logOffset(),
lastRecord.timestamp()));
- // has arrived into the end of the split,
- if (lastRecord.logOffset() >= stoppingOffset - 1) {
- currentTableStoppingOffsets.remove(bucket);
- if (bucket.getPartitionId() != null) {
- currentLogScanner.unsubscribe(bucket.getPartitionId(),
bucket.getBucket());
- } else {
- // todo: should unsubscribe the log split if unsubscribe
bucket for
- // un-partitioned table is supported
- }
- TieringSplit currentTieringSplit =
currentTableSplitsByBucket.remove(bucket);
- String currentSplitId = currentTieringSplit.splitId();
- // put write result of the bucket
- writeResults.put(
- bucket,
- completeLakeWriter(
- bucket,
- currentTieringSplit.getPartitionName(),
- stoppingOffset,
- lastRecord.timestamp()));
- // put split of the bucket
- finishedSplitIds.put(bucket, currentSplitId);
- LOG.info(
- "Finish tier bucket {} for table {}, split: {}.",
- bucket,
- currentTablePath,
- currentSplitId);
+ bucket, new LogOffsetAndTimestamp(tieredLogEndOffset - 1,
tieredTimestamp));
+
+ // The split owns offsets below stoppingOffset. If the scanner has
not consumed up to
+ // that exclusive bound yet, keep the split active.
+ if (consumedUpToOffset < stoppingOffset) {
+ continue;
}
+
+ currentTableStoppingOffsets.remove(bucket);
+ if (bucket.getPartitionId() != null) {
+ currentLogScanner.unsubscribe(bucket.getPartitionId(),
bucket.getBucket());
+ } else {
+ // todo: should unsubscribe the log split if unsubscribe
bucket for
+ // un-partitioned table is supported
+ }
+ TieringSplit currentTieringSplit =
currentTableSplitsByBucket.remove(bucket);
+ String currentSplitId = currentTieringSplit.splitId();
+ writeResults.put(
+ bucket,
+ completeLakeWriter(
+ bucket,
+ currentTieringSplit.getPartitionName(),
+ stoppingOffset,
+ tieredTimestamp));
+ finishedSplitIds.put(bucket, currentSplitId);
+ LOG.info(
+ "Finish tier bucket {} for table {}, split: {}.",
+ bucket,
+ currentTablePath,
+ currentSplitId);
}
if (!finishedSplitIds.isEmpty()) {
diff --git
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSplitReaderTest.java
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSplitReaderTest.java
index 171f521e0..9aad5d1c3 100644
---
a/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSplitReaderTest.java
+++
b/fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSplitReaderTest.java
@@ -24,6 +24,7 @@ import org.apache.fluss.client.table.writer.AppendWriter;
import org.apache.fluss.client.table.writer.TableWriter;
import org.apache.fluss.client.table.writer.UpsertWriter;
import org.apache.fluss.client.write.HashBucketAssigner;
+import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.flink.tiering.TestingLakeTieringFactory;
import org.apache.fluss.flink.tiering.TestingWriteResult;
import org.apache.fluss.flink.tiering.source.metrics.TieringMetrics;
@@ -33,12 +34,14 @@ import
org.apache.fluss.flink.tiering.source.split.TieringSplit;
import org.apache.fluss.flink.utils.FlinkTestBase;
import org.apache.fluss.lake.writer.LakeWriter;
import org.apache.fluss.lake.writer.WriterInitContext;
+import org.apache.fluss.metadata.MergeEngineType;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.record.LogRecord;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.row.encode.CompactedKeyEncoder;
+import org.apache.fluss.server.replica.Replica;
import org.apache.flink.api.connector.source.SourceSplit;
import org.apache.flink.connector.base.source.reader.RecordsWithSplitIds;
@@ -335,6 +338,93 @@ class TieringSplitReaderTest extends FlinkTestBase {
}
}
+ /**
+ * Verifies that the tiering service finishes under {@code first_row}
merge engine even when
+ * duplicate upserts produce empty WAL batches.
+ */
+ @Test
+ void testTieringFirstRowMergeEngineFinishes() throws Exception {
+ TablePath tablePath = TablePath.of("fluss",
"tiering_first_row_finish");
+ TableDescriptor descriptor =
+ TableDescriptor.builder()
+ .schema(DEFAULT_PK_TABLE_SCHEMA)
+ .distributedBy(DEFAULT_BUCKET_NUM, "id")
+ .property(ConfigOptions.TABLE_MERGE_ENGINE,
MergeEngineType.FIRST_ROW)
+ .build();
+ long tableId = createTable(tablePath, descriptor);
+
+ // Duplicate upserts under FIRST_ROW: only the first per id yields a
CDC
+ // record, the rest become empty WAL batches that still advance the
offset.
+ int distinctKeys = 5;
+ int duplicatesPerKey = 10;
+ try (Table table = conn.getTable(tablePath)) {
+ for (int round = 0; round < duplicatesPerKey; round++) {
+ UpsertWriter writer = table.newUpsert().createWriter();
+ for (int id = 0; id < distinctKeys; id++) {
+ writer.upsert(row(id, "v" + round));
+ }
+ writer.flush();
+ }
+ }
+
+ // Build log splits whose stoppingOffset equals the leader's current
logEndOffset.
+ List<TieringSplit> logSplits = new ArrayList<>();
+ Set<String> splitIds = new HashSet<>();
+ long totalLogEndOffset = 0L;
+ for (int bucket = 0; bucket < DEFAULT_BUCKET_NUM; bucket++) {
+ TableBucket tb = new TableBucket(tableId, bucket);
+ Replica leader =
FLUSS_CLUSTER_EXTENSION.waitAndGetLeaderReplica(tb);
+ long stoppingOffset = leader.getLogTablet().localLogEndOffset();
+ totalLogEndOffset += stoppingOffset;
+ if (stoppingOffset <= 0) {
+ continue;
+ }
+ TieringLogSplit split =
+ createLogSplit(tablePath, tableId, bucket,
EARLIEST_OFFSET, stoppingOffset);
+ logSplits.add(split);
+ splitIds.add(split.splitId());
+ }
+ assertThat(logSplits).isNotEmpty();
+ // Pre-condition: total log offsets must exceed distinct-key count,
otherwise
+ // no empty batch was produced.
+ assertThat(totalLogEndOffset)
+ .as(
+ "Expected logEndOffset (%d) to exceed distinctKeys
(%d) so that "
+ + "empty batches are produced under FIRST_ROW",
+ totalLogEndOffset, distinctKeys)
+ .isGreaterThan(distinctKeys);
+
+ try (Connection connection =
+ ConnectionFactory.createConnection(
+ FLUSS_CLUSTER_EXTENSION.getClientConfig());
+ TieringSplitReader<TestingWriteResult> tieringSplitReader =
+ createTieringReader(connection)) {
+ tieringSplitReader.handleSplitsChanges(new
SplitsAddition<>(logSplits));
+
+ // With the fix every split must finish within a few fetch rounds.
+ Set<String> finished = new HashSet<>();
+ int maxRounds = 10;
+ for (int i = 0; i < maxRounds && !finished.containsAll(splitIds);
i++) {
+
RecordsWithSplitIds<TableBucketWriteResult<TestingWriteResult>> fetchResult =
+ tieringSplitReader.fetch();
+ finished.addAll(fetchResult.finishedSplits());
+ // drain the iterator so that the reader advances internal
state
+ while (fetchResult.nextSplit() != null) {
+ while (fetchResult.nextRecordFromSplit() != null) {
+ // consume
+ }
+ }
+ }
+
+ assertThat(finished)
+ .as(
+ "All tiering splits must finish under FIRST_ROW
merge engine "
+ + "with duplicate keys. Finished: %s,
expected: %s",
+ finished, splitIds)
+ .containsAll(splitIds);
+ }
+ }
+
private TieringSplitReader<TestingWriteResult>
createTieringReader(Connection connection) {
final TieringMetrics tieringMetrics =
new TieringMetrics(
