Skip to content

[client] feat: remote manifest v2 client #3834

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
gyang94 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

[client] feat: remote manifest v2 client #3834

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
8ade347
feat: remote manifest v2
gyang94 Aug 2, 2026
4112180
feat: remote manifest v2
gyang94 Aug 2, 2026
d716b39
feat: fix rust test
gyang94 Aug 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 58 additions & 8 deletions fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ public abstract class CompletedFetch {
final long highWatermark;
private final long fetchOffset;
private final long filteredEndOffset;
private final long logicalEndOffset;

private final boolean isCheckCrcs;
private final Iterator<LogRecordBatch> batches;
Expand Down Expand Up @@ -92,6 +93,34 @@ public CompletedFetch(
boolean isCheckCrcs,
long fetchOffset,
long filteredEndOffset) {
this(
tableBucket,
tablePath,
error,
sizeInBytes,
highWatermark,
batches,
readContext,
logScannerStatus,
isCheckCrcs,
fetchOffset,
filteredEndOffset,
Long.MAX_VALUE);
}

public CompletedFetch(
TableBucket tableBucket,
TablePath tablePath,
ApiError error,
int sizeInBytes,
long highWatermark,
Iterator<LogRecordBatch> batches,
LogRecordReadContext readContext,
LogScannerStatus logScannerStatus,
boolean isCheckCrcs,
long fetchOffset,
long filteredEndOffset,
long logicalEndOffset) {
this.tableBucket = tableBucket;
this.tablePath = tablePath;
this.error = error;
Expand All @@ -110,6 +139,13 @@ public CompletedFetch(
fetchOffset,
tableBucket);
this.filteredEndOffset = filteredEndOffset;
checkArgument(
logicalEndOffset > fetchOffset,
"logicalEndOffset (%s) must be greater than fetchOffset (%s) for bucket %s.",
logicalEndOffset,
fetchOffset,
tableBucket);
this.logicalEndOffset = logicalEndOffset;
this.nextFetchOffset = fetchOffset;
}

Expand Down Expand Up @@ -370,15 +406,29 @@ private LogRecord nextFetchedRecord() throws Exception {

private LogRecordBatch nextFetchedBatch() {
maybeCloseRecordStream();
if (!batches.hasNext()) {
finishFetchedBatches();
return null;
while (batches.hasNext()) {
LogRecordBatch nextBatch = batches.next();
if (nextBatch.baseLogOffset() >= logicalEndOffset) {
finishFetchedBatches();
return null;
}
if (nextBatch.nextLogOffset() > logicalEndOffset) {
throw new FetchException(
"Remote logical end offset "
+ logicalEndOffset
+ " is not a batch boundary for bucket "
+ tableBucket);
}
if (nextBatch.nextLogOffset() <= nextFetchOffset) {
currentBatch = nextBatch;
continue;
}
currentBatch = nextBatch;
maybeEnsureValid(currentBatch);
return currentBatch;
}

currentBatch = batches.next();
// TODO get last epoch.
maybeEnsureValid(currentBatch);
return currentBatch;
finishFetchedBatches();
return null;
}

private void finishFetchedBatches() {
Expand Down
53 changes: 48 additions & 5 deletions fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,9 @@
import org.apache.fluss.record.LogRecords;
import org.apache.fluss.record.MemoryLogRecords;
import org.apache.fluss.remote.RemoteLogFetchInfo;
import org.apache.fluss.remote.RemoteLogFetchInfoV2;
import org.apache.fluss.remote.RemoteLogSegment;
import org.apache.fluss.remote.RemoteLogSegmentReference;
import org.apache.fluss.rpc.entity.FetchLogResultForBucket;
import org.apache.fluss.rpc.gateway.TabletServerGateway;
import org.apache.fluss.rpc.messages.FetchLogRequest;
Expand Down Expand Up @@ -459,11 +461,19 @@ private synchronized void handleFetchLogResponse(
tb);
} else {
if (fetchResultForBucket.fetchFromRemote()) {
pendRemoteFetches(
trc,
fetchResultForBucket.remoteLogFetchInfo(),
fetchOffset,
fetchResultForBucket.getHighWatermark());
if (fetchResultForBucket.remoteLogFetchInfoV2() != null) {
pendRemoteFetchesV2(
trc,
fetchResultForBucket.remoteLogFetchInfoV2(),
fetchOffset,
fetchResultForBucket.getHighWatermark());
} else {
pendRemoteFetches(
trc,
fetchResultForBucket.remoteLogFetchInfo(),
fetchOffset,
fetchResultForBucket.getHighWatermark());
}
} else {
LogRecords logRecords = fetchResultForBucket.recordsOrEmpty();
boolean hasRecords = !MemoryLogRecords.EMPTY.equals(logRecords);
Expand Down Expand Up @@ -561,6 +571,39 @@ private void pendRemoteFetches(
}
}

private void pendRemoteFetchesV2(
TableReadContext tableReadContext,
RemoteLogFetchInfoV2 remoteLogFetchInfo,
long firstFetchOffset,
long highWatermark) {
FsPath remoteLogTabletDir = new FsPath(remoteLogFetchInfo.remoteLogTabletDir());
List<RemoteLogSegmentReference> references = remoteLogFetchInfo.activeReferences();
for (int i = 0; i < references.size(); i++) {
RemoteLogSegmentReference reference = references.get(i);
RemoteLogSegment segment = reference.remoteLogSegment();
long fetchOffset =
i == 0
? Math.max(firstFetchOffset, reference.logicalStartOffset())
: reference.logicalStartOffset();
RemoteLogDownloadFuture downloadFuture =
remoteLogDownloader.requestRemoteLog(remoteLogTabletDir, segment);
RemotePendingFetch pendingFetch =
new RemotePendingFetch(
segment,
downloadFuture,
tableReadContext.tablePath,
0,
fetchOffset,
reference.logicalEndOffset(),
highWatermark,
tableReadContext.remoteReadContext,
logScannerStatus,
isCheckCrcs);
logFetchBuffer.pend(pendingFetch);
downloadFuture.onComplete(() -> logFetchBuffer.tryComplete(segment.tableBucket()));
}
}

@VisibleForTesting
Map<Integer, FetchLogRequest> prepareFetchLogRequests(List<TableBucket> fetchableBuckets) {
// Outer key: leaderId, inner key: tableId -> list of bucket requests
Expand Down
27 changes: 26 additions & 1 deletion ...-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemoteCompletedFetch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,30 @@ class RemoteCompletedFetch extends CompletedFetch {
boolean isCheckCrc,
long fetchOffset,
Runnable recycleCallback) {
this(
tableBucket,
tablePath,
fileLogRecords,
highWatermark,
readContext,
logScannerStatus,
isCheckCrc,
fetchOffset,
Long.MAX_VALUE,
recycleCallback);
}

RemoteCompletedFetch(
TableBucket tableBucket,
TablePath tablePath,
FileLogRecords fileLogRecords,
long highWatermark,
LogRecordReadContext readContext,
LogScannerStatus logScannerStatus,
boolean isCheckCrc,
long fetchOffset,
long logicalEndOffset,
Runnable recycleCallback) {
super(
tableBucket,
tablePath,
Expand All @@ -59,7 +83,8 @@ class RemoteCompletedFetch extends CompletedFetch {
logScannerStatus,
isCheckCrc,
fetchOffset,
CompletedFetch.NO_FILTERED_END_OFFSET);
CompletedFetch.NO_FILTERED_END_OFFSET,
logicalEndOffset);
this.fileLogRecords = fileLogRecords;
this.recycleCallback = recycleCallback;
}
Expand Down
27 changes: 27 additions & 0 deletions fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/RemotePendingFetch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@ class RemotePendingFetch implements PendingFetch {

private final int posInLogSegment;
private final long fetchOffset;
private final long logicalEndOffset;
private final long highWatermark;
private final LogRecordReadContext readContext;
private final LogScannerStatus logScannerStatus;
Expand All @@ -50,11 +51,36 @@ class RemotePendingFetch implements PendingFetch {
LogRecordReadContext readContext,
LogScannerStatus logScannerStatus,
boolean isCheckCrc) {
this(
remoteLogSegment,
downloadFuture,
tablePath,
posInLogSegment,
fetchOffset,
remoteLogSegment.remoteLogEndOffset(),
highWatermark,
readContext,
logScannerStatus,
isCheckCrc);
}

RemotePendingFetch(
RemoteLogSegment remoteLogSegment,
RemoteLogDownloadFuture downloadFuture,
TablePath tablePath,
int posInLogSegment,
long fetchOffset,
long logicalEndOffset,
long highWatermark,
LogRecordReadContext readContext,
LogScannerStatus logScannerStatus,
boolean isCheckCrc) {
this.remoteLogSegment = remoteLogSegment;
this.downloadFuture = downloadFuture;
this.tablePath = tablePath;
this.posInLogSegment = posInLogSegment;
this.fetchOffset = fetchOffset;
this.logicalEndOffset = logicalEndOffset;
this.highWatermark = highWatermark;
this.readContext = readContext;
this.logScannerStatus = logScannerStatus;
Expand Down Expand Up @@ -83,6 +109,7 @@ public CompletedFetch toCompletedFetch() {
logScannerStatus,
isCheckCrc,
fetchOffset,
logicalEndOffset,
downloadFuture.getRecycleCallback());
}

Expand Down
Loading
Loading