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 e32ba4709 [client] Support log scanner scan to arrow record batch
(#2995)
e32ba4709 is described below
commit e32ba47099fb1ea03e3565ce6db903495d58a350
Author: yuxia Luo <[email protected]>
AuthorDate: Thu Jun 4 10:23:21 2026 +0800
[client] Support log scanner scan to arrow record batch (#2995)
---
fluss-client/pom.xml | 18 ++
...llector.java => AbstractLogFetchCollector.java} | 192 ++++++++--------
.../table/scanner/log/ArrowLogFetchCollector.java | 84 +++++++
.../client/table/scanner/log/ArrowScanRecords.java | 123 +++++++++++
.../client/table/scanner/log/CompletedFetch.java | 130 +++++++++--
.../table/scanner/log/LogFetchCollector.java | 245 +--------------------
.../fluss/client/table/scanner/log/LogFetcher.java | 7 +
.../client/table/scanner/log/LogScannerImpl.java | 157 ++++++++-----
.../client/table/scanner/log/LogScannerITCase.java | 101 +++++++++
fluss-common/pom.xml | 18 ++
.../UnshadedArrowCompressionFactory.java | 60 +++++
.../UnshadedLz4ArrowCompressionCodec.java | 93 ++++++++
.../UnshadedZstdArrowCompressionCodec.java | 112 ++++++++++
.../org/apache/fluss/record/ArrowBatchData.java | 103 +++++++++
.../fluss/record/ArrowRecordBatchContext.java | 43 ++++
.../apache/fluss/record/DefaultLogRecordBatch.java | 40 +++-
.../apache/fluss/record/FileLogInputStream.java | 5 +
.../org/apache/fluss/record/LogRecordBatch.java | 10 +
.../apache/fluss/record/LogRecordReadContext.java | 120 +++++++++-
.../fluss/record/MemoryLogRecordsArrowBuilder.java | 12 +
.../fluss/record/UnshadedFlussVectorLoader.java | 132 +++++++++++
.../java/org/apache/fluss/row/ProjectedRow.java | 10 +-
.../apache/fluss/utils/UnshadedArrowReadUtils.java | 122 ++++++++++
.../fluss/record/FileLogInputStreamTest.java | 50 +++++
.../org/apache/fluss/testutils/DataTestUtils.java | 61 ++++-
25 files changed, 1633 insertions(+), 415 deletions(-)
diff --git a/fluss-client/pom.xml b/fluss-client/pom.xml
index 8e4ec956e..df80059c2 100644
--- a/fluss-client/pom.xml
+++ b/fluss-client/pom.xml
@@ -49,6 +49,24 @@
<version>${project.version}</version>
</dependency>
+ <!-- Unshaded Arrow dependencies for the Arrow scan batch API
(ArrowBatchData).
+ Marked as "provided" to avoid Arrow class conflicts with
consumers that
+ bring their own Arrow version. Currently only the internal
tiering service
+ uses this API and is responsible for including Arrow on its
classpath. -->
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-vector</artifactId>
+ <version>${arrow.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-memory-netty</artifactId>
+ <version>${arrow.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
<!-- test dependency -->
<dependency>
<groupId>org.apache.fluss</groupId>
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/AbstractLogFetchCollector.java
similarity index 76%
copy from
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java
copy to
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogFetchCollector.java
index 05c43eca0..f4daf890e 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/AbstractLogFetchCollector.java
@@ -19,21 +19,16 @@ package org.apache.fluss.client.table.scanner.log;
import org.apache.fluss.annotation.Internal;
import org.apache.fluss.client.metadata.MetadataUpdater;
-import org.apache.fluss.client.table.scanner.ScanRecord;
import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.exception.AuthorizationException;
import org.apache.fluss.exception.FetchException;
-import org.apache.fluss.exception.LogOffsetOutOfRangeException;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
-import org.apache.fluss.record.LogRecord;
-import org.apache.fluss.record.LogRecordBatch;
import org.apache.fluss.rpc.protocol.ApiError;
import org.apache.fluss.rpc.protocol.Errors;
import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
@@ -44,30 +39,23 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
-/* This file is based on source code of Apache Kafka Project
(https://kafka.apache.org/), licensed by the Apache
- * Software Foundation (ASF) under the Apache License, Version 2.0. See the
NOTICE file distributed with this work for
- * additional information regarding copyright ownership. */
-
-/**
- * {@link LogFetchCollector} operates at the {@link LogRecordBatch} level, as
that is what is stored
- * in the {@link LogFetchBuffer}. Each {@link LogRecord} in the {@link
LogRecordBatch} is converted
- * to a {@link ScanRecord} and added to the returned {@link LogFetcher}.
- */
+/** Shared implementation for polling completed fetches into scanner results.
*/
@ThreadSafe
@Internal
-public class LogFetchCollector {
- private static final Logger LOG =
LoggerFactory.getLogger(LogFetchCollector.class);
-
- private final TablePath tablePath;
- private final LogScannerStatus logScannerStatus;
+abstract class AbstractLogFetchCollector<T, R> {
+ protected final Logger log;
+ protected final TablePath tablePath;
+ protected final LogScannerStatus logScannerStatus;
private final int maxPollRecords;
private final MetadataUpdater metadataUpdater;
- public LogFetchCollector(
+ protected AbstractLogFetchCollector(
+ Logger log,
TablePath tablePath,
LogScannerStatus logScannerStatus,
Configuration conf,
MetadataUpdater metadataUpdater) {
+ this.log = log;
this.tablePath = tablePath;
this.logScannerStatus = logScannerStatus;
this.maxPollRecords =
conf.getInt(ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS);
@@ -80,11 +68,11 @@ public class LogFetchCollector {
* <p>NOTE: returning empty records guarantees the consumed position are
NOT updated.
*
* @return The fetched records per partition
- * @throws LogOffsetOutOfRangeException If there is OffsetOutOfRange error
in fetchResponse and
- * the defaultResetPolicy is NONE
+ * @throws FetchException If there is OffsetOutOfRange error in
fetchResponse and the
+ * defaultResetPolicy is NONE
*/
- public ScanRecords collectFetch(final LogFetchBuffer logFetchBuffer) {
- Map<TableBucket, List<ScanRecord>> fetched = new HashMap<>();
+ public R collectFetch(final LogFetchBuffer logFetchBuffer) {
+ Map<TableBucket, List<T>> fetched = new HashMap<>();
int recordsRemaining = maxPollRecords;
try {
@@ -106,8 +94,8 @@ public class LogFetchCollector {
} catch (Exception e) {
// Remove a completedFetch upon a parse with
exception if
// (1) it contains no records, and
- // (2) there are no fetched records with actual
content preceding this
- // exception.
+ // (2) there are no fetched records with actual
content
+ // preceding this exception.
if (fetched.isEmpty() &&
completedFetch.sizeInBytes == 0) {
logFetchBuffer.poll();
}
@@ -119,25 +107,25 @@ public class LogFetchCollector {
logFetchBuffer.poll();
} else {
- List<ScanRecord> records = fetchRecords(nextInLineFetch,
recordsRemaining);
+ List<T> records = fetchRecords(nextInLineFetch,
recordsRemaining);
if (!records.isEmpty()) {
TableBucket tableBucket = nextInLineFetch.tableBucket;
- List<ScanRecord> currentRecords =
fetched.get(tableBucket);
+ List<T> currentRecords = fetched.get(tableBucket);
if (currentRecords == null) {
fetched.put(tableBucket, records);
} else {
- // this case shouldn't usually happen because we
only send one fetch at
- // a time per bucket, but it might conceivably
happen in some rare
+ // this case shouldn't usually happen because we
only send one fetch
+ // at a time per bucket, but it might conceivably
happen in some rare
// cases (such as bucket leader changes). we have
to copy to a new list
// because the old one may be immutable
- List<ScanRecord> newScanRecords =
+ List<T> mergedRecords =
new ArrayList<>(records.size() +
currentRecords.size());
- newScanRecords.addAll(currentRecords);
- newScanRecords.addAll(records);
- fetched.put(tableBucket, newScanRecords);
+ mergedRecords.addAll(currentRecords);
+ mergedRecords.addAll(records);
+ fetched.put(tableBucket, mergedRecords);
}
- recordsRemaining -= records.size();
+ recordsRemaining -= recordCount(records);
}
}
}
@@ -145,58 +133,19 @@ public class LogFetchCollector {
if (fetched.isEmpty()) {
throw e;
}
+ } catch (Exception e) {
+ // Release any off-heap resources (e.g. Arrow buffers) held by
+ // already-fetched records before propagating the unexpected error.
+ closeFetchedRecords(fetched);
+ throw e;
}
- return new ScanRecords(fetched);
- }
-
- private List<ScanRecord> fetchRecords(CompletedFetch nextInLineFetch, int
maxRecords) {
- TableBucket tb = nextInLineFetch.tableBucket;
- Long offset = logScannerStatus.getBucketOffset(tb);
- if (offset == null) {
- LOG.debug(
- "Ignoring fetched records for {} at offset {} since the
current offset is null which means the "
- + "bucket has been unsubscribe.",
- tb,
- nextInLineFetch.fetchOffset());
- } else {
- if (nextInLineFetch.nextFetchOffset() == offset) {
- List<ScanRecord> records =
nextInLineFetch.fetchRecords(maxRecords);
- LOG.trace(
- "Returning {} fetched records at offset {} for
assigned bucket {}.",
- records.size(),
- offset,
- tb);
-
- if (nextInLineFetch.nextFetchOffset() > offset) {
- LOG.trace(
- "Updating fetch offset from {} to {} for bucket {}
and returning {} records from poll()",
- offset,
- nextInLineFetch.nextFetchOffset(),
- tb,
- records.size());
- logScannerStatus.updateOffset(tb,
nextInLineFetch.nextFetchOffset());
- }
- return records;
- } else {
- // these records aren't next in line based on the last
consumed offset, ignore them
- // they must be from an obsolete request
- LOG.warn(
- "Ignoring fetched records for {} at offset {} since
the current offset is {}",
- nextInLineFetch.tableBucket,
- nextInLineFetch.nextFetchOffset(),
- offset);
- }
- }
-
- LOG.trace("Draining fetched records for bucket {}",
nextInLineFetch.tableBucket);
- nextInLineFetch.drain();
-
- return Collections.emptyList();
+ return toResult(fetched);
}
/** Initialize a {@link CompletedFetch} object. */
- private @Nullable CompletedFetch initialize(CompletedFetch completedFetch)
{
+ @Nullable
+ private CompletedFetch initialize(CompletedFetch completedFetch) {
TableBucket tb = completedFetch.tableBucket;
ApiError error = completedFetch.error;
@@ -219,20 +168,19 @@ public class LogFetchCollector {
private @Nullable CompletedFetch handleInitializeSuccess(CompletedFetch
completedFetch) {
TableBucket tb = completedFetch.tableBucket;
- long fetchOffset = completedFetch.fetchOffset();
+ long fetchOffset = completedFetch.nextFetchOffset();
// we are interested in this fetch only if the beginning offset
matches the
// current consumed position.
Long offset = logScannerStatus.getBucketOffset(tb);
if (offset == null) {
- LOG.debug(
- "Discarding stale fetch response for bucket {} since the
expected offset is null which means the bucket has been "
- + "unsubscribed.",
+ log.debug(
+ "Discarding stale fetch response for bucket {} since the
expected offset is null which means the bucket has been unsubscribed.",
tb);
return null;
}
if (offset != fetchOffset) {
- LOG.warn(
+ log.warn(
"Discarding stale fetch response for bucket {} since its
offset {} does not match the expected offset {}.",
tb,
fetchOffset,
@@ -242,7 +190,7 @@ public class LogFetchCollector {
long highWatermark = completedFetch.highWatermark;
if (highWatermark >= 0) {
- LOG.trace("Updating high watermark for bucket {} to {}.", tb,
highWatermark);
+ log.trace("Updating high watermark for bucket {} to {}.", tb,
highWatermark);
logScannerStatus.updateHighWatermark(tb, highWatermark);
}
@@ -253,20 +201,20 @@ public class LogFetchCollector {
private void handleInitializeErrors(
CompletedFetch completedFetch, Errors error, String errorMessage) {
TableBucket tb = completedFetch.tableBucket;
- long fetchOffset = completedFetch.fetchOffset();
+ long fetchOffset = completedFetch.nextFetchOffset();
if (error == Errors.NOT_LEADER_OR_FOLLOWER
|| error == Errors.LOG_STORAGE_EXCEPTION
|| error == Errors.KV_STORAGE_EXCEPTION
|| error == Errors.STORAGE_EXCEPTION
|| error == Errors.FENCED_LEADER_EPOCH_EXCEPTION) {
- LOG.debug(
+ log.debug(
"Error in fetch for bucket {}: {}:{}",
tb,
error.exceptionName(),
error.exception(errorMessage));
metadataUpdater.checkAndUpdateMetadata(tablePath, tb);
} else if (error == Errors.UNKNOWN_TABLE_OR_BUCKET_EXCEPTION) {
- LOG.warn("Received unknown table or bucket error in fetch for
bucket {}", tb);
+ log.warn("Received unknown table or bucket error in fetch for
bucket {}", tb);
metadataUpdater.checkAndUpdateMetadata(tablePath, tb);
} else if (error == Errors.LOG_OFFSET_OUT_OF_RANGE_EXCEPTION) {
throw new FetchException(
@@ -276,7 +224,7 @@ public class LogFetchCollector {
} else if (error == Errors.AUTHORIZATION_EXCEPTION) {
throw new AuthorizationException(errorMessage);
} else if (error == Errors.UNKNOWN_SERVER_ERROR) {
- LOG.warn(
+ log.warn(
"Unknown server error while fetching offset {} for bucket
{}: {}",
fetchOffset,
tb,
@@ -293,4 +241,64 @@ public class LogFetchCollector {
error, fetchOffset, tb,
error.exception(errorMessage)));
}
}
+
+ protected List<T> fetchRecords(CompletedFetch nextInLineFetch, int
maxRecords) {
+ TableBucket tb = nextInLineFetch.tableBucket;
+ Long offset = logScannerStatus.getBucketOffset(tb);
+ if (offset == null) {
+ log.debug(
+ "Ignoring fetched records for {} at offset {} since the
current offset is null which means the bucket has been unsubscribed.",
+ tb,
+ nextInLineFetch.fetchOffset());
+ } else {
+ if (nextInLineFetch.nextFetchOffset() == offset) {
+ List<T> records = doFetchRecords(nextInLineFetch, maxRecords);
+ log.trace(
+ "Returning {} fetched records at offset {} for
assigned bucket {}.",
+ records.size(),
+ offset,
+ tb);
+
+ if (nextInLineFetch.nextFetchOffset() > offset) {
+ log.trace(
+ "Updating fetch offset from {} to {} for bucket {}
and returning {} records from poll()",
+ offset,
+ nextInLineFetch.nextFetchOffset(),
+ tb,
+ records.size());
+ logScannerStatus.updateOffset(tb,
nextInLineFetch.nextFetchOffset());
+ }
+ return records;
+ } else {
+ // these records aren't next in line based on the last
consumed offset, ignore them
+ // they must be from an obsolete request
+ log.warn(
+ "Ignoring fetched records for {} at offset {} since
the current offset is {}",
+ nextInLineFetch.tableBucket,
+ nextInLineFetch.nextFetchOffset(),
+ offset);
+ }
+ }
+
+ log.trace("Draining fetched records for bucket {}",
nextInLineFetch.tableBucket);
+ nextInLineFetch.drain();
+ return Collections.emptyList();
+ }
+
+ /**
+ * Fetch records from the given {@link CompletedFetch}. Subclasses
implement this to call the
+ * appropriate method on {@link CompletedFetch} for their record type.
+ */
+ protected abstract List<T> doFetchRecords(CompletedFetch nextInLineFetch,
int maxRecords);
+
+ protected abstract int recordCount(List<T> fetchedRecords);
+
+ protected abstract R toResult(Map<TableBucket, List<T>> fetchedRecords);
+
+ /**
+ * Release resources held by fetched records on failure. The default
implementation is a no-op,
+ * suitable for record types that are plain Java objects. Subclasses whose
record types hold
+ * off-heap resources (e.g. Arrow buffers) should override this to close
them.
+ */
+ protected void closeFetchedRecords(Map<TableBucket, List<T>> fetched) {}
}
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
new file mode 100644
index 000000000..39b6aac02
--- /dev/null
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ArrowLogFetchCollector.java
@@ -0,0 +1,84 @@
+/*
+ * 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.fluss.client.table.scanner.log;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.client.metadata.MetadataUpdater;
+import org.apache.fluss.config.Configuration;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ArrowBatchData;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.concurrent.ThreadSafe;
+
+import java.util.List;
+import java.util.Map;
+
+/** Collects Arrow batches from completed fetches. */
+@ThreadSafe
+@Internal
+public class ArrowLogFetchCollector
+ extends AbstractLogFetchCollector<ArrowBatchData, ArrowScanRecords> {
+ private static final Logger LOG =
LoggerFactory.getLogger(ArrowLogFetchCollector.class);
+
+ public ArrowLogFetchCollector(
+ TablePath tablePath,
+ LogScannerStatus logScannerStatus,
+ Configuration conf,
+ MetadataUpdater metadataUpdater) {
+ super(LOG, tablePath, logScannerStatus, conf, metadataUpdater);
+ }
+
+ @Override
+ protected List<ArrowBatchData> doFetchRecords(CompletedFetch
nextInLineFetch, int maxRecords) {
+ return nextInLineFetch.fetchArrowBatches(maxRecords);
+ }
+
+ @Override
+ protected int recordCount(List<ArrowBatchData> fetchedRecords) {
+ int count = 0;
+ for (ArrowBatchData fetchedRecord : fetchedRecords) {
+ count += fetchedRecord.getRecordCount();
+ }
+ return count;
+ }
+
+ @Override
+ protected ArrowScanRecords toResult(Map<TableBucket, List<ArrowBatchData>>
fetchedRecords) {
+ return new ArrowScanRecords(fetchedRecords);
+ }
+
+ @Override
+ protected void closeFetchedRecords(Map<TableBucket, List<ArrowBatchData>>
fetched) {
+ for (List<ArrowBatchData> batches : fetched.values()) {
+ for (ArrowBatchData batch : batches) {
+ try {
+ batch.close();
+ } catch (Exception e) {
+ LOG.warn(
+ "Failed to close Arrow batch during cleanup for
table {}",
+ tablePath,
+ e);
+ }
+ }
+ }
+ }
+}
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ArrowScanRecords.java
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ArrowScanRecords.java
new file mode 100644
index 000000000..d5cb63a8e
--- /dev/null
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ArrowScanRecords.java
@@ -0,0 +1,123 @@
+/*
+ * 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.fluss.client.table.scanner.log;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.record.ArrowBatchData;
+import org.apache.fluss.utils.AbstractIterator;
+import org.apache.fluss.utils.IOUtils;
+
+import javax.annotation.Nonnull;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A container that holds the scanned Arrow batches per bucket for a
particular table.
+ *
+ * <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
+public class ArrowScanRecords implements Iterable<ArrowBatchData>,
AutoCloseable {
+ public static final ArrowScanRecords EMPTY = new
ArrowScanRecords(Collections.emptyMap());
+
+ private final Map<TableBucket, List<ArrowBatchData>> records;
+
+ public ArrowScanRecords(Map<TableBucket, List<ArrowBatchData>> records) {
+ this.records = records;
+ }
+
+ /** Get just the Arrow batches for the given bucket. */
+ public List<ArrowBatchData> records(TableBucket scanBucket) {
+ List<ArrowBatchData> recs = records.get(scanBucket);
+ if (recs == null) {
+ return Collections.emptyList();
+ }
+ return Collections.unmodifiableList(recs);
+ }
+
+ /** Returns the buckets that contain Arrow batches. */
+ public Set<TableBucket> buckets() {
+ return Collections.unmodifiableSet(records.keySet());
+ }
+
+ /** Returns the total number of rows in all batches. */
+ public int count() {
+ int count = 0;
+ for (List<ArrowBatchData> recs : records.values()) {
+ for (ArrowBatchData rec : recs) {
+ count += rec.getRecordCount();
+ }
+ }
+ return count;
+ }
+
+ public boolean isEmpty() {
+ return records.isEmpty();
+ }
+
+ /** Closes all Arrow batches held by this container, releasing off-heap
memory. */
+ @Override
+ public void close() {
+ for (List<ArrowBatchData> recs : records.values()) {
+ for (ArrowBatchData rec : recs) {
+ IOUtils.closeQuietly(rec);
+ }
+ }
+ }
+
+ @Override
+ @Nonnull
+ public Iterator<ArrowBatchData> iterator() {
+ return new ConcatenatedIterable(records.values()).iterator();
+ }
+
+ private static class ConcatenatedIterable implements
Iterable<ArrowBatchData> {
+
+ private final Iterable<? extends Iterable<ArrowBatchData>> iterables;
+
+ private ConcatenatedIterable(Iterable<? extends
Iterable<ArrowBatchData>> iterables) {
+ this.iterables = iterables;
+ }
+
+ @Override
+ @Nonnull
+ public Iterator<ArrowBatchData> iterator() {
+ return new AbstractIterator<ArrowBatchData>() {
+ final Iterator<? extends Iterable<ArrowBatchData>> iters =
iterables.iterator();
+ Iterator<ArrowBatchData> current;
+
+ public ArrowBatchData makeNext() {
+ while (current == null || !current.hasNext()) {
+ if (iters.hasNext()) {
+ current = iters.next().iterator();
+ } else {
+ return allDone();
+ }
+ }
+ return current.next();
+ }
+ };
+ }
+ }
+}
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java
index e7baeb963..f22a65213 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/CompletedFetch.java
@@ -22,6 +22,7 @@ import org.apache.fluss.client.table.scanner.ScanRecord;
import org.apache.fluss.exception.CorruptRecordException;
import org.apache.fluss.exception.FetchException;
import org.apache.fluss.metadata.TableBucket;
+import org.apache.fluss.record.ArrowBatchData;
import org.apache.fluss.record.ChangeType;
import org.apache.fluss.record.CompactedLogRecord;
import org.apache.fluss.record.IndexedLogRecord;
@@ -50,7 +51,7 @@ import static
org.apache.fluss.utils.Preconditions.checkArgument;
* to maintain state between calls to {@link #fetchRecords(int)}.
*/
@Internal
-abstract class CompletedFetch {
+public abstract class CompletedFetch {
static final Logger LOG = LoggerFactory.getLogger(CompletedFetch.class);
static final long NO_FILTERED_END_OFFSET = -1L;
@@ -262,33 +263,89 @@ abstract class CompletedFetch {
cachedRecordException = null;
}
+ /**
+ * The {@link LogRecordBatch batches} are loaded as {@link ArrowBatchData
Arrow batches} and
+ * returned.
+ *
+ * @param maxRecords A soft upper bound on the number of records to
return. Because batches are
+ * returned whole (never split), the actual number of records may
exceed this value. At
+ * least one batch is always returned if available, even if it alone
exceeds the limit.
+ * @return {@link ArrowBatchData Arrow batches}
+ */
+ List<ArrowBatchData> fetchArrowBatches(int maxRecords) {
+ if (cachedRecordException != null) {
+ throw new FetchException(
+ "Received exception when fetching the next Arrow batch
from "
+ + tableBucket
+ + ". If needed, please back past the batch to
continue scanning.",
+ cachedRecordException);
+ }
+
+ if (isConsumed) {
+ return Collections.emptyList();
+ }
+
+ List<ArrowBatchData> arrowBatches = new ArrayList<>();
+ int recordsFetched = 0;
+ try {
+ while (recordsFetched < maxRecords || arrowBatches.isEmpty()) {
+ LogRecordBatch batch = nextFetchedBatch();
+ if (batch == null) {
+ break;
+ }
+
+ ArrowBatchData arrowBatchData =
batch.loadArrowBatch(readContext);
+ if (arrowBatchData.getRecordCount() == 0) {
+ arrowBatchData.close();
+ continue;
+ }
+
+ // Skip records that are before nextFetchOffset, analogous to
the
+ // record-level filtering in nextFetchedRecord() for the
row-based path.
+ long batchBaseOffset = arrowBatchData.getBaseLogOffset();
+ if (batchBaseOffset < nextFetchOffset) {
+ int skipRows = (int) (nextFetchOffset - batchBaseOffset);
+ if (skipRows >= arrowBatchData.getRecordCount()) {
+ arrowBatchData.close();
+ continue;
+ }
+ arrowBatchData =
arrowBatchData.sliceAndTransferOwnership(skipRows);
+ }
+
+ arrowBatches.add(arrowBatchData);
+ recordsRead += arrowBatchData.getRecordCount();
+ recordsFetched += arrowBatchData.getRecordCount();
+ nextFetchOffset = batch.nextLogOffset();
+ }
+ } catch (Exception e) {
+ // Deliver partial results when possible, mirroring fetchRecords()
semantics.
+ // The batch iterator is single-pass (batches.next()), so
already-consumed
+ // batches cannot be re-read. Rolling back nextFetchOffset would
leave it
+ // pointing at batches the iterator has already passed, causing a
different
+ // kind of inconsistency. Delivering partial results keeps
offsets, recordsRead,
+ // and the iterator position all in sync.
+ cachedRecordException = e;
+ if (arrowBatches.isEmpty()) {
+ throw new FetchException(
+ "Received exception when fetching the next Arrow batch
from "
+ + tableBucket
+ + ". If needed, please back past the batch to
continue scanning.",
+ e);
+ }
+ }
+
+ return arrowBatches;
+ }
+
private LogRecord nextFetchedRecord() throws Exception {
while (true) {
if (records == null || !records.hasNext()) {
- maybeCloseRecordStream();
-
- if (!batches.hasNext()) {
- // In batch, we preserve the last offset in a batch. By
using the next offset
- // computed from the last offset in the batch, we ensure
that the offset of the
- // next fetch will point to the next batch, which avoids
unnecessary re-fetching
- // of the same batch (in the worst case, the scanner could
get stuck fetching
- // the same batch repeatedly).
- // When filteredEndOffset is set, use the max of the
batch-derived offset and
- // filteredEndOffset to skip already-scanned-and-filtered
trailing batches.
- if (currentBatch != null) {
- nextFetchOffset =
Math.max(currentBatch.nextLogOffset(), filteredEndOffset);
- } else if (filteredEndOffset != NO_FILTERED_END_OFFSET) {
- nextFetchOffset = filteredEndOffset;
- }
- drain();
+ LogRecordBatch batch = nextFetchedBatch();
+ if (batch == null) {
return null;
}
- currentBatch = batches.next();
- // TODO get last epoch.
- maybeEnsureValid(currentBatch);
-
- records = currentBatch.records(readContext);
+ records = batch.records(readContext);
} else {
LogRecord record = records.next();
// skip any records out of range.
@@ -299,6 +356,35 @@ abstract class CompletedFetch {
}
}
+ private LogRecordBatch nextFetchedBatch() {
+ maybeCloseRecordStream();
+ if (!batches.hasNext()) {
+ finishFetchedBatches();
+ return null;
+ }
+
+ currentBatch = batches.next();
+ // TODO get last epoch.
+ maybeEnsureValid(currentBatch);
+ return currentBatch;
+ }
+
+ private void finishFetchedBatches() {
+ // In batch, we preserve the last offset in a batch. By using the next
offset
+ // computed from the last offset in the batch, we ensure that the
offset of the
+ // next fetch will point to the next batch, which avoids unnecessary
re-fetching
+ // of the same batch (in the worst case, the scanner could get stuck
fetching
+ // the same batch repeatedly).
+ // When filteredEndOffset is set, use the max of the batch-derived
offset and
+ // filteredEndOffset to skip already-scanned-and-filtered trailing
batches.
+ if (currentBatch != null) {
+ nextFetchOffset = Math.max(currentBatch.nextLogOffset(),
filteredEndOffset);
+ } else if (filteredEndOffset != NO_FILTERED_END_OFFSET) {
+ nextFetchOffset = filteredEndOffset;
+ }
+ drain();
+ }
+
private void maybeEnsureValid(LogRecordBatch batch) {
if (isCheckCrcs) {
if (readContext.isProjectionPushDowned()) {
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 05c43eca0..bcf47c07f 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
@@ -20,27 +20,17 @@ package org.apache.fluss.client.table.scanner.log;
import org.apache.fluss.annotation.Internal;
import org.apache.fluss.client.metadata.MetadataUpdater;
import org.apache.fluss.client.table.scanner.ScanRecord;
-import org.apache.fluss.config.ConfigOptions;
import org.apache.fluss.config.Configuration;
-import org.apache.fluss.exception.AuthorizationException;
-import org.apache.fluss.exception.FetchException;
-import org.apache.fluss.exception.LogOffsetOutOfRangeException;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.record.LogRecord;
import org.apache.fluss.record.LogRecordBatch;
-import org.apache.fluss.rpc.protocol.ApiError;
-import org.apache.fluss.rpc.protocol.Errors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -55,242 +45,29 @@ import java.util.Map;
*/
@ThreadSafe
@Internal
-public class LogFetchCollector {
+public class LogFetchCollector extends AbstractLogFetchCollector<ScanRecord,
ScanRecords> {
private static final Logger LOG =
LoggerFactory.getLogger(LogFetchCollector.class);
- private final TablePath tablePath;
- private final LogScannerStatus logScannerStatus;
- private final int maxPollRecords;
- private final MetadataUpdater metadataUpdater;
-
public LogFetchCollector(
TablePath tablePath,
LogScannerStatus logScannerStatus,
Configuration conf,
MetadataUpdater metadataUpdater) {
- this.tablePath = tablePath;
- this.logScannerStatus = logScannerStatus;
- this.maxPollRecords =
conf.getInt(ConfigOptions.CLIENT_SCANNER_LOG_MAX_POLL_RECORDS);
- this.metadataUpdater = metadataUpdater;
+ super(LOG, tablePath, logScannerStatus, conf, metadataUpdater);
}
- /**
- * 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.
- *
- * @return The fetched records per partition
- * @throws LogOffsetOutOfRangeException If there is OffsetOutOfRange error
in fetchResponse and
- * the defaultResetPolicy is NONE
- */
- public ScanRecords collectFetch(final LogFetchBuffer logFetchBuffer) {
- Map<TableBucket, List<ScanRecord>> fetched = new HashMap<>();
- int recordsRemaining = maxPollRecords;
-
- try {
- while (recordsRemaining > 0) {
- CompletedFetch nextInLineFetch =
logFetchBuffer.nextInLineFetch();
- if (nextInLineFetch == null || nextInLineFetch.isConsumed()) {
- CompletedFetch completedFetch = logFetchBuffer.peek();
- if (completedFetch == null) {
- break;
- }
-
- if (!completedFetch.isInitialized()) {
- try {
- CompletedFetch initialized =
initialize(completedFetch);
- logFetchBuffer.setNextInLineFetch(initialized);
- if (initialized == null) {
- completedFetch.drain();
- }
- } catch (Exception e) {
- // Remove a completedFetch upon a parse with
exception if
- // (1) it contains no records, and
- // (2) there are no fetched records with actual
content preceding this
- // exception.
- if (fetched.isEmpty() &&
completedFetch.sizeInBytes == 0) {
- logFetchBuffer.poll();
- }
- throw e;
- }
- } else {
- logFetchBuffer.setNextInLineFetch(completedFetch);
- }
-
- logFetchBuffer.poll();
- } else {
- List<ScanRecord> records = fetchRecords(nextInLineFetch,
recordsRemaining);
- if (!records.isEmpty()) {
- TableBucket tableBucket = nextInLineFetch.tableBucket;
- List<ScanRecord> currentRecords =
fetched.get(tableBucket);
- if (currentRecords == null) {
- fetched.put(tableBucket, records);
- } else {
- // this case shouldn't usually happen because we
only send one fetch at
- // a time per bucket, but it might conceivably
happen in some rare
- // cases (such as bucket leader changes). we have
to copy to a new list
- // because the old one may be immutable
- List<ScanRecord> newScanRecords =
- new ArrayList<>(records.size() +
currentRecords.size());
- newScanRecords.addAll(currentRecords);
- newScanRecords.addAll(records);
- fetched.put(tableBucket, newScanRecords);
- }
-
- recordsRemaining -= records.size();
- }
- }
- }
- } catch (FetchException e) {
- if (fetched.isEmpty()) {
- throw e;
- }
- }
-
- return new ScanRecords(fetched);
+ @Override
+ protected List<ScanRecord> doFetchRecords(CompletedFetch nextInLineFetch,
int maxRecords) {
+ return nextInLineFetch.fetchRecords(maxRecords);
}
- private List<ScanRecord> fetchRecords(CompletedFetch nextInLineFetch, int
maxRecords) {
- TableBucket tb = nextInLineFetch.tableBucket;
- Long offset = logScannerStatus.getBucketOffset(tb);
- if (offset == null) {
- LOG.debug(
- "Ignoring fetched records for {} at offset {} since the
current offset is null which means the "
- + "bucket has been unsubscribe.",
- tb,
- nextInLineFetch.fetchOffset());
- } else {
- if (nextInLineFetch.nextFetchOffset() == offset) {
- List<ScanRecord> records =
nextInLineFetch.fetchRecords(maxRecords);
- LOG.trace(
- "Returning {} fetched records at offset {} for
assigned bucket {}.",
- records.size(),
- offset,
- tb);
-
- if (nextInLineFetch.nextFetchOffset() > offset) {
- LOG.trace(
- "Updating fetch offset from {} to {} for bucket {}
and returning {} records from poll()",
- offset,
- nextInLineFetch.nextFetchOffset(),
- tb,
- records.size());
- logScannerStatus.updateOffset(tb,
nextInLineFetch.nextFetchOffset());
- }
- return records;
- } else {
- // these records aren't next in line based on the last
consumed offset, ignore them
- // they must be from an obsolete request
- LOG.warn(
- "Ignoring fetched records for {} at offset {} since
the current offset is {}",
- nextInLineFetch.tableBucket,
- nextInLineFetch.nextFetchOffset(),
- offset);
- }
- }
-
- LOG.trace("Draining fetched records for bucket {}",
nextInLineFetch.tableBucket);
- nextInLineFetch.drain();
-
- return Collections.emptyList();
- }
-
- /** Initialize a {@link CompletedFetch} object. */
- private @Nullable CompletedFetch initialize(CompletedFetch completedFetch)
{
- TableBucket tb = completedFetch.tableBucket;
- ApiError error = completedFetch.error;
-
- try {
- if (error.isSuccess()) {
- return handleInitializeSuccess(completedFetch);
- } else {
- handleInitializeErrors(completedFetch, error.error(),
error.messageWithFallback());
- return null;
- }
- } finally {
- if (error.isFailure()) {
- // we move the bucket to the end if there was an error. This
way,
- // it's more likely that buckets for the same table can remain
together
- // (allowing for more efficient serialization).
- logScannerStatus.moveBucketToEnd(tb);
- }
- }
- }
-
- private @Nullable CompletedFetch handleInitializeSuccess(CompletedFetch
completedFetch) {
- TableBucket tb = completedFetch.tableBucket;
- long fetchOffset = completedFetch.fetchOffset();
-
- // we are interested in this fetch only if the beginning offset
matches the
- // current consumed position.
- Long offset = logScannerStatus.getBucketOffset(tb);
- if (offset == null) {
- LOG.debug(
- "Discarding stale fetch response for bucket {} since the
expected offset is null which means the bucket has been "
- + "unsubscribed.",
- tb);
- return null;
- }
- if (offset != fetchOffset) {
- LOG.warn(
- "Discarding stale fetch response for bucket {} since its
offset {} does not match the expected offset {}.",
- tb,
- fetchOffset,
- offset);
- return null;
- }
-
- long highWatermark = completedFetch.highWatermark;
- if (highWatermark >= 0) {
- LOG.trace("Updating high watermark for bucket {} to {}.", tb,
highWatermark);
- logScannerStatus.updateHighWatermark(tb, highWatermark);
- }
-
- completedFetch.setInitialized();
- return completedFetch;
+ @Override
+ protected int recordCount(List<ScanRecord> fetchedRecords) {
+ return fetchedRecords.size();
}
- private void handleInitializeErrors(
- CompletedFetch completedFetch, Errors error, String errorMessage) {
- TableBucket tb = completedFetch.tableBucket;
- long fetchOffset = completedFetch.fetchOffset();
- if (error == Errors.NOT_LEADER_OR_FOLLOWER
- || error == Errors.LOG_STORAGE_EXCEPTION
- || error == Errors.KV_STORAGE_EXCEPTION
- || error == Errors.STORAGE_EXCEPTION
- || error == Errors.FENCED_LEADER_EPOCH_EXCEPTION) {
- LOG.debug(
- "Error in fetch for bucket {}: {}:{}",
- tb,
- error.exceptionName(),
- error.exception(errorMessage));
- metadataUpdater.checkAndUpdateMetadata(tablePath, tb);
- } else if (error == Errors.UNKNOWN_TABLE_OR_BUCKET_EXCEPTION) {
- LOG.warn("Received unknown table or bucket error in fetch for
bucket {}", tb);
- metadataUpdater.checkAndUpdateMetadata(tablePath, tb);
- } else if (error == Errors.LOG_OFFSET_OUT_OF_RANGE_EXCEPTION) {
- throw new FetchException(
- String.format(
- "The fetching offset %s is out of range: %s",
- fetchOffset, error.exception(errorMessage)));
- } else if (error == Errors.AUTHORIZATION_EXCEPTION) {
- throw new AuthorizationException(errorMessage);
- } else if (error == Errors.UNKNOWN_SERVER_ERROR) {
- LOG.warn(
- "Unknown server error while fetching offset {} for bucket
{}: {}",
- fetchOffset,
- tb,
- error.exception(errorMessage));
- } else if (error == Errors.CORRUPT_MESSAGE) {
- throw new FetchException(
- String.format(
- "Encountered corrupt message when fetching offset
%s for bucket %s: %s",
- fetchOffset, tb, error.exception(errorMessage)));
- } else {
- throw new FetchException(
- String.format(
- "Unexpected error code %s while fetching at offset
%s from bucket %s: %s",
- error, fetchOffset, tb,
error.exception(errorMessage)));
- }
+ @Override
+ protected ScanRecords toResult(Map<TableBucket, List<ScanRecord>>
fetchedRecords) {
+ return new ScanRecords(fetchedRecords);
}
}
diff --git
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
index 2625716cd..2a1fc4361 100644
---
a/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
+++
b/fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetcher.java
@@ -107,6 +107,7 @@ public class LogFetcher implements Closeable {
private final LogScannerStatus logScannerStatus;
private final LogFetchBuffer logFetchBuffer;
private final LogFetchCollector logFetchCollector;
+ private final ArrowLogFetchCollector arrowLogFetchCollector;
private final RemoteLogDownloader remoteLogDownloader;
@GuardedBy("this")
@@ -162,6 +163,8 @@ public class LogFetcher implements Closeable {
this.metadataUpdater = metadataUpdater;
this.logFetchCollector =
new LogFetchCollector(tablePath, logScannerStatus, conf,
metadataUpdater);
+ this.arrowLogFetchCollector =
+ new ArrowLogFetchCollector(tablePath, logScannerStatus, conf,
metadataUpdater);
this.scannerMetricGroup = scannerMetricGroup;
this.remoteLogDownloader =
new RemoteLogDownloader(tablePath, conf, remoteFileDownloader,
scannerMetricGroup);
@@ -181,6 +184,10 @@ public class LogFetcher implements Closeable {
return logFetchCollector.collectFetch(logFetchBuffer);
}
+ public ArrowScanRecords collectArrowFetch() {
+ return arrowLogFetchCollector.collectFetch(logFetchBuffer);
+ }
+
/**
* Set up a fetch request for any node that we have assigned buckets for
which doesn't already
* have an in-flight fetch or pending fetch data.
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 9a2dbf0b4..e7178f066 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
@@ -17,12 +17,15 @@
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;
import org.apache.fluss.metadata.SchemaGetter;
import org.apache.fluss.metadata.TableBucket;
import org.apache.fluss.metadata.TableInfo;
@@ -43,6 +46,7 @@ import java.util.Collections;
import java.util.ConcurrentModificationException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Supplier;
/**
* The default impl of {@link LogScanner}.
@@ -56,24 +60,27 @@ import java.util.concurrent.atomic.AtomicLong;
@PublicEvolving
public class LogScannerImpl implements LogScanner {
private static final Logger LOG =
LoggerFactory.getLogger(LogScannerImpl.class);
-
private static final long NO_CURRENT_THREAD = -1L;
+
private final TablePath tablePath;
private final LogScannerStatus logScannerStatus;
private final MetadataUpdater metadataUpdater;
private final LogFetcher logFetcher;
private final long tableId;
private final boolean isPartitionedTable;
-
- private volatile boolean closed = false;
+ private final boolean isArrowLogFormat;
+ private final boolean isLogTable;
+ private final boolean hasProjection;
+ // metrics
+ private final ScannerMetricGroup scannerMetricGroup;
// currentThread holds the threadId of the current thread accessing
FlussLogScanner
// and is used to prevent multithreaded access
private final AtomicLong currentThread = new AtomicLong(NO_CURRENT_THREAD);
// refCount is used to allow reentrant access by the thread who has
acquired currentThread.
private final AtomicInteger refCount = new AtomicInteger(0);
- // metrics
- private final ScannerMetricGroup scannerMetricGroup;
+
+ private volatile boolean closed = false;
public LogScannerImpl(
Configuration conf,
@@ -87,6 +94,9 @@ public class LogScannerImpl implements LogScanner {
this.tablePath = tableInfo.getTablePath();
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.logScannerStatus = new LogScannerStatus();
@@ -131,43 +141,34 @@ public class LogScannerImpl implements LogScanner {
@Override
public ScanRecords poll(Duration timeout) {
- acquireAndEnsureOpen();
- try {
- if (!logScannerStatus.prepareToPoll()) {
- throw new IllegalStateException("LogScanner is not subscribed
any buckets.");
- }
-
- scannerMetricGroup.recordPollStart(System.currentTimeMillis());
- long timeoutNanos = timeout.toNanos();
- long startNanos = System.nanoTime();
- do {
- ScanRecords scanRecords = pollForFetches();
- if (scanRecords.isEmpty()) {
- try {
- if (!logFetcher.awaitNotEmpty(startNanos +
timeoutNanos)) {
- // logFetcher waits for the timeout and no data in
buffer,
- // so we return empty
- return scanRecords;
- }
- } catch (WakeupException e) {
- // wakeup() is called, we need to return empty
- return scanRecords;
- }
- } else {
- // before returning the fetched records, we can send off
the next round of
- // fetches and avoid block waiting for their responses to
enable pipelining
- // while the user is handling the fetched records.
- logFetcher.sendFetches();
-
- return scanRecords;
- }
- } while (System.nanoTime() - startNanos < timeoutNanos);
+ return doPoll(timeout, this::pollForFetches, ScanRecords::isEmpty, ()
-> ScanRecords.EMPTY);
+ }
- return ScanRecords.EMPTY;
- } finally {
- release();
- scannerMetricGroup.recordPollEnd(System.currentTimeMillis());
+ /**
+ * Polls Arrow record batches for internal callers.
+ *
+ * <p>This method is intentionally kept off the public {@link Scan} API
surface for now.
+ */
+ @Internal
+ 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.");
+ }
+
+ return doPoll(
+ timeout,
+ this::pollForRecordBatches,
+ ArrowScanRecords::isEmpty,
+ () -> ArrowScanRecords.EMPTY);
}
@Override
@@ -181,8 +182,8 @@ public class LogScannerImpl implements LogScanner {
acquireAndEnsureOpen();
try {
TableBucket tableBucket = new TableBucket(tableId, bucket);
-
this.metadataUpdater.checkAndUpdateTableMetadata(Collections.singleton(tablePath));
-
this.logScannerStatus.assignScanBuckets(Collections.singletonMap(tableBucket,
offset));
+
metadataUpdater.checkAndUpdateTableMetadata(Collections.singleton(tablePath));
+
logScannerStatus.assignScanBuckets(Collections.singletonMap(tableBucket,
offset));
} finally {
release();
}
@@ -202,9 +203,9 @@ public class LogScannerImpl implements LogScanner {
// we make assumption that the partition id must belong to the
current table
// if we can't find the partition id from the table path, we'll
consider the table
// is not exist
- this.metadataUpdater.checkAndUpdatePartitionMetadata(
+ metadataUpdater.checkAndUpdatePartitionMetadata(
tablePath, Collections.singleton(partitionId));
-
this.logScannerStatus.assignScanBuckets(Collections.singletonMap(tableBucket,
offset));
+
logScannerStatus.assignScanBuckets(Collections.singletonMap(tableBucket,
offset));
} finally {
release();
}
@@ -219,7 +220,7 @@ public class LogScannerImpl implements LogScanner {
acquireAndEnsureOpen();
try {
TableBucket tableBucket = new TableBucket(tableId, partitionId,
bucket);
-
this.logScannerStatus.unassignScanBuckets(Collections.singletonList(tableBucket));
+
logScannerStatus.unassignScanBuckets(Collections.singletonList(tableBucket));
} finally {
release();
}
@@ -236,7 +237,7 @@ public class LogScannerImpl implements LogScanner {
acquireAndEnsureOpen();
try {
TableBucket tableBucket = new TableBucket(tableId, bucket);
-
this.logScannerStatus.unassignScanBuckets(Collections.singletonList(tableBucket));
+
logScannerStatus.unassignScanBuckets(Collections.singletonList(tableBucket));
} finally {
release();
}
@@ -255,12 +256,66 @@ public class LogScannerImpl implements LogScanner {
// send any new fetches (won't resend pending fetches).
logFetcher.sendFetches();
-
return logFetcher.collectFetch();
}
+ private ArrowScanRecords pollForRecordBatches() {
+ ArrowScanRecords scanRecords = logFetcher.collectArrowFetch();
+ if (!scanRecords.isEmpty()) {
+ return scanRecords;
+ }
+
+ // send any new fetches (won't resend pending fetches).
+ logFetcher.sendFetches();
+ return logFetcher.collectArrowFetch();
+ }
+
+ /** Shared polling loop for row and Arrow scan results. */
+ private <T> T doPoll(
+ Duration timeout,
+ Supplier<T> pollForFetches,
+ java.util.function.Predicate<T> isEmpty,
+ Supplier<T> emptyResult) {
+ acquireAndEnsureOpen();
+ try {
+ if (!logScannerStatus.prepareToPoll()) {
+ throw new IllegalStateException("LogScanner is not subscribed
any buckets.");
+ }
+
+ scannerMetricGroup.recordPollStart(System.currentTimeMillis());
+ long timeoutNanos = timeout.toNanos();
+ long startNanos = System.nanoTime();
+ do {
+ T scanRecords = pollForFetches.get();
+ if (isEmpty.test(scanRecords)) {
+ try {
+ if (!logFetcher.awaitNotEmpty(startNanos +
timeoutNanos)) {
+ // logFetcher waits for the timeout and no data in
buffer,
+ // so we return empty
+ return scanRecords;
+ }
+ } catch (WakeupException e) {
+ // wakeup() is called, we need to return empty
+ return scanRecords;
+ }
+ } else {
+ // before returning the fetched records, we can send off
the next round of
+ // fetches and avoid block waiting for their responses to
enable pipelining
+ // while the user is handling the fetched records.
+ logFetcher.sendFetches();
+ return scanRecords;
+ }
+ } while (System.nanoTime() - startNanos < timeoutNanos);
+
+ return emptyResult.get();
+ } finally {
+ release();
+ scannerMetricGroup.recordPollEnd(System.currentTimeMillis());
+ }
+ }
+
/**
- * Acquire the light lock and ensure that the consumer hasn't been closed.
+ * Acquire the light lock and ensure that the scanner hasn't been closed.
*
* @throws IllegalStateException If the scanner has been closed
*/
@@ -280,8 +335,8 @@ public class LogScannerImpl implements LogScanner {
* @throws ConcurrentModificationException if another thread already has
the lock
*/
private void acquire() {
- final Thread thread = Thread.currentThread();
- final long threadId = thread.getId();
+ Thread thread = Thread.currentThread();
+ long threadId = thread.getId();
if (threadId != currentThread.get()
&& !currentThread.compareAndSet(NO_CURRENT_THREAD, threadId)) {
throw new ConcurrentModificationException(
@@ -298,7 +353,7 @@ public class LogScannerImpl implements LogScanner {
refCount.incrementAndGet();
}
- /** Release the light lock protecting the consumer from multithreaded
access. */
+ /** Release the light lock protecting the scanner from multithreaded
access. */
private void release() {
if (refCount.decrementAndGet() == 0) {
currentThread.set(NO_CURRENT_THREAD);
diff --git
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogScannerITCase.java
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogScannerITCase.java
index cfe215fb0..e3f60e2df 100644
---
a/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogScannerITCase.java
+++
b/fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogScannerITCase.java
@@ -23,21 +23,27 @@ import org.apache.fluss.client.table.scanner.ScanRecord;
import org.apache.fluss.client.table.writer.AppendWriter;
import org.apache.fluss.client.table.writer.UpsertWriter;
import org.apache.fluss.exception.FetchException;
+import org.apache.fluss.metadata.LogFormat;
import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TableChange;
import org.apache.fluss.metadata.TableDescriptor;
import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ArrowBatchData;
import org.apache.fluss.record.ChangeType;
import org.apache.fluss.row.GenericRow;
import org.apache.fluss.row.InternalRow;
import org.apache.fluss.types.DataTypes;
import org.apache.fluss.types.RowType;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.VarCharVector;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
@@ -467,4 +473,99 @@ public class LogScannerITCase extends
ClientToServerITCaseBase {
}
}
}
+
+ @Test
+ void testPollArrowBatchesWithSchemaEvolution() throws Exception {
+ TablePath tablePath = TablePath.of("test_db_1",
"test_arrow_batches_with_schema_evolution");
+ TableDescriptor tableDescriptor =
+ TableDescriptor.builder()
+ .schema(DATA1_SCHEMA)
+ .distributedBy(1)
+ .logFormat(LogFormat.ARROW)
+ .build();
+ createTable(tablePath, tableDescriptor, false);
+
+ // write 3 rows with the original schema (a: INT, b: STRING)
+ try (Table table = conn.getTable(tablePath)) {
+ AppendWriter appendWriter = table.newAppend().createWriter();
+ for (int i = 0; i < 3; i++) {
+ appendWriter.append(row(i, "value-" + i));
+ }
+ appendWriter.flush();
+ }
+
+ // add column c: STRING
+ admin.alterTable(
+ tablePath,
+ Collections.singletonList(
+ TableChange.addColumn(
+ "c",
+ DataTypes.STRING(),
+ null,
+ TableChange.ColumnPosition.last())),
+ false)
+ .get();
+
+ // write 3 more rows with the evolved schema (a: INT, b: STRING, c:
STRING)
+ try (Table table = conn.getTable(tablePath)) {
+ AppendWriter appendWriter = table.newAppend().createWriter();
+ for (int i = 3; i < 6; i++) {
+ appendWriter.append(row(i, "value-" + i, "extra-" + i));
+ }
+ appendWriter.flush();
+
+ int totalRecords = 6;
+ // subscribe from beginning and verify all 6 records
+ try (LogScannerImpl scanner = (LogScannerImpl)
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()) {
+ scanner2.subscribe(0, subscribeOffset);
+ pollAndVerifyArrowBatches(
+ scanner2, totalRecords - subscribeOffset,
subscribeOffset);
+ }
+ }
+ }
+
+ private void pollAndVerifyArrowBatches(
+ LogScannerImpl scanner, int expectedRecords, int
minExpectedOffset) {
+ int count = 0;
+ long deadline = System.nanoTime() + Duration.ofSeconds(30).toNanos();
+ while (count < expectedRecords) {
+ assertThat(System.nanoTime())
+ .as("Timed out waiting for %s records, got %s",
expectedRecords, count)
+ .isLessThan(deadline);
+ try (ArrowScanRecords records =
scanner.pollRecordBatch(Duration.ofSeconds(1))) {
+ for (ArrowBatchData batch : records) {
+ try (ArrowBatchData b = batch) {
+ IntVector intVector = (IntVector)
b.getVectorSchemaRoot().getVector(0);
+ VarCharVector stringVector =
+ (VarCharVector)
b.getVectorSchemaRoot().getVector(1);
+ VarCharVector extraVector =
+ (VarCharVector)
b.getVectorSchemaRoot().getVector(2);
+ for (int rowId = 0; rowId < b.getRecordCount();
rowId++) {
+ int expectedValue = (int) (b.getBaseLogOffset() +
rowId);
+
assertThat(expectedValue).isGreaterThanOrEqualTo(minExpectedOffset);
+
assertThat(intVector.get(rowId)).isEqualTo(expectedValue);
+
assertThat(stringVector.getObject(rowId).toString())
+ .isEqualTo("value-" + expectedValue);
+ if (expectedValue < 3) {
+ assertThat(extraVector.isNull(rowId)).isTrue();
+ } else {
+
assertThat(extraVector.getObject(rowId).toString())
+ .isEqualTo("extra-" + expectedValue);
+ }
+ count++;
+ }
+ }
+ }
+ }
+ }
+ assertThat(count).isEqualTo(expectedRecords);
+ }
}
diff --git a/fluss-common/pom.xml b/fluss-common/pom.xml
index 206f203a6..dded3a331 100644
--- a/fluss-common/pom.xml
+++ b/fluss-common/pom.xml
@@ -62,6 +62,24 @@
<artifactId>fluss-shaded-arrow</artifactId>
</dependency>
+ <!-- Unshaded Arrow dependencies used by UnshadedArrowReadUtils to
load Arrow batches
+ with original Arrow classes (for ArrowBatchData / log scanner
arrow batch scan).
+ Marked as provided because the actual Arrow classes are supplied
at runtime by
+ downstream modules (e.g. fluss-client); fluss-common only needs
them for compilation. -->
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-vector</artifactId>
+ <version>${arrow.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-memory-netty</artifactId>
+ <version>${arrow.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
<!-- TODO: these two dependencies need to be shaded. -->
<dependency>
<groupId>at.yawk.lz4</groupId>
diff --git
a/fluss-common/src/main/java/org/apache/fluss/compression/UnshadedArrowCompressionFactory.java
b/fluss-common/src/main/java/org/apache/fluss/compression/UnshadedArrowCompressionFactory.java
new file mode 100644
index 000000000..0e7da37bb
--- /dev/null
+++
b/fluss-common/src/main/java/org/apache/fluss/compression/UnshadedArrowCompressionFactory.java
@@ -0,0 +1,60 @@
+/*
+ * 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.fluss.compression;
+
+import org.apache.fluss.annotation.Internal;
+
+import org.apache.arrow.vector.compression.CompressionCodec;
+import org.apache.arrow.vector.compression.CompressionUtil;
+import org.apache.arrow.vector.compression.NoCompressionCodec;
+
+/** Unshaded Arrow compression factory for scanner/read path. */
+@Internal
+public class UnshadedArrowCompressionFactory implements
CompressionCodec.Factory {
+
+ public static final UnshadedArrowCompressionFactory INSTANCE =
+ new UnshadedArrowCompressionFactory();
+
+ @Override
+ public CompressionCodec createCodec(CompressionUtil.CodecType codecType) {
+ switch (codecType) {
+ case LZ4_FRAME:
+ return new UnshadedLz4ArrowCompressionCodec();
+ case ZSTD:
+ return new UnshadedZstdArrowCompressionCodec();
+ case NO_COMPRESSION:
+ return NoCompressionCodec.INSTANCE;
+ default:
+ throw new IllegalArgumentException("Compression type not
supported: " + codecType);
+ }
+ }
+
+ @Override
+ public CompressionCodec createCodec(CompressionUtil.CodecType codecType,
int compressionLevel) {
+ switch (codecType) {
+ case LZ4_FRAME:
+ return new UnshadedLz4ArrowCompressionCodec();
+ case ZSTD:
+ return new UnshadedZstdArrowCompressionCodec(compressionLevel);
+ case NO_COMPRESSION:
+ return NoCompressionCodec.INSTANCE;
+ default:
+ throw new IllegalArgumentException("Compression type not
supported: " + codecType);
+ }
+ }
+}
diff --git
a/fluss-common/src/main/java/org/apache/fluss/compression/UnshadedLz4ArrowCompressionCodec.java
b/fluss-common/src/main/java/org/apache/fluss/compression/UnshadedLz4ArrowCompressionCodec.java
new file mode 100644
index 000000000..bc41aa802
--- /dev/null
+++
b/fluss-common/src/main/java/org/apache/fluss/compression/UnshadedLz4ArrowCompressionCodec.java
@@ -0,0 +1,93 @@
+/*
+ * 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.fluss.compression;
+
+import org.apache.fluss.utils.IOUtils;
+
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.compression.AbstractCompressionCodec;
+import org.apache.arrow.vector.compression.CompressionUtil;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+
+/** Unshaded Arrow compression codec for the LZ4 algorithm. */
+public class UnshadedLz4ArrowCompressionCodec extends AbstractCompressionCodec
{
+ @Override
+ protected ArrowBuf doCompress(BufferAllocator allocator, ArrowBuf
uncompressedBuffer) {
+ checkArgument(
+ uncompressedBuffer.writerIndex() <= Integer.MAX_VALUE,
+ "The uncompressed buffer size exceeds the integer limit");
+
+ byte[] inBytes = new byte[(int) uncompressedBuffer.writerIndex()];
+ uncompressedBuffer.getBytes(0, inBytes);
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ try (InputStream in = new ByteArrayInputStream(inBytes);
+ OutputStream out = new FlussLZ4BlockOutputStream(baos)) {
+ IOUtils.copyBytes(in, out);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ byte[] outBytes = baos.toByteArray();
+ ArrowBuf compressedBuffer =
+ allocator.buffer(CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH +
outBytes.length);
+ compressedBuffer.setBytes(CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH,
outBytes);
+
compressedBuffer.writerIndex(CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH +
outBytes.length);
+ return compressedBuffer;
+ }
+
+ @Override
+ protected ArrowBuf doDecompress(BufferAllocator allocator, ArrowBuf
compressedBuffer) {
+ checkArgument(
+ compressedBuffer.writerIndex() <= Integer.MAX_VALUE,
+ "The compressed buffer size exceeds the integer limit");
+
+ long decompressedLength = readUncompressedLength(compressedBuffer);
+ ByteBuffer inByteBuffer =
+ compressedBuffer.nioBuffer(
+ CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH,
+ (int)
+ (compressedBuffer.writerIndex()
+ -
CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH));
+ ByteArrayOutputStream out = new ByteArrayOutputStream((int)
decompressedLength);
+ try (InputStream in = new FlussLZ4BlockInputStream(inByteBuffer)) {
+ IOUtils.copyBytes(in, out);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+
+ byte[] outBytes = out.toByteArray();
+ ArrowBuf decompressedBuffer = allocator.buffer(outBytes.length);
+ decompressedBuffer.setBytes(0, outBytes);
+ decompressedBuffer.writerIndex(decompressedLength);
+ return decompressedBuffer;
+ }
+
+ @Override
+ public CompressionUtil.CodecType getCodecType() {
+ return CompressionUtil.CodecType.LZ4_FRAME;
+ }
+}
diff --git
a/fluss-common/src/main/java/org/apache/fluss/compression/UnshadedZstdArrowCompressionCodec.java
b/fluss-common/src/main/java/org/apache/fluss/compression/UnshadedZstdArrowCompressionCodec.java
new file mode 100644
index 000000000..803889521
--- /dev/null
+++
b/fluss-common/src/main/java/org/apache/fluss/compression/UnshadedZstdArrowCompressionCodec.java
@@ -0,0 +1,112 @@
+/*
+ * 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.fluss.compression;
+
+import com.github.luben.zstd.Zstd;
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.compression.AbstractCompressionCodec;
+import org.apache.arrow.vector.compression.CompressionUtil;
+
+import java.nio.ByteBuffer;
+
+/** Unshaded Arrow compression codec for the Zstd algorithm. */
+public class UnshadedZstdArrowCompressionCodec extends
AbstractCompressionCodec {
+ private static final int DEFAULT_COMPRESSION_LEVEL = 3;
+ private final int compressionLevel;
+
+ public UnshadedZstdArrowCompressionCodec() {
+ this(DEFAULT_COMPRESSION_LEVEL);
+ }
+
+ public UnshadedZstdArrowCompressionCodec(int compressionLevel) {
+ this.compressionLevel = compressionLevel;
+ }
+
+ @Override
+ protected ArrowBuf doCompress(BufferAllocator allocator, ArrowBuf
uncompressedBuffer) {
+ long maxSize = Zstd.compressBound(uncompressedBuffer.writerIndex());
+ ByteBuffer uncompressedDirectBuffer = uncompressedBuffer.nioBuffer();
+
+ long compressedSize = CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH +
maxSize;
+ ArrowBuf compressedBuffer = allocator.buffer(compressedSize);
+ ByteBuffer compressedDirectBuffer =
+ compressedBuffer.nioBuffer(
+ CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH, (int)
maxSize);
+
+ long bytesWritten =
+ Zstd.compressDirectByteBuffer(
+ compressedDirectBuffer,
+ 0,
+ (int) maxSize,
+ uncompressedDirectBuffer,
+ 0,
+ (int) uncompressedBuffer.writerIndex(),
+ compressionLevel);
+
+ if (Zstd.isError(bytesWritten)) {
+ compressedBuffer.close();
+ throw new RuntimeException("Error compressing: " +
Zstd.getErrorName(bytesWritten));
+ }
+
+
compressedBuffer.writerIndex(CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH +
bytesWritten);
+ return compressedBuffer;
+ }
+
+ @Override
+ protected ArrowBuf doDecompress(BufferAllocator allocator, ArrowBuf
compressedBuffer) {
+ long decompressedLength = readUncompressedLength(compressedBuffer);
+
+ ByteBuffer compressedDirectBuffer = compressedBuffer.nioBuffer();
+ ArrowBuf uncompressedBuffer = allocator.buffer(decompressedLength);
+ ByteBuffer uncompressedDirectBuffer =
+ uncompressedBuffer.nioBuffer(0, (int) decompressedLength);
+
+ long decompressedSize =
+ Zstd.decompressDirectByteBuffer(
+ uncompressedDirectBuffer,
+ 0,
+ (int) decompressedLength,
+ compressedDirectBuffer,
+ (int) CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH,
+ (int)
+ (compressedBuffer.writerIndex()
+ -
CompressionUtil.SIZE_OF_UNCOMPRESSED_LENGTH));
+ if (Zstd.isError(decompressedSize)) {
+ uncompressedBuffer.close();
+ throw new RuntimeException(
+ "Error decompressing: " +
Zstd.getErrorName(decompressedSize));
+ }
+
+ if (decompressedLength != decompressedSize) {
+ uncompressedBuffer.close();
+ throw new RuntimeException(
+ "Expected != actual decompressed length: "
+ + decompressedLength
+ + " != "
+ + decompressedSize);
+ }
+ uncompressedBuffer.writerIndex(decompressedLength);
+ return uncompressedBuffer;
+ }
+
+ @Override
+ public CompressionUtil.CodecType getCodecType() {
+ return CompressionUtil.CodecType.ZSTD;
+ }
+}
diff --git
a/fluss-common/src/main/java/org/apache/fluss/record/ArrowBatchData.java
b/fluss-common/src/main/java/org/apache/fluss/record/ArrowBatchData.java
new file mode 100644
index 000000000..515f73d5a
--- /dev/null
+++ b/fluss-common/src/main/java/org/apache/fluss/record/ArrowBatchData.java
@@ -0,0 +1,103 @@
+/*
+ * 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.fluss.record;
+
+import org.apache.fluss.annotation.Internal;
+
+import org.apache.arrow.vector.VectorSchemaRoot;
+
+import static org.apache.fluss.utils.Preconditions.checkArgument;
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Holds a scanned Arrow batch together with the log metadata of the batch.
+ *
+ * <p>This class only supports append-only log tables. CDC tables are not
supported.
+ *
+ * <p>The caller must close this object after use in order to release the
underlying Arrow memory.
+ */
+@Internal
+public class ArrowBatchData implements AutoCloseable {
+
+ private final VectorSchemaRoot vectorSchemaRoot;
+ private final long baseLogOffset;
+ private final long timestamp;
+ private final int schemaId;
+
+ public ArrowBatchData(
+ VectorSchemaRoot vectorSchemaRoot, long baseLogOffset, long
timestamp, int schemaId) {
+ this.vectorSchemaRoot = checkNotNull(vectorSchemaRoot,
"vectorSchemaRoot must not be null");
+ this.baseLogOffset = baseLogOffset;
+ this.timestamp = timestamp;
+ this.schemaId = schemaId;
+ }
+
+ /** Returns the Arrow vectors of this batch. */
+ public VectorSchemaRoot getVectorSchemaRoot() {
+ return vectorSchemaRoot;
+ }
+
+ /** Returns the schema id of this batch. */
+ public int getSchemaId() {
+ return schemaId;
+ }
+
+ /** Returns the base log offset of this batch. */
+ public long getBaseLogOffset() {
+ return baseLogOffset;
+ }
+
+ /** Returns the commit timestamp of this batch. */
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+ /** Returns the number of rows in this batch. */
+ public int getRecordCount() {
+ return vectorSchemaRoot.getRowCount();
+ }
+
+ /**
+ * Creates a new {@link ArrowBatchData} containing a contiguous slice of
this batch's rows and
+ * releases the original vector data.
+ *
+ * <p>After this method returns, the original {@link ArrowBatchData}
instance MUST NOT be used
+ * or closed. The caller is responsible for closing the returned instance.
+ *
+ * @param skipRows the number of leading rows to skip
+ * @return a new {@link ArrowBatchData} containing the remaining rows
after skipping
+ */
+ public ArrowBatchData sliceAndTransferOwnership(int skipRows) {
+ checkArgument(skipRows >= 0, "skipRows must be >= 0, but is %s",
skipRows);
+ checkArgument(
+ skipRows < getRecordCount(),
+ "skipRows(%s) must be < recordCount(%s)",
+ skipRows,
+ getRecordCount());
+ int remainingRows = getRecordCount() - skipRows;
+ VectorSchemaRoot slicedRoot = vectorSchemaRoot.slice(skipRows,
remainingRows);
+ // release original vector buffers; sliced vectors hold independent
copies
+ vectorSchemaRoot.close();
+ return new ArrowBatchData(slicedRoot, baseLogOffset + skipRows,
timestamp, schemaId);
+ }
+
+ @Override
+ public void close() {
+ vectorSchemaRoot.close();
+ }
+}
diff --git
a/fluss-common/src/main/java/org/apache/fluss/record/ArrowRecordBatchContext.java
b/fluss-common/src/main/java/org/apache/fluss/record/ArrowRecordBatchContext.java
new file mode 100644
index 000000000..64cf19ba4
--- /dev/null
+++
b/fluss-common/src/main/java/org/apache/fluss/record/ArrowRecordBatchContext.java
@@ -0,0 +1,43 @@
+/*
+ * 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.fluss.record;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.memory.MemorySegment;
+
+/** Internal context for loading Arrow record batches with unshaded Arrow
resources. */
+@Internal
+interface ArrowRecordBatchContext extends LogRecordBatch.ReadContext {
+
+ /** Creates a batch-scoped access wrapper for unshaded Arrow resources. */
+ UnshadedArrowBatchAccess createUnshadedArrowBatchAccess(int schemaId);
+
+ /** Internal wrapper that hides unshaded Arrow types from shared
signatures. */
+ @Internal
+ interface UnshadedArrowBatchAccess extends AutoCloseable {
+
+ /** Loads one Arrow batch from the given memory segment into the
internal read root. */
+ void loadArrowBatch(MemorySegment segment, int arrowOffset, int
arrowLength);
+
+ /**
+ * Applies schema-evolution projection (if needed) internally and
builds the final {@link
+ * ArrowBatchData}, transferring ownership to the caller.
+ */
+ ArrowBatchData createArrowBatchData(long baseLogOffset, long
timestamp, int schemaId);
+ }
+}
diff --git
a/fluss-common/src/main/java/org/apache/fluss/record/DefaultLogRecordBatch.java
b/fluss-common/src/main/java/org/apache/fluss/record/DefaultLogRecordBatch.java
index 288c4b906..e5e961fb1 100644
---
a/fluss-common/src/main/java/org/apache/fluss/record/DefaultLogRecordBatch.java
+++
b/fluss-common/src/main/java/org/apache/fluss/record/DefaultLogRecordBatch.java
@@ -31,6 +31,7 @@ import org.apache.fluss.types.DataType;
import org.apache.fluss.types.RowType;
import org.apache.fluss.utils.ArrowUtils;
import org.apache.fluss.utils.CloseableIterator;
+import org.apache.fluss.utils.IOUtils;
import org.apache.fluss.utils.MurmurHashUtils;
import org.apache.fluss.utils.crc.Crc32C;
@@ -62,6 +63,7 @@ import static
org.apache.fluss.record.LogRecordBatchFormat.schemaIdOffset;
import static
org.apache.fluss.record.LogRecordBatchFormat.statisticsDataOffset;
import static
org.apache.fluss.record.LogRecordBatchFormat.statisticsLengthOffset;
import static org.apache.fluss.record.LogRecordBatchFormat.writeClientIdOffset;
+import static org.apache.fluss.utils.Preconditions.checkArgument;
/* This file is based on source code of Apache Kafka Project
(https://kafka.apache.org/), licensed by the Apache
* Software Foundation (ASF) under the Apache License, Version 2.0. See the
NOTICE file distributed with this work for
@@ -262,6 +264,38 @@ public class DefaultLogRecordBatch implements
LogRecordBatch {
}
}
+ @Override
+ public ArrowBatchData loadArrowBatch(ReadContext context) {
+ if (context.getLogFormat() != LogFormat.ARROW) {
+ throw new UnsupportedOperationException(
+ "loadArrowBatch is only supported for ARROW log format.");
+ }
+
+ int schemaId = schemaId();
+ checkArgument(
+ context instanceof ArrowRecordBatchContext,
+ "Arrow batch loading requires context to implement
ArrowRecordBatchContext, but is %s.",
+ context.getClass().getName());
+ ArrowRecordBatchContext arrowRecordBatchContext =
(ArrowRecordBatchContext) context;
+ checkArgument(isAppendOnly(), "Arrow batch loading only supports
append-only batches.");
+ ArrowRecordBatchContext.UnshadedArrowBatchAccess batchAccess =
+
arrowRecordBatchContext.createUnshadedArrowBatchAccess(schemaId);
+
+ try {
+ int recordsDataOffset = recordsDataOffset();
+ int arrowOffset = position + recordsDataOffset;
+ int arrowLength = sizeInBytes() - recordsDataOffset;
+ batchAccess.loadArrowBatch(segment, arrowOffset, arrowLength);
+ ArrowBatchData arrowBatchData =
+ batchAccess.createArrowBatchData(baseLogOffset(),
commitTimestamp(), schemaId);
+ batchAccess = null;
+ return arrowBatchData;
+ } catch (Throwable t) {
+ IOUtils.closeQuietly(batchAccess);
+ throw t;
+ }
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -352,7 +386,7 @@ public class DefaultLogRecordBatch implements
LogRecordBatch {
VectorSchemaRoot root,
BufferAllocator allocator,
long timestamp) {
- boolean isAppendOnly = (attributes() & APPEND_ONLY_FLAG_MASK) > 0;
+ boolean isAppendOnly = isAppendOnly();
int recordsDataOffset = recordsDataOffset();
if (isAppendOnly) {
// append only batch, no change type vector,
@@ -388,6 +422,10 @@ public class DefaultLogRecordBatch implements
LogRecordBatch {
}
}
+ private boolean isAppendOnly() {
+ return (attributes() & APPEND_ONLY_FLAG_MASK) > 0;
+ }
+
/** The basic implementation for Arrow log record iterator. */
private abstract class ArrowLogRecordIterator extends LogRecordIterator {
private final VectorSchemaRoot root;
diff --git
a/fluss-common/src/main/java/org/apache/fluss/record/FileLogInputStream.java
b/fluss-common/src/main/java/org/apache/fluss/record/FileLogInputStream.java
index fca1a2c49..9e15a508e 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/FileLogInputStream.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/FileLogInputStream.java
@@ -179,6 +179,11 @@ public class FileLogInputStream
return loadFullBatch().records(context);
}
+ @Override
+ public ArrowBatchData loadArrowBatch(ReadContext context) {
+ return loadFullBatch().loadArrowBatch(context);
+ }
+
@Override
public boolean isValid() {
return loadFullBatch().isValid();
diff --git
a/fluss-common/src/main/java/org/apache/fluss/record/LogRecordBatch.java
b/fluss-common/src/main/java/org/apache/fluss/record/LogRecordBatch.java
index f856471b6..5d38bb2f6 100644
--- a/fluss-common/src/main/java/org/apache/fluss/record/LogRecordBatch.java
+++ b/fluss-common/src/main/java/org/apache/fluss/record/LogRecordBatch.java
@@ -173,6 +173,16 @@ public interface LogRecordBatch {
*/
CloseableIterator<LogRecord> records(ReadContext context);
+ /**
+ * Loads the underlying Arrow batch directly.
+ *
+ * <p>This method is only supported for Arrow log batches.
+ */
+ default ArrowBatchData loadArrowBatch(ReadContext context) {
+ throw new UnsupportedOperationException(
+ "loadArrowBatch is only supported for ARROW log format.");
+ }
+
/** The read context of a {@link LogRecordBatch} to read records. */
interface ReadContext {
diff --git
a/fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java
b/fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java
index b65a3ac8e..904fbd505 100644
---
a/fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java
+++
b/fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java
@@ -18,6 +18,7 @@
package org.apache.fluss.record;
import org.apache.fluss.annotation.VisibleForTesting;
+import org.apache.fluss.memory.MemorySegment;
import org.apache.fluss.metadata.LogFormat;
import org.apache.fluss.metadata.Schema;
import org.apache.fluss.metadata.SchemaGetter;
@@ -33,7 +34,9 @@ import
org.apache.fluss.shaded.arrow.org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.fluss.types.DataType;
import org.apache.fluss.types.RowType;
import org.apache.fluss.utils.ArrowUtils;
+import org.apache.fluss.utils.IOUtils;
import org.apache.fluss.utils.Projection;
+import org.apache.fluss.utils.UnshadedArrowReadUtils;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
@@ -42,11 +45,10 @@ import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.IntStream;
-import static org.apache.fluss.utils.Preconditions.checkNotNull;
-
/** A simple implementation for {@link LogRecordBatch.ReadContext}. */
@ThreadSafe
-public class LogRecordReadContext implements LogRecordBatch.ReadContext,
AutoCloseable {
+public class LogRecordReadContext
+ implements LogRecordBatch.ReadContext, ArrowRecordBatchContext,
AutoCloseable {
// the log format of the table
private final LogFormat logFormat;
@@ -54,15 +56,18 @@ public class LogRecordReadContext implements
LogRecordBatch.ReadContext, AutoClo
private final RowType dataRowType;
// the static schemaId of the table, should support dynamic schema
evolution in the future
private final int targetSchemaId;
- // the Arrow memory buffer allocator for the table, should be null if not
ARROW log format
+ // the shaded Arrow memory buffer allocator, should be null if not ARROW
log format
@Nullable private final BufferAllocator bufferAllocator;
- // the final selected fields of the read data
+ // the unshaded Arrow memory buffer allocator, declared as AutoCloseable
to avoid
+ // importing unshaded Arrow classes in this module (they are
provided-scope)
+ @Nullable private volatile AutoCloseable unshadedBufferAllocator;
private final FieldGetter[] selectedFieldGetters;
// whether the projection is push downed to the server side and the
returned data is pruned.
private final boolean projectionPushDowned;
private final SchemaGetter schemaGetter;
private final ConcurrentHashMap<Integer, VectorSchemaRoot>
vectorSchemaRootMap =
new ConcurrentHashMap<>();
+ private final Object unshadedArrowResourceLock = new Object();
public static LogRecordReadContext createReadContext(
TableInfo tableInfo,
@@ -318,6 +323,22 @@ public class LogRecordReadContext implements
LogRecordBatch.ReadContext, AutoClo
return schemaGetter;
}
+ /**
+ * Returns the column index mapping for schema evolution, or {@code null}
if the batch schema
+ * matches the current schema and no remapping is needed.
+ *
+ * <p>Each entry maps an output column to its position in the batch's
schema. A value of {@code
+ * -1} means the column does not exist in the batch's schema and should be
filled with nulls.
+ */
+ @Nullable
+ private int[] getSchemaEvolutionMapping(int schemaId) {
+ ProjectedRow projectedRow = getOutputProjectedRow(schemaId);
+ if (projectedRow == null) {
+ return null;
+ }
+ return projectedRow.getIndexMapping();
+ }
+
@Override
public VectorSchemaRoot getVectorSchemaRoot(int schemaId) {
if (logFormat != LogFormat.ARROW) {
@@ -338,10 +359,18 @@ public class LogRecordReadContext implements
LogRecordBatch.ReadContext, AutoClo
if (logFormat != LogFormat.ARROW) {
throw new IllegalArgumentException("Only Arrow log format provides
buffer allocator.");
}
- checkNotNull(bufferAllocator, "The buffer allocator is not
available.");
return bufferAllocator;
}
+ @Override
+ public UnshadedArrowBatchAccess createUnshadedArrowBatchAccess(int
schemaId) {
+ if (logFormat != LogFormat.ARROW) {
+ throw new IllegalArgumentException(
+ "Only Arrow log format provides unshaded Arrow
resources.");
+ }
+ return new UnshadedArrowBatchAccessImpl(schemaId);
+ }
+
@Nullable
@Override
public ProjectedRow getOutputProjectedRow(int schemaId) {
@@ -359,9 +388,88 @@ public class LogRecordReadContext implements
LogRecordBatch.ReadContext, AutoClo
public void close() {
vectorSchemaRootMap.values().forEach(VectorSchemaRoot::close);
+ vectorSchemaRootMap.clear();
if (bufferAllocator != null) {
bufferAllocator.close();
}
+
+ synchronized (unshadedArrowResourceLock) {
+ if (unshadedBufferAllocator != null) {
+ try {
+ unshadedBufferAllocator.close();
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to close Arrow buffer allocator. "
+ + "Arrow batches returned by
pollRecordBatch() must be closed "
+ + "before closing the scanner.",
+ e);
+ }
+ unshadedBufferAllocator = null;
+ }
+ }
+ }
+
+ private AutoCloseable getOrCreateUnshadedBufferAllocator() {
+ AutoCloseable allocator = unshadedBufferAllocator;
+ if (allocator != null) {
+ return allocator;
+ }
+
+ synchronized (unshadedArrowResourceLock) {
+ if (unshadedBufferAllocator == null) {
+ unshadedBufferAllocator = new
org.apache.arrow.memory.RootAllocator(Long.MAX_VALUE);
+ }
+ return unshadedBufferAllocator;
+ }
+ }
+
+ private final class UnshadedArrowBatchAccessImpl implements
UnshadedArrowBatchAccess {
+ private final org.apache.arrow.memory.BufferAllocator allocator;
+ private org.apache.arrow.vector.VectorSchemaRoot readRoot;
+ private org.apache.arrow.vector.VectorSchemaRoot outputRoot;
+
+ private UnshadedArrowBatchAccessImpl(int schemaId) {
+ this.allocator =
+ (org.apache.arrow.memory.BufferAllocator)
getOrCreateUnshadedBufferAllocator();
+ this.readRoot =
+ org.apache.arrow.vector.VectorSchemaRoot.create(
+
org.apache.fluss.utils.UnshadedArrowReadUtils.toArrowSchema(
+ getRowType(schemaId)),
+ allocator);
+ this.outputRoot = readRoot;
+ }
+
+ @Override
+ public void loadArrowBatch(MemorySegment segment, int arrowOffset, int
arrowLength) {
+ UnshadedArrowReadUtils.loadArrowBatch(
+ segment, arrowOffset, arrowLength, readRoot, allocator);
+ }
+
+ @Override
+ public ArrowBatchData createArrowBatchData(
+ long baseLogOffset, long timestamp, int schemaId) {
+ int[] schemaMapping = getSchemaEvolutionMapping(schemaId);
+ if (schemaMapping != null) {
+ outputRoot =
+ UnshadedArrowReadUtils.projectVectorSchemaRoot(
+ readRoot, dataRowType, schemaMapping,
allocator);
+ readRoot.close();
+ readRoot = null;
+ }
+ ArrowBatchData arrowBatchData =
+ new ArrowBatchData(outputRoot, baseLogOffset, timestamp,
schemaId);
+ outputRoot = null;
+ readRoot = null;
+ return arrowBatchData;
+ }
+
+ @Override
+ public void close() {
+ IOUtils.closeQuietly(outputRoot);
+ if (outputRoot != readRoot) {
+ IOUtils.closeQuietly(readRoot);
+ }
+ }
}
private boolean isSameRowType(int schemaId) {
diff --git
a/fluss-common/src/main/java/org/apache/fluss/record/MemoryLogRecordsArrowBuilder.java
b/fluss-common/src/main/java/org/apache/fluss/record/MemoryLogRecordsArrowBuilder.java
index 3276771d0..8ca7bc440 100644
---
a/fluss-common/src/main/java/org/apache/fluss/record/MemoryLogRecordsArrowBuilder.java
+++
b/fluss-common/src/main/java/org/apache/fluss/record/MemoryLogRecordsArrowBuilder.java
@@ -142,6 +142,18 @@ public class MemoryLogRecordsArrowBuilder implements
AutoCloseable {
baseLogOffset, schemaId, magic, arrowWriter, outputView,
false, null);
}
+ @VisibleForTesting
+ public static MemoryLogRecordsArrowBuilder builder(
+ long baseLogOffset,
+ byte magic,
+ int schemaId,
+ ArrowWriter arrowWriter,
+ AbstractPagedOutputView outputView,
+ boolean appendOnly) {
+ return new MemoryLogRecordsArrowBuilder(
+ baseLogOffset, schemaId, magic, arrowWriter, outputView,
appendOnly, null);
+ }
+
/** Builder with limited write size and the memory segment used to
serialize records. */
public static MemoryLogRecordsArrowBuilder builder(
int schemaId,
diff --git
a/fluss-common/src/main/java/org/apache/fluss/record/UnshadedFlussVectorLoader.java
b/fluss-common/src/main/java/org/apache/fluss/record/UnshadedFlussVectorLoader.java
new file mode 100644
index 000000000..7798f8ef6
--- /dev/null
+++
b/fluss-common/src/main/java/org/apache/fluss/record/UnshadedFlussVectorLoader.java
@@ -0,0 +1,132 @@
+/*
+ * 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.fluss.record;
+
+import org.apache.arrow.memory.ArrowBuf;
+import org.apache.arrow.util.Collections2;
+import org.apache.arrow.util.Preconditions;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.TypeLayout;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.compression.CompressionCodec;
+import org.apache.arrow.vector.compression.CompressionUtil;
+import org.apache.arrow.vector.compression.CompressionUtil.CodecType;
+import org.apache.arrow.vector.compression.NoCompressionCodec;
+import org.apache.arrow.vector.ipc.message.ArrowFieldNode;
+import org.apache.arrow.vector.ipc.message.ArrowRecordBatch;
+import org.apache.arrow.vector.types.pojo.Field;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+/** Unshaded variant of {@link FlussVectorLoader} for scanner/read path. */
+public class UnshadedFlussVectorLoader {
+ private final VectorSchemaRoot root;
+ private final CompressionCodec.Factory factory;
+ private boolean decompressionNeeded;
+
+ public UnshadedFlussVectorLoader(VectorSchemaRoot root,
CompressionCodec.Factory factory) {
+ this.root = root;
+ this.factory = factory;
+ }
+
+ public void load(ArrowRecordBatch recordBatch) {
+ Iterator<ArrowBuf> buffers = recordBatch.getBuffers().iterator();
+ Iterator<ArrowFieldNode> nodes = recordBatch.getNodes().iterator();
+ CompressionUtil.CodecType codecType =
+
CodecType.fromCompressionType(recordBatch.getBodyCompression().getCodec());
+ this.decompressionNeeded = codecType != CodecType.NO_COMPRESSION;
+ CompressionCodec codec =
+ this.decompressionNeeded
+ ? this.factory.createCodec(codecType)
+ : NoCompressionCodec.INSTANCE;
+
+ for (FieldVector fieldVector : this.root.getFieldVectors()) {
+ this.loadBuffers(fieldVector, fieldVector.getField(), buffers,
nodes, codec);
+ }
+
+ this.root.setRowCount(recordBatch.getLength());
+ if (nodes.hasNext() || buffers.hasNext()) {
+ throw new IllegalArgumentException(
+ "not all nodes and buffers were consumed. nodes: "
+ + Collections2.toString(nodes)
+ + " buffers: "
+ + Collections2.toString(buffers));
+ }
+ }
+
+ private void loadBuffers(
+ FieldVector vector,
+ Field field,
+ Iterator<ArrowBuf> buffers,
+ Iterator<ArrowFieldNode> nodes,
+ CompressionCodec codec) {
+ Preconditions.checkArgument(
+ nodes.hasNext(), "no more field nodes for field %s and vector
%s", field, vector);
+ ArrowFieldNode fieldNode = nodes.next();
+ int bufferLayoutCount = TypeLayout.getTypeBufferCount(field.getType());
+ List<ArrowBuf> ownBuffers = new ArrayList<>(bufferLayoutCount);
+
+ try {
+ for (int j = 0; j < bufferLayoutCount; ++j) {
+ ArrowBuf nextBuf = buffers.next();
+ ArrowBuf bufferToAdd =
+ nextBuf.writerIndex() > 0L
+ ? codec.decompress(vector.getAllocator(),
nextBuf)
+ : nextBuf;
+ ownBuffers.add(bufferToAdd);
+ if (this.decompressionNeeded) {
+ nextBuf.getReferenceManager().retain();
+ }
+ }
+ vector.loadFieldBuffers(fieldNode, ownBuffers);
+ } catch (RuntimeException e) {
+ throw new IllegalArgumentException(
+ "Could not load buffers for field "
+ + field
+ + ". error message: "
+ + e.getMessage(),
+ e);
+ } finally {
+ if (this.decompressionNeeded) {
+ for (ArrowBuf buf : ownBuffers) {
+ buf.close();
+ }
+ }
+ }
+
+ List<Field> children = field.getChildren();
+ if (!children.isEmpty()) {
+ List<FieldVector> childrenFromFields =
vector.getChildrenFromFields();
+ Preconditions.checkArgument(
+ children.size() == childrenFromFields.size(),
+ "should have as many children as in the schema: found %s
expected %s",
+ childrenFromFields.size(),
+ children.size());
+
+ for (int i = 0; i < childrenFromFields.size(); ++i) {
+ Field child = children.get(i);
+ FieldVector fieldVector = childrenFromFields.get(i);
+ this.loadBuffers(fieldVector, child, buffers, nodes, codec);
+ }
+ }
+ }
+}
diff --git a/fluss-common/src/main/java/org/apache/fluss/row/ProjectedRow.java
b/fluss-common/src/main/java/org/apache/fluss/row/ProjectedRow.java
index 25656204d..0af0f40bd 100644
--- a/fluss-common/src/main/java/org/apache/fluss/row/ProjectedRow.java
+++ b/fluss-common/src/main/java/org/apache/fluss/row/ProjectedRow.java
@@ -19,11 +19,10 @@ package org.apache.fluss.row;
import org.apache.fluss.annotation.PublicEvolving;
import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.utils.SchemaUtil;
import java.util.Arrays;
-import static org.apache.fluss.utils.SchemaUtil.getIndexMapping;
-
/**
* An implementation of {@link InternalRow} which provides a projected view of
the underlying {@link
* InternalRow}.
@@ -190,10 +189,15 @@ public class ProjectedRow implements InternalRow {
}
public static ProjectedRow from(Schema originSchema, Schema
expectedSchema) {
- int[] indexMapping = getIndexMapping(originSchema, expectedSchema);
+ int[] indexMapping = SchemaUtil.getIndexMapping(originSchema,
expectedSchema);
return new ProjectedRow(indexMapping);
}
+ /** Returns the index mapping used by this projected row. */
+ public int[] getIndexMapping() {
+ return Arrays.copyOf(indexMapping, indexMapping.length);
+ }
+
/**
* Returns the underlying row before column projection.
*
diff --git
a/fluss-common/src/main/java/org/apache/fluss/utils/UnshadedArrowReadUtils.java
b/fluss-common/src/main/java/org/apache/fluss/utils/UnshadedArrowReadUtils.java
new file mode 100644
index 000000000..dff13f040
--- /dev/null
+++
b/fluss-common/src/main/java/org/apache/fluss/utils/UnshadedArrowReadUtils.java
@@ -0,0 +1,122 @@
+/*
+ * 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.fluss.utils;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.compression.UnshadedArrowCompressionFactory;
+import org.apache.fluss.memory.MemorySegment;
+import org.apache.fluss.record.UnshadedFlussVectorLoader;
+import org.apache.fluss.types.RowType;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ReadChannel;
+import org.apache.arrow.vector.ipc.message.ArrowRecordBatch;
+import org.apache.arrow.vector.ipc.message.MessageSerializer;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.arrow.vector.util.TransferPair;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+/** Utilities for loading and projecting Arrow scan batches with unshaded
Arrow classes. */
+@Internal
+public final class UnshadedArrowReadUtils {
+
+ private UnshadedArrowReadUtils() {}
+
+ /**
+ * Converts a Fluss RowType to an unshaded Arrow Schema.
+ *
+ * <p>Converts via FlatBuffer binary serialization: shaded Schema → bytes
→ unshaded Schema.
+ * Both shaded and unshaded Arrow use the same FlatBuffer wire format, so
the binary
+ * representation is identical regardless of Java package relocation.
+ */
+ public static Schema toArrowSchema(RowType rowType) {
+
org.apache.fluss.shaded.arrow.org.apache.arrow.vector.types.pojo.Schema
shadedSchema =
+ ArrowUtils.toArrowSchema(rowType);
+ return
Schema.deserializeMessage(ByteBuffer.wrap(shadedSchema.serializeAsMessage()));
+ }
+
+ public static void loadArrowBatch(
+ MemorySegment segment,
+ int arrowOffset,
+ int arrowLength,
+ VectorSchemaRoot schemaRoot,
+ BufferAllocator allocator) {
+ ByteBuffer arrowBatchBuffer = segment.wrap(arrowOffset, arrowLength);
+ try (ReadChannel channel =
+ new ReadChannel(new
ByteBufferReadableChannel(arrowBatchBuffer));
+ ArrowRecordBatch batch =
+ MessageSerializer.deserializeRecordBatch(channel,
allocator)) {
+ new UnshadedFlussVectorLoader(schemaRoot,
UnshadedArrowCompressionFactory.INSTANCE)
+ .load(batch);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to deserialize
ArrowRecordBatch.", e);
+ }
+ }
+
+ /**
+ * Projects a VectorSchemaRoot from old schema to target schema using a
column projection
+ * mapping.
+ *
+ * @param sourceRoot the source VectorSchemaRoot with old schema data
+ * @param targetRowType the target RowType to project to
+ * @param columnProjection mapping from target column index to source
column index; -1 means
+ * column doesn't exist in source (filled with nulls)
+ * @param allocator the allocator for creating new vectors
+ * @return a new VectorSchemaRoot with the target schema
+ */
+ public static VectorSchemaRoot projectVectorSchemaRoot(
+ VectorSchemaRoot sourceRoot,
+ RowType targetRowType,
+ int[] columnProjection,
+ BufferAllocator allocator) {
+ int rowCount = sourceRoot.getRowCount();
+ Schema targetSchema = toArrowSchema(targetRowType);
+ List<Field> targetFields = targetSchema.getFields();
+ List<FieldVector> targetVectors = new
ArrayList<>(columnProjection.length);
+ try {
+ for (int i = 0; i < columnProjection.length; i++) {
+ if (columnProjection[i] < 0) {
+ // Column doesn't exist in source, fill with nulls
+ FieldVector targetVector =
targetFields.get(i).createVector(allocator);
+ targetVector.allocateNew();
+ for (int rowId = 0; rowId < rowCount; rowId++) {
+ targetVector.setNull(rowId);
+ }
+ targetVector.setValueCount(rowCount);
+ targetVectors.add(targetVector);
+ } else {
+ FieldVector sourceVector =
sourceRoot.getVector(columnProjection[i]);
+ TransferPair transfer =
sourceVector.getTransferPair(allocator);
+ transfer.splitAndTransfer(0, rowCount);
+ targetVectors.add((FieldVector) transfer.getTo());
+ }
+ }
+ } catch (Exception e) {
+ targetVectors.forEach(FieldVector::close);
+ throw e;
+ }
+ return new VectorSchemaRoot(targetFields, targetVectors, rowCount);
+ }
+}
diff --git
a/fluss-common/src/test/java/org/apache/fluss/record/FileLogInputStreamTest.java
b/fluss-common/src/test/java/org/apache/fluss/record/FileLogInputStreamTest.java
index 6afb25549..5a994ba8a 100644
---
a/fluss-common/src/test/java/org/apache/fluss/record/FileLogInputStreamTest.java
+++
b/fluss-common/src/test/java/org/apache/fluss/record/FileLogInputStreamTest.java
@@ -20,6 +20,8 @@ package org.apache.fluss.record;
import org.apache.fluss.metadata.LogFormat;
import org.apache.fluss.utils.CloseableIterator;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.VarCharVector;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
@@ -27,11 +29,13 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.io.File;
import java.util.Collections;
+import java.util.List;
import java.util.Optional;
import static org.apache.fluss.record.LogRecordBatchFormat.LOG_MAGIC_VALUE_V0;
import static org.apache.fluss.record.LogRecordBatchFormat.LOG_MAGIC_VALUE_V1;
import static org.apache.fluss.record.LogRecordBatchFormat.LOG_MAGIC_VALUE_V2;
+import static org.apache.fluss.record.TestData.DATA1;
import static org.apache.fluss.record.TestData.DATA1_ROW_TYPE;
import static org.apache.fluss.record.TestData.DATA1_SCHEMA;
import static org.apache.fluss.record.TestData.DEFAULT_SCHEMA_ID;
@@ -257,4 +261,50 @@ public class FileLogInputStreamTest extends LogTestBase {
assertThat(memoryStats.getMaxValues().getString(1).toString()).isEqualTo("j");
}
}
+
+ @ParameterizedTest
+ @ValueSource(bytes = {LOG_MAGIC_VALUE_V0, LOG_MAGIC_VALUE_V1})
+ void testLoadArrowBatch(byte recordBatchMagic) throws Exception {
+ List<Object[]> data = DATA1;
+ try (FileLogRecords fileLogRecords =
+ FileLogRecords.open(new File(tempDir,
"test_arrow_batch.tmp"))) {
+ fileLogRecords.append(
+ createRecordsWithoutBaseLogOffset(
+ DATA1_ROW_TYPE,
+ DEFAULT_SCHEMA_ID,
+ 0L,
+ -1L,
+ recordBatchMagic,
+ data,
+ LogFormat.ARROW,
+ true));
+ fileLogRecords.flush();
+
+ FileLogInputStream logInputStream =
+ new FileLogInputStream(fileLogRecords, 0,
fileLogRecords.sizeInBytes());
+ FileLogInputStream.FileChannelLogRecordBatch batch =
logInputStream.nextBatch();
+ assertThat(batch).isNotNull();
+
+ TestingSchemaGetter schemaGetter =
+ new TestingSchemaGetter(DEFAULT_SCHEMA_ID, DATA1_SCHEMA);
+ try (LogRecordReadContext readContext =
+ LogRecordReadContext.createArrowReadContext(
+ DATA1_ROW_TYPE, DEFAULT_SCHEMA_ID, schemaGetter)) {
+ try (ArrowBatchData arrowBatchData =
batch.loadArrowBatch(readContext)) {
+
assertThat(arrowBatchData.getRecordCount()).isEqualTo(data.size());
+
assertThat(arrowBatchData.getBaseLogOffset()).isEqualTo(0L);
+
assertThat(arrowBatchData.getSchemaId()).isEqualTo(DEFAULT_SCHEMA_ID);
+
+ org.apache.arrow.vector.VectorSchemaRoot root =
+ arrowBatchData.getVectorSchemaRoot();
+ IntVector intVector = (IntVector) root.getVector(0);
+ VarCharVector stringVector = (VarCharVector)
root.getVector(1);
+ for (int i = 0; i < data.size(); i++) {
+ assertThat(intVector.get(i)).isEqualTo(data.get(i)[0]);
+
assertThat(stringVector.getObject(i).toString()).isEqualTo(data.get(i)[1]);
+ }
+ }
+ }
+ }
+ }
}
diff --git
a/fluss-common/src/test/java/org/apache/fluss/testutils/DataTestUtils.java
b/fluss-common/src/test/java/org/apache/fluss/testutils/DataTestUtils.java
index 5ee50776a..aaa449d46 100644
--- a/fluss-common/src/test/java/org/apache/fluss/testutils/DataTestUtils.java
+++ b/fluss-common/src/test/java/org/apache/fluss/testutils/DataTestUtils.java
@@ -494,6 +494,20 @@ public class DataTestUtils {
List<Object[]> objects,
LogFormat logFormat)
throws Exception {
+ return createRecordsWithoutBaseLogOffset(
+ rowType, schemaId, offsetBase, maxTimestamp, magic, objects,
logFormat, false);
+ }
+
+ public static MemoryLogRecords createRecordsWithoutBaseLogOffset(
+ RowType rowType,
+ int schemaId,
+ long offsetBase,
+ long maxTimestamp,
+ byte magic,
+ List<Object[]> objects,
+ LogFormat logFormat,
+ boolean appendOnly)
+ throws Exception {
List<ChangeType> changeTypes =
objects.stream().map(row ->
ChangeType.APPEND_ONLY).collect(Collectors.toList());
return createBasicMemoryLogRecords(
@@ -507,7 +521,8 @@ public class DataTestUtils {
changeTypes,
objects,
logFormat,
- DEFAULT_COMPRESSION);
+ DEFAULT_COMPRESSION,
+ appendOnly);
}
public static MemoryLogRecords createBasicMemoryLogRecords(
@@ -534,7 +549,37 @@ public class DataTestUtils {
changeTypes,
objects,
logFormat,
- arrowCompressionInfo);
+ arrowCompressionInfo,
+ false);
+ }
+
+ public static MemoryLogRecords createBasicMemoryLogRecords(
+ RowType rowType,
+ int schemaId,
+ long offsetBase,
+ long maxTimestamp,
+ byte magic,
+ long writerId,
+ int batchSequence,
+ List<ChangeType> changeTypes,
+ List<Object[]> objects,
+ LogFormat logFormat,
+ ArrowCompressionInfo arrowCompressionInfo,
+ boolean appendOnly)
+ throws Exception {
+ return createMemoryLogRecords(
+ rowType,
+ schemaId,
+ offsetBase,
+ maxTimestamp,
+ magic,
+ writerId,
+ batchSequence,
+ changeTypes,
+ objects,
+ logFormat,
+ arrowCompressionInfo,
+ appendOnly);
}
public static MemoryLogRecords createMemoryLogRecords(
@@ -548,7 +593,8 @@ public class DataTestUtils {
List<ChangeType> changeTypes,
List<Object[]> objects,
LogFormat logFormat,
- ArrowCompressionInfo arrowCompressionInfo)
+ ArrowCompressionInfo arrowCompressionInfo,
+ boolean appendOnly)
throws Exception {
if (logFormat == LogFormat.ARROW) {
List<InternalRow> rows =
@@ -563,7 +609,8 @@ public class DataTestUtils {
batchSequence,
changeTypes,
rows,
- arrowCompressionInfo);
+ arrowCompressionInfo,
+ appendOnly);
} else {
return createIndexedMemoryLogRecords(
offsetBase,
@@ -614,7 +661,8 @@ public class DataTestUtils {
int batchSequence,
List<ChangeType> changeTypes,
List<InternalRow> rows,
- ArrowCompressionInfo arrowCompressionInfo)
+ ArrowCompressionInfo arrowCompressionInfo,
+ boolean appendOnly)
throws Exception {
try (BufferAllocator allocator = new RootAllocator(Integer.MAX_VALUE);
ArrowWriterPool provider = new ArrowWriterPool(allocator)) {
@@ -627,7 +675,8 @@ public class DataTestUtils {
magic,
schemaId,
writer,
- new ManagedPagedOutputView(new
TestingMemorySegmentPool(10 * 1024)));
+ new ManagedPagedOutputView(new
TestingMemorySegmentPool(10 * 1024)),
+ appendOnly);
for (int i = 0; i < changeTypes.size(); i++) {
builder.append(changeTypes.get(i), rows.get(i));
}
