Skip to content

Commit 0afc21a

Browse files
Kaixuan-DuanKaixuan-Duan
authored
[client] Fix tiering hang on first_row merge engine empty batches (#3242)
1 parent 258e9f1 commit 0afc21a

10 files changed

Lines changed: 271 additions & 63 deletions

File tree

‎fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogFetchCollector.java‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -65,14 +65,15 @@ protected AbstractLogFetchCollector(
6565
/**
6666
* Return the fetched log records, empty the record buffer and update the consumed position.
6767
*
68-
* <p>NOTE: returning empty records guarantees the consumed position are NOT updated.
68+
* <p>NOTE: empty record lists may still advance the consumed position.
6969
*
7070
* @return The fetched records per partition
7171
* @throws FetchException If there is OffsetOutOfRange error in fetchResponse and the
7272
* defaultResetPolicy is NONE
7373
*/
7474
public R collectFetch(final LogFetchBuffer logFetchBuffer) {
7575
Map<TableBucket, List<T>> fetched = new HashMap<>();
76+
Map<TableBucket, Long> consumedUpToOffsets = new HashMap<>();
7677
int recordsRemaining = maxPollRecords;
7778

7879
try {
@@ -108,8 +109,11 @@ public R collectFetch(final LogFetchBuffer logFetchBuffer) {
108109
logFetchBuffer.poll();
109110
} else {
110111
List<T> records = fetchRecords(nextInLineFetch, recordsRemaining);
112+
TableBucket tableBucket = nextInLineFetch.tableBucket;
113+
// Always record the advanced next fetch offset for this bucket, even when
114+
// the materialized record list is empty.
115+
consumedUpToOffsets.put(tableBucket, nextInLineFetch.nextFetchOffset());
111116
if (!records.isEmpty()) {
112-
TableBucket tableBucket = nextInLineFetch.tableBucket;
113117
List<T> currentRecords = fetched.get(tableBucket);
114118
if (currentRecords == null) {
115119
fetched.put(tableBucket, records);
@@ -126,6 +130,8 @@ public R collectFetch(final LogFetchBuffer logFetchBuffer) {
126130
}
127131

128132
recordsRemaining -= recordCount(records);
133+
} else {
134+
fetched.putIfAbsent(tableBucket, Collections.emptyList());
129135
}
130136
}
131137
}
@@ -140,7 +146,7 @@ public R collectFetch(final LogFetchBuffer logFetchBuffer) {
140146
throw e;
141147
}
142148

143-
return toResult(fetched);
149+
return toResult(fetched, consumedUpToOffsets);
144150
}
145151

146152
/** Initialize a {@link CompletedFetch} object. */
@@ -293,7 +299,8 @@ protected List<T> fetchRecords(CompletedFetch nextInLineFetch, int maxRecords) {
293299

294300
protected abstract int recordCount(List<T> fetchedRecords);
295301

296-
protected abstract R toResult(Map<TableBucket, List<T>> fetchedRecords);
302+
protected abstract R toResult(
303+
Map<TableBucket, List<T>> fetchedRecords, Map<TableBucket, Long> consumedUpToOffsets);
297304

298305
/**
299306
* Release resources held by fetched records on failure. The default implementation is a no-op,

‎fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ArrowLogFetchCollector.java‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -62,7 +62,11 @@ protected int recordCount(List<ArrowBatchData> fetchedRecords) {
6262
}
6363

6464
@Override
65-
protected ArrowScanRecords toResult(Map<TableBucket, List<ArrowBatchData>> fetchedRecords) {
65+
protected ArrowScanRecords toResult(
66+
Map<TableBucket, List<ArrowBatchData>> fetchedRecords,
67+
Map<TableBucket, Long> consumedUpToOffsets) {
68+
// Arrow scan paths don't need consumedUpToOffsets (issue #2371 is specific to
69+
// row-based tiering), so it's discarded here rather than carried in ArrowScanRecords.
6670
return new ArrowScanRecords(fetchedRecords);
6771
}
6872

‎fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -67,7 +67,9 @@ protected int recordCount(List<ScanRecord> fetchedRecords) {
6767
}
6868

6969
@Override
70-
protected ScanRecords toResult(Map<TableBucket, List<ScanRecord>> fetchedRecords) {
71-
return new ScanRecords(fetchedRecords);
70+
protected ScanRecords toResult(
71+
Map<TableBucket, List<ScanRecord>> fetchedRecords,
72+
Map<TableBucket, Long> consumedUpToOffsets) {
73+
return new ScanRecords(fetchedRecords, consumedUpToOffsets);
7274
}
7375
}

‎fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerImpl.java‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -141,7 +141,11 @@ private Projection sanityProjection(@Nullable int[] projectedFields, TableInfo t
141141

142142
@Override
143143
public ScanRecords poll(Duration timeout) {
144-
return doPoll(timeout, this::pollForFetches, ScanRecords::isEmpty, () -> ScanRecords.EMPTY);
144+
return doPoll(
145+
timeout,
146+
this::pollForFetches,
147+
scanRecords -> scanRecords.buckets().isEmpty(),
148+
() -> ScanRecords.EMPTY);
145149
}
146150

147151
/**
@@ -250,7 +254,8 @@ public void wakeup() {
250254

251255
private ScanRecords pollForFetches() {
252256
ScanRecords scanRecords = logFetcher.collectFetch();
253-
if (!scanRecords.isEmpty()) {
257+
// Check buckets() (includes progress-only buckets).
258+
if (!scanRecords.buckets().isEmpty()) {
254259
return scanRecords;
255260
}
256261

‎fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java‎

Lines changed: 28 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,8 @@
2222
import org.apache.fluss.metadata.TableBucket;
2323
import org.apache.fluss.utils.AbstractIterator;
2424

25+
import javax.annotation.Nullable;
26+
2527
import java.util.Collections;
2628
import java.util.Iterator;
2729
import java.util.List;
@@ -41,8 +43,18 @@ public class ScanRecords implements Iterable<ScanRecord> {
4143

4244
private final Map<TableBucket, List<ScanRecord>> records;
4345

46+
/** The exclusive upper bound of consumed offsets per polled bucket in this round. */
47+
private final Map<TableBucket, Long> consumedUpToOffsets;
48+
4449
public ScanRecords(Map<TableBucket, List<ScanRecord>> records) {
50+
this(records, Collections.emptyMap());
51+
}
52+
53+
public ScanRecords(
54+
Map<TableBucket, List<ScanRecord>> records,
55+
Map<TableBucket, Long> consumedUpToOffsets) {
4556
this.records = records;
57+
this.consumedUpToOffsets = consumedUpToOffsets;
4658
}
4759

4860
/**
@@ -59,15 +71,25 @@ public List<ScanRecord> records(TableBucket scanBucket) {
5971
}
6072

6173
/**
62-
* Get the bucket ids which have records contained in this record set.
63-
*
64-
* @return the set of partitions with data in this record set (maybe empty if no data was
65-
* returned)
74+
* Get the bucket ids that were polled in this round, including buckets whose record list is
75+
* empty but whose log offset still advanced.
6676
*/
6777
public Set<TableBucket> buckets() {
6878
return Collections.unmodifiableSet(records.keySet());
6979
}
7080

81+
/**
82+
* Get the exclusive upper bound of offsets consumed for the given bucket in this poll round.
83+
*
84+
* @param bucket the bucket to query
85+
* @return the exclusive upper bound offset, or {@code null} if the bucket was not polled in
86+
* this round
87+
*/
88+
@Nullable
89+
public Long consumedUpToOffset(TableBucket bucket) {
90+
return consumedUpToOffsets.get(bucket);
91+
}
92+
7193
/** The number of records for all buckets. */
7294
public int count() {
7395
int count = 0;
@@ -77,8 +99,9 @@ public int count() {
7799
return count;
78100
}
79101

102+
/** Returns {@code true} if this {@code ScanRecords} contains no materialized records. */
80103
public boolean isEmpty() {
81-
return records.isEmpty();
104+
return count() == 0;
82105
}
83106

84107
@Override

‎fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -258,6 +258,9 @@ void testFilteredEmptyResponseAdvancesOffset() {
258258
assertThat(scanRecords.records(tb)).isEmpty();
259259
assertThat(logScannerStatus.getBucketOffset(tb)).isEqualTo(20L);
260260
assertThat(completedFetch.isConsumed()).isTrue();
261+
// Empty record list, but bucket exposed via buckets() with an advanced consumedUpToOffset.
262+
assertThat(scanRecords.buckets()).contains(tb);
263+
assertThat(scanRecords.consumedUpToOffset(tb)).isEqualTo(20L);
261264
}
262265

263266
private DefaultCompletedFetch makeCompletedFetch(

‎fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetcherITCase.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -149,7 +149,10 @@ void testFetchWithSchemaChange() throws Exception {
149149
assertThat(logFetcher.getCompletedFetchesSize()).isEqualTo(2);
150150
});
151151
ScanRecords records = logFetcher.collectFetch();
152-
assertThat(records.buckets().size()).isEqualTo(1);
152+
// Both polled buckets are exposed; tb1 was polled but produced no records.
153+
TableBucket tb1 = new TableBucket(tableId, bucketId1);
154+
assertThat(records.buckets()).containsExactlyInAnyOrder(tb0, tb1);
155+
assertThat(records.records(tb1)).isEmpty();
153156
List<ScanRecord> scanRecords = records.records(tb0);
154157
assertThat(scanRecords.stream().map(ScanRecord::getRow).collect(Collectors.toList()))
155158
.isEqualTo(expectedRows);
@@ -195,7 +198,8 @@ void testFetchWithSchemaChange() throws Exception {
195198
assertThat(newSchemaLogFetcher.getCompletedFetchesSize()).isEqualTo(2);
196199
});
197200
records = newSchemaLogFetcher.collectFetch();
198-
assertThat(records.buckets().size()).isEqualTo(1);
201+
assertThat(records.buckets()).containsExactlyInAnyOrder(tb0, tb1);
202+
assertThat(records.records(tb1)).isEmpty();
199203
assertThat(records.records(tb0)).hasSize(20);
200204
scanRecords = records.records(tb0);
201205
assertThat(scanRecords.stream().map(ScanRecord::getRow).collect(Collectors.toList()))

‎fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/ScanRecordsTest.java‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@
2525

2626
import java.util.ArrayList;
2727
import java.util.Arrays;
28+
import java.util.Collections;
29+
import java.util.HashMap;
2830
import java.util.Iterator;
2931
import java.util.LinkedHashMap;
3032
import java.util.List;
@@ -57,4 +59,46 @@ void iterator() {
5759
}
5860
assertThat(c).isEqualTo(4);
5961
}
62+
63+
/**
64+
* Verifies buckets(), isEmpty(), and consumedUpToOffset() semantics for progress-only polls.
65+
*/
66+
@Test
67+
void bucketsAndIsEmptySemantics() {
68+
TableBucket tb = new TableBucket(0L, 0);
69+
70+
// No records and no progress: both isEmpty() and buckets() must be empty.
71+
ScanRecords trulyEmpty = ScanRecords.EMPTY;
72+
assertThat(trulyEmpty.isEmpty()).isTrue();
73+
assertThat(trulyEmpty.buckets()).isEmpty();
74+
75+
// Progress-only round: isEmpty() stays true (no materialized records),
76+
// but buckets() exposes the advanced buckets and consumedUpToOffset carries the offset.
77+
TableBucket emptyBucket = new TableBucket(0L, 1);
78+
Map<TableBucket, List<ScanRecord>> progressRecords = new HashMap<>();
79+
progressRecords.put(tb, Collections.emptyList());
80+
progressRecords.put(emptyBucket, Collections.emptyList());
81+
Map<TableBucket, Long> progressOffsets = new HashMap<>();
82+
progressOffsets.put(tb, 42L);
83+
progressOffsets.put(emptyBucket, 10L);
84+
ScanRecords progressOnly = new ScanRecords(progressRecords, progressOffsets);
85+
assertThat(progressOnly.isEmpty()).isTrue();
86+
assertThat(progressOnly.buckets()).containsExactlyInAnyOrder(tb, emptyBucket);
87+
assertThat(progressOnly.records(emptyBucket)).isEmpty();
88+
assertThat(progressOnly.consumedUpToOffset(tb)).isEqualTo(42L);
89+
assertThat(progressOnly.consumedUpToOffset(emptyBucket)).isEqualTo(10L);
90+
assertThat(progressOnly.consumedUpToOffset(new TableBucket(0L, 99))).isNull();
91+
92+
// Materialized records present: isEmpty() flips to false;
93+
// the legacy single-arg constructor has no consumedUpToOffset.
94+
Map<TableBucket, List<ScanRecord>> matRecords = new HashMap<>();
95+
matRecords.put(
96+
tb,
97+
Collections.singletonList(
98+
new ScanRecord(0L, 1000L, ChangeType.INSERT, row(1, "a"))));
99+
ScanRecords withRecords = new ScanRecords(matRecords);
100+
assertThat(withRecords.isEmpty()).isFalse();
101+
assertThat(withRecords.buckets()).containsExactly(tb);
102+
assertThat(withRecords.consumedUpToOffset(tb)).isNull();
103+
}
60104
}

‎fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java‎

Lines changed: 73 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -355,63 +355,89 @@ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> forLogRecords(
355355
Map<TableBucket, TableBucketWriteResult<WriteResult>> writeResults = new HashMap<>();
356356
Map<TableBucket, String> finishedSplitIds = new HashMap<>();
357357

358+
// Iterate every polled bucket, including those that only advanced their offset.
358359
for (TableBucket bucket : scanRecords.buckets()) {
359-
List<ScanRecord> bucketScanRecords = scanRecords.records(bucket);
360-
if (bucketScanRecords.isEmpty()) {
361-
continue;
362-
}
363-
// no any stopping offset, just skip handle the records for the bucket
364360
Long stoppingOffset = currentTableStoppingOffsets.get(bucket);
365361
if (stoppingOffset == null) {
366362
continue;
367363
}
364+
365+
List<ScanRecord> records = scanRecords.records(bucket);
368366
LakeWriter<WriteResult> lakeWriter = null;
369-
for (ScanRecord record : bucketScanRecords) {
370-
// if record is less than stopping offset
371-
if (record.logOffset() < stoppingOffset) {
372-
if (lakeWriter == null) {
373-
lakeWriter =
374-
getOrCreateLakeWriter(
375-
bucket,
376-
currentTableSplitsByBucket.get(bucket).getPartitionName());
377-
}
378-
lakeWriter.write(record);
379-
if (record.getSizeInBytes() > 0) {
380-
tieringMetrics.recordBytesRead(record.getSizeInBytes());
381-
}
367+
ScanRecord lastRecord = null;
368+
369+
for (ScanRecord record : records) {
370+
lastRecord = record;
371+
372+
// The scanner may return records beyond this split's exclusive stopping offset.
373+
// Those records belong to the next split and must not be tiered here.
374+
if (record.logOffset() >= stoppingOffset) {
375+
continue;
382376
}
377+
378+
if (lakeWriter == null) {
379+
lakeWriter =
380+
getOrCreateLakeWriter(
381+
bucket,
382+
currentTableSplitsByBucket.get(bucket).getPartitionName());
383+
}
384+
lakeWriter.write(record);
385+
if (record.getSizeInBytes() > 0) {
386+
tieringMetrics.recordBytesRead(record.getSizeInBytes());
387+
}
388+
}
389+
390+
// consumedUpToOffset is an exclusive upper bound: all offsets before it have been
391+
// consumed by the scanner in this poll round. It may advance even when records is
392+
// empty, for example when FIRST_ROW filters duplicate upserts into empty WAL batches.
393+
Long consumedUpToOffset = scanRecords.consumedUpToOffset(bucket);
394+
checkState(
395+
consumedUpToOffset != null,
396+
"Missing consumed-up-to offset for polled bucket %s.",
397+
bucket);
398+
399+
// The split owns offsets before stoppingOffset only. If the scanner consumed past
400+
// the split boundary, cap the tiered progress at stoppingOffset so the next split
401+
// still owns later data.
402+
long tieredLogEndOffset = Math.min(consumedUpToOffset, stoppingOffset);
403+
long tieredTimestamp;
404+
if (lastRecord != null) {
405+
tieredTimestamp = lastRecord.timestamp();
406+
} else {
407+
LogOffsetAndTimestamp latest = currentTableTieredOffsetAndTimestamp.get(bucket);
408+
tieredTimestamp = latest != null ? latest.timestamp : UNKNOWN_BUCKET_TIMESTAMP;
383409
}
384-
ScanRecord lastRecord = bucketScanRecords.get(bucketScanRecords.size() - 1);
385410
currentTableTieredOffsetAndTimestamp.put(
386-
bucket,
387-
new LogOffsetAndTimestamp(lastRecord.logOffset(), lastRecord.timestamp()));
388-
// has arrived into the end of the split,
389-
if (lastRecord.logOffset() >= stoppingOffset - 1) {
390-
currentTableStoppingOffsets.remove(bucket);
391-
if (bucket.getPartitionId() != null) {
392-
currentLogScanner.unsubscribe(bucket.getPartitionId(), bucket.getBucket());
393-
} else {
394-
// todo: should unsubscribe the log split if unsubscribe bucket for
395-
// un-partitioned table is supported
396-
}
397-
TieringSplit currentTieringSplit = currentTableSplitsByBucket.remove(bucket);
398-
String currentSplitId = currentTieringSplit.splitId();
399-
// put write result of the bucket
400-
writeResults.put(
401-
bucket,
402-
completeLakeWriter(
403-
bucket,
404-
currentTieringSplit.getPartitionName(),
405-
stoppingOffset,
406-
lastRecord.timestamp()));
407-
// put split of the bucket
408-
finishedSplitIds.put(bucket, currentSplitId);
409-
LOG.info(
410-
"Finish tier bucket {} for table {}, split: {}.",
411-
bucket,
412-
currentTablePath,
413-
currentSplitId);
411+
bucket, new LogOffsetAndTimestamp(tieredLogEndOffset - 1, tieredTimestamp));
412+
413+
// The split owns offsets below stoppingOffset. If the scanner has not consumed up to
414+
// that exclusive bound yet, keep the split active.
415+
if (consumedUpToOffset < stoppingOffset) {
416+
continue;
414417
}
418+
419+
currentTableStoppingOffsets.remove(bucket);
420+
if (bucket.getPartitionId() != null) {
421+
currentLogScanner.unsubscribe(bucket.getPartitionId(), bucket.getBucket());
422+
} else {
423+
// todo: should unsubscribe the log split if unsubscribe bucket for
424+
// un-partitioned table is supported
425+
}
426+
TieringSplit currentTieringSplit = currentTableSplitsByBucket.remove(bucket);
427+
String currentSplitId = currentTieringSplit.splitId();
428+
writeResults.put(
429+
bucket,
430+
completeLakeWriter(
431+
bucket,
432+
currentTieringSplit.getPartitionName(),
433+
stoppingOffset,
434+
tieredTimestamp));
435+
finishedSplitIds.put(bucket, currentSplitId);
436+
LOG.info(
437+
"Finish tier bucket {} for table {}, split: {}.",
438+
bucket,
439+
currentTablePath,
440+
currentSplitId);
415441
}
416442

417443
if (!finishedSplitIds.isEmpty()) {

0 commit comments

Comments
 (0)