Skip to content

[client] Expose Arrow changelog batches #4006

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
jordepic wants to merge 2 commits into
base: main
Choose a base branch
from
Open

[client] Expose Arrow changelog batches #4006

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1d32bf5
[client] Expose Arrow changelog batches
Aug 15, 2026
a7e9371
[client] Address Arrow batch review comments
Aug 17, 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
4 changes: 2 additions & 2 deletions fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ArrowScanRecords.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package org.apache.fluss.client.table.scanner.log;

import org.apache.fluss.annotation.Internal;
import org.apache.fluss.annotation.PublicEvolving;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.record.ArrowBatchData;
import org.apache.fluss.utils.AbstractIterator;
Expand All @@ -38,7 +38,7 @@
* <p>Each {@link ArrowBatchData} holds off-heap Arrow memory. Callers should use try-with-resources
* on this container to ensure all batches are released if processing fails mid-iteration.
*/
@Internal
@PublicEvolving
public class ArrowScanRecords implements Iterable<ArrowBatchData>, AutoCloseable {
public static final ArrowScanRecords EMPTY = new ArrowScanRecords(Collections.emptyMap());

Expand Down
18 changes: 18 additions & 0 deletions fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScanner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,6 +56,24 @@ public interface LogScanner extends AutoCloseable {
*/
ScanRecords poll(Duration timeout);

/**
* Polls Arrow record batches from the tablet server without materializing individual rows.
*
* <p>This method is supported for tables whose log format is {@code ARROW}. Each returned batch
* owns its Arrow memory and, for primary-key tables, carries the stored per-row changelog
* types. Callers must close the returned {@link ArrowScanRecords} after processing it.
*
* @param timeout the timeout to poll
* @return the Arrow batches returned by this poll
* @throws UnsupportedOperationException if the table does not use the {@code ARROW} log format
* or the scanner implementation does not support Arrow record batch polling
* @throws IllegalStateException if the scanner is not subscribed to any buckets
*/
default ArrowScanRecords pollRecordBatch(Duration timeout) {
throw new UnsupportedOperationException(
"This LogScanner implementation does not support Arrow record batch polling");
}

/**
* Subscribe to the given table bucket in given offset dynamically. If the table bucket is
* already subscribed, the offset will be updated.
Expand Down
22 changes: 1 addition & 21 deletions fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScannerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,10 @@

package org.apache.fluss.client.table.scanner.log;

import org.apache.fluss.annotation.Internal;
import org.apache.fluss.annotation.PublicEvolving;
import org.apache.fluss.client.metadata.MetadataUpdater;
import org.apache.fluss.client.metrics.ScannerMetricGroup;
import org.apache.fluss.client.table.scanner.RemoteFileDownloader;
import org.apache.fluss.client.table.scanner.Scan;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.WakeupException;
import org.apache.fluss.metadata.LogFormat;
Expand Down Expand Up @@ -59,8 +57,6 @@ public class LogScannerImpl extends AbstractLogScanner<ScanRecords> implements L
private final long tableId;
private final boolean isPartitionedTable;
private final boolean isArrowLogFormat;
private final boolean isLogTable;
private final boolean hasProjection;

private final TableInfo tableInfo;
private final @Nullable Projection projection;
Expand Down Expand Up @@ -114,8 +110,6 @@ private LogScannerImpl(
this.tableId = tableInfo.getTableId();
this.isPartitionedTable = tableInfo.isPartitioned();
this.isArrowLogFormat = tableInfo.getTableConfig().getLogFormat() == LogFormat.ARROW;
this.isLogTable = !tableInfo.hasPrimaryKey();
this.hasProjection = projectedFields != null;
// add this table to metadata updater.
metadataUpdater.checkAndUpdateTableMetadata(Collections.singleton(tablePath));
this.metadataUpdater = metadataUpdater;
Expand Down Expand Up @@ -164,26 +158,12 @@ protected String notSubscribedMessage() {
return "LogScanner is not subscribed any buckets.";
}

/**
* Polls Arrow record batches for internal callers.
*
* <p>This method is intentionally kept off the public {@link Scan} API surface for now.
*/
@Internal
@Override
public ArrowScanRecords pollRecordBatch(Duration timeout) {
if (!isArrowLogFormat) {
throw new UnsupportedOperationException(
"Arrow record batch polling is only supported for tables whose log format is ARROW.");
}
if (!isLogTable) {
throw new UnsupportedOperationException(
"Arrow record batch polling is only supported for log tables. CDC scanning is not supported.");
}
if (hasProjection) {
throw new UnsupportedOperationException(
"Arrow record batch polling does not support projection. Please create the scanner without projection.");
}

acquireAndEnsureOpen();
try {
if (!logScannerStatus.prepareToPoll()) {
Expand Down
138 changes: 135 additions & 3 deletions fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogScannerITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;

import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
Expand Down Expand Up @@ -506,6 +507,65 @@ void testSubscribeOutOfRangeLog() throws Exception {
}
}

@Test
void testPollArrowBatchesWithPrimaryKeyChangelog() throws Exception {
TablePath tablePath = TablePath.of("test_db_1", "test_arrow_batches_with_changelog");
TableDescriptor tableDescriptor =
TableDescriptor.builder()
.schema(
Schema.newBuilder()
.column("a", DataTypes.INT())
.column("b", DataTypes.STRING())
.primaryKey("a")
.build())
.distributedBy(1)
.logFormat(LogFormat.ARROW)
.build();
createTable(tablePath, tableDescriptor, false);

try (Table table = conn.getTable(tablePath)) {
UpsertWriter writer = table.newUpsert().createWriter();
writer.upsert(row(1, "old")).get();
writer.flush();
writer.upsert(row(1, "new")).get();
writer.flush();
writer.upsert(row(2, "deleted")).get();
writer.flush();
writer.delete(row(2, "deleted")).get();
writer.flush();

ChangeType[] expectedChangeTypes = {
ChangeType.INSERT,
ChangeType.UPDATE_BEFORE,
ChangeType.UPDATE_AFTER,
ChangeType.INSERT,
ChangeType.DELETE
};
int[] expectedKeys = {1, 1, 1, 2, 2};
String[] expectedValues = {"old", "old", "new", "deleted", "deleted"};

try (LogScanner scanner = table.newScan().createLogScanner()) {
scanner.subscribeFromBeginning(0);
pollAndVerifyChangelogArrowBatches(
scanner, expectedChangeTypes, expectedKeys, expectedValues, 0);
}

try (LogScanner scanner = table.newScan().project(new int[] {1}).createLogScanner()) {
scanner.subscribeFromBeginning(0);
pollAndVerifyProjectedChangelogArrowBatches(
scanner, expectedChangeTypes, expectedValues);
}

// Offset 2 starts in the middle of the update batch and verifies that slicing keeps
// the change-type vector aligned with the Arrow rows.
try (LogScanner scanner = table.newScan().createLogScanner()) {
scanner.subscribe(0, 2L);
pollAndVerifyChangelogArrowBatches(
scanner, expectedChangeTypes, expectedKeys, expectedValues, 2);
}
}
}

@Test
void testPollArrowBatchesWithSchemaEvolution() throws Exception {
TablePath tablePath = TablePath.of("test_db_1", "test_arrow_batches_with_schema_evolution");
Expand Down Expand Up @@ -548,15 +608,15 @@ void testPollArrowBatchesWithSchemaEvolution() throws Exception {

int totalRecords = 6;
// subscribe from beginning and verify all 6 records
try (LogScannerImpl scanner = (LogScannerImpl) table.newScan().createLogScanner()) {
try (LogScanner scanner = table.newScan().createLogScanner()) {
scanner.subscribeFromBeginning(0);
pollAndVerifyArrowBatches(scanner, totalRecords, 0);
}

// subscribe from the middle of the first batch (offset 1)
// to ensure records before the subscribe offset are not returned
int subscribeOffset = 1;
try (LogScannerImpl scanner2 = (LogScannerImpl) table.newScan().createLogScanner()) {
try (LogScanner scanner2 = table.newScan().createLogScanner()) {
scanner2.subscribe(0, subscribeOffset);
pollAndVerifyArrowBatches(
scanner2, totalRecords - subscribeOffset, subscribeOffset);
Expand All @@ -565,7 +625,7 @@ void testPollArrowBatchesWithSchemaEvolution() throws Exception {
}

private void pollAndVerifyArrowBatches(
LogScannerImpl scanner, int expectedRecords, int minExpectedOffset) {
LogScanner scanner, int expectedRecords, int minExpectedOffset) {
int count = 0;
long deadline = System.nanoTime() + Duration.ofSeconds(30).toNanos();
while (count < expectedRecords) {
Expand Down Expand Up @@ -601,6 +661,78 @@ private void pollAndVerifyArrowBatches(
assertThat(count).isEqualTo(expectedRecords);
}

private void pollAndVerifyChangelogArrowBatches(
LogScanner scanner,
ChangeType[] expectedChangeTypes,
int[] expectedKeys,
String[] expectedValues,
int startingOffset) {
int count = startingOffset;
long deadline = System.nanoTime() + Duration.ofSeconds(30).toNanos();
while (count < expectedChangeTypes.length) {
assertThat(System.nanoTime())
.as(
"Timed out waiting for %s changelog records, got %s",
expectedChangeTypes.length - startingOffset, count - startingOffset)
.isLessThan(deadline);
try (ArrowScanRecords records = scanner.pollRecordBatch(Duration.ofSeconds(1))) {
for (ArrowBatchData batch : records) {
try (ArrowBatchData b = batch) {
assertThat(b.isAppendOnly()).isFalse();
ByteBuffer changeTypes = b.getChangeTypes().get();
assertThat(changeTypes.remaining()).isEqualTo(b.getRecordCount());
IntVector keys = (IntVector) b.getVectorSchemaRoot().getVector(0);
VarCharVector values = (VarCharVector) b.getVectorSchemaRoot().getVector(1);
for (int rowId = 0; rowId < b.getRecordCount(); rowId++) {
long offset = b.getBaseLogOffset() + rowId;
assertThat(offset).isEqualTo(count);
assertThat(b.getChangeType(rowId))
.isEqualTo(expectedChangeTypes[count]);
assertThat(changeTypes.get(rowId))
.isEqualTo(expectedChangeTypes[count].toByteValue());
assertThat(keys.get(rowId)).isEqualTo(expectedKeys[count]);
assertThat(values.getObject(rowId).toString())
.isEqualTo(expectedValues[count]);
count++;
}
}
}
}
}
assertThat(count).isEqualTo(expectedChangeTypes.length);
}

private void pollAndVerifyProjectedChangelogArrowBatches(
LogScanner scanner, ChangeType[] expectedChangeTypes, String[] expectedValues) {
int count = 0;
long deadline = System.nanoTime() + Duration.ofSeconds(30).toNanos();
while (count < expectedChangeTypes.length) {
assertThat(System.nanoTime())
.as(
"Timed out waiting for %s projected changelog records, got %s",
expectedChangeTypes.length, count)
.isLessThan(deadline);
try (ArrowScanRecords records = scanner.pollRecordBatch(Duration.ofSeconds(1))) {
for (ArrowBatchData batch : records) {
try (ArrowBatchData b = batch) {
assertThat(b.getVectorSchemaRoot().getFieldVectors()).hasSize(1);
VarCharVector values = (VarCharVector) b.getVectorSchemaRoot().getVector(0);
for (int rowId = 0; rowId < b.getRecordCount(); rowId++) {
long offset = b.getBaseLogOffset() + rowId;
assertThat(offset).isEqualTo(count);
assertThat(b.getChangeType(rowId))
.isEqualTo(expectedChangeTypes[count]);
assertThat(values.getObject(rowId).toString())
.isEqualTo(expectedValues[count]);
count++;
}
}
}
}
}
assertThat(count).isEqualTo(expectedChangeTypes.length);
}

private static ScanRecords pollUntilProgressOnly(
LogScanner logScanner, TableBucket tableBucket) {
long deadline = System.nanoTime() + Duration.ofSeconds(30).toNanos();
Expand Down
Loading
Loading