You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "luoyuxia (via GitHub)" <gi...@apache.org> on 2026/06/04 02:34:45 UTC
[PR] [lake/paimon] TieringSourceReader adjust to scan arrow record batch and write arrow record batch to lake [fluss]
luoyuxia opened a new pull request, #3430:
URL: https://github.com/apache/fluss/pull/3430
## Summary
- Adjust `TieringSourceReader` / `TieringSplitReader` to scan arrow record batch and write arrow record batch to lake
- Add `AppendOnlyArrowBatchHelper` and `Arrow2PaimonVectorConverter` to support writing arrow batches to Paimon
- Fix split completion check and arrow type conversion for tiering
Closes #2966
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] TieringSourceReader adjust to scan arrow record batch and write arrow record batch to lake [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #3430:
URL: https://github.com/apache/fluss/pull/3430#discussion_r3354288738
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -171,8 +187,22 @@ public RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> fetch() throws I
if (reachTieringMaxDurationTables.contains(currentTableId)) {
return forceCompleteTieringLogRecords();
}
- ScanRecords scanRecords = currentLogScanner.poll(pollTimeout);
- return forLogRecords(scanRecords);
+ if (useRecordBatchPath()) {
+ ArrowScanRecords arrowScanRecords =
+ ((LogScannerImpl) currentLogScanner).pollRecordBatch(pollTimeout);
+ return processLogRecords(
+ arrowScanRecords.buckets(),
+ arrowScanRecords::records,
+ this::handleArrowBatchRecords,
Review Comment:
If processLogRecords throws partway through iterating buckets, the remaining buckets' Arrow batches won't be closed. Since ArrowScanRecords is AutoCloseable, wrapping it might be the cleanest way to ensure all off-heap memory is released on error paths.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] TieringSourceReader adjust to scan arrow record batch and write arrow record batch to lake [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on PR #3430:
URL: https://github.com/apache/fluss/pull/3430#issuecomment-4620739569
LGTM!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] TieringSourceReader adjust to scan arrow record batch and write arrow record batch to lake [fluss]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #3430:
URL: https://github.com/apache/fluss/pull/3430#discussion_r3363615126
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -732,6 +874,41 @@ public Set<String> finishedSplits() {
}
}
+ /**
+ * Callback interface for processing records within a single bucket. Encapsulates the
+ * differences in write strategy between the row-based (ScanRecord) and Arrow batch
+ * (ArrowBatchData) paths.
+ *
+ * @param <R> the record type (ScanRecord or ArrowBatchData)
+ */
+ @FunctionalInterface
+ private interface BucketRecordsHandler<R> {
+
+ /**
+ * Processes the records for a bucket and writes them to the lake.
+ *
+ * @param records the records for this bucket
+ * @param lakeWriterSupplier supplier for lazily creating the lake writer
+ * @param stoppingOffset the stopping offset for this bucket
+ * @return the timestamp of the last written record, or -1 if no records were written
+ * @throws IOException if an I/O error occurs during writing
+ */
+ long handleRecords(
+ List<R> records,
+ SupplierWithException<LakeWriter<?>, IOException> lakeWriterSupplier,
+ long stoppingOffset)
+ throws IOException;
+ }
+
+ private static boolean checkUnshadedArrowAvailable(ClassLoader classLoader) {
Review Comment:
shall we check for ArrowBundleRecords from paimon-arrow?
it's provided, so it might fail, if I understand this relationship correctly
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] TieringSourceReader adjust to scan arrow record batch and write arrow record batch to lake [fluss]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #3430:
URL: https://github.com/apache/fluss/pull/3430#discussion_r3354241145
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -421,6 +486,101 @@ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> forLogRecords(
return new TableBucketWriteResultWithSplitIds(writeResults, finishedSplitIds);
}
+ /** Handles row-based ScanRecord writing for the log path. */
+ private LogOffsetAndTimestamp handleLogRecords(
+ List<ScanRecord> records,
+ SupplierWithException<LakeWriter<?>, IOException> lakeWriterSupplier,
+ long stoppingOffset)
+ throws IOException {
+ long lastWrittenOffset = -1;
+ long lastWrittenTimestamp = -1;
+ LakeWriter<?> lakeWriter = null;
+ for (ScanRecord record : records) {
+ if (record.logOffset() < stoppingOffset) {
+ if (lakeWriter == null) {
+ lakeWriter = lakeWriterSupplier.get();
+ }
+ lakeWriter.write(record);
+ lastWrittenOffset = record.logOffset();
+ lastWrittenTimestamp = record.timestamp();
+ if (record.getSizeInBytes() > 0) {
+ tieringMetrics.recordBytesRead(record.getSizeInBytes());
+ }
+ }
+ }
+ ScanRecord lastRecord = records.get(records.size() - 1);
+ return new LogOffsetAndTimestamp(
+ lastWrittenOffset, lastWrittenTimestamp, lastRecord.logOffset());
+ }
+
+ /** Handles Arrow batch writing for the record batch path. */
+ private LogOffsetAndTimestamp handleArrowBatchRecords(
+ List<ArrowBatchData> batches,
+ SupplierWithException<LakeWriter<?>, IOException> lakeWriterSupplier,
+ long stoppingOffset)
+ throws IOException {
+ SupportsRecordBatchWrite batchWriter = null;
+ long lastWrittenBatchEndOffset = -1;
+ long lastWrittenTimestamp = -1;
+ long lastScannedOffset = -1;
+ try {
+ for (ArrowBatchData batch : batches) {
+ long batchBaseOffset = batch.getBaseLogOffset();
+ long batchRecordCount = batch.getRecordCount();
+ long batchTimestamp = batch.getTimestamp();
+ lastScannedOffset = batchBaseOffset + batchRecordCount - 1L;
+ if (batchBaseOffset >= stoppingOffset) {
+ batch.close();
+ continue;
+ }
+
+ long writableRowCount = stoppingOffset - batchBaseOffset;
+ int writableRows = (int) Math.min(batchRecordCount, writableRowCount);
+ if (writableRows <= 0) {
+ batch.close();
+ continue;
+ }
+
+ ArrowBatchData batchToWrite = batch;
+ if (writableRows < batchRecordCount) {
+ batchToWrite = batch.truncateAndTransferOwnership(writableRows);
Review Comment:
There's a potential resource leak here: if Exception throws after truncateAndTransferOwnership, the newly created batchToWrite isn't tracked in batches and won't be closed by the catch block.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] TieringSourceReader adjust to scan arrow record batch and write arrow record batch to lake [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #3430:
URL: https://github.com/apache/fluss/pull/3430
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] TieringSourceReader adjust to scan arrow record batch and write arrow record batch to lake [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3430:
URL: https://github.com/apache/fluss/pull/3430#discussion_r3353112887
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -171,8 +187,22 @@ public RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> fetch() throws I
if (reachTieringMaxDurationTables.contains(currentTableId)) {
return forceCompleteTieringLogRecords();
}
- ScanRecords scanRecords = currentLogScanner.poll(pollTimeout);
- return forLogRecords(scanRecords);
+ if (useRecordBatchPath()) {
+ ArrowScanRecords arrowScanRecords =
+ ((LogScannerImpl) currentLogScanner).pollRecordBatch(pollTimeout);
+ return processLogRecords(
+ arrowScanRecords.buckets(),
Review Comment:
ArrowScanRecords holds off-heap Arrow memory and is AutoCloseable, but fetch() doesn't close the polled container. If processLogRecords throws or returns early, batches can leak until the scanner is closed (and the ArrowScanRecords javadoc explicitly recommends try-with-resources). Wrap pollRecordBatch(...) in try-with-resources and keep the existing per-batch close logic as a safety net.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -421,6 +486,88 @@ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> forLogRecords(
return new TableBucketWriteResultWithSplitIds(writeResults, finishedSplitIds);
}
+ /** Handles row-based ScanRecord writing for the log path. */
+ private LogOffsetAndTimestamp handleLogRecords(
+ List<ScanRecord> records,
+ SupplierWithException<LakeWriter<?>, IOException> lakeWriterSupplier,
+ long stoppingOffset)
+ throws IOException {
+ long lastWrittenOffset = -1;
+ long lastWrittenTimestamp = -1;
+ LakeWriter<?> lakeWriter = null;
+ for (ScanRecord record : records) {
+ if (record.logOffset() < stoppingOffset) {
+ if (lakeWriter == null) {
+ lakeWriter = lakeWriterSupplier.get();
+ }
+ lakeWriter.write(record);
+ lastWrittenOffset = record.logOffset();
+ lastWrittenTimestamp = record.timestamp();
+ if (record.getSizeInBytes() > 0) {
+ tieringMetrics.recordBytesRead(record.getSizeInBytes());
+ }
+ }
+ }
+ ScanRecord lastRecord = records.get(records.size() - 1);
+ return new LogOffsetAndTimestamp(
+ lastWrittenOffset, lastWrittenTimestamp, lastRecord.logOffset());
+ }
+
+ /** Handles Arrow batch writing for the record batch path. */
+ private LogOffsetAndTimestamp handleArrowBatchRecords(
+ List<ArrowBatchData> batches,
+ SupplierWithException<LakeWriter<?>, IOException> lakeWriterSupplier,
+ long stoppingOffset)
+ throws IOException {
+ SupportsRecordBatchWrite batchWriter = null;
+ long lastWrittenBatchEndOffset = -1;
+ long lastWrittenTimestamp = -1;
+ for (ArrowBatchData batch : batches) {
+ long batchBaseOffset = batch.getBaseLogOffset();
+ if (batchBaseOffset >= stoppingOffset) {
+ batch.close();
+ continue;
+ }
+
+ long writableRowCount = stoppingOffset - batchBaseOffset;
+ int writableRows = (int) Math.min(batch.getRecordCount(), writableRowCount);
+ if (writableRows <= 0) {
+ batch.close();
+ continue;
+ }
+
+ ArrowBatchData batchToWrite = batch;
+ if (writableRows < batch.getRecordCount()) {
+ batchToWrite = batch.truncateAndTransferOwnership(writableRows);
+ }
+
+ if (batchWriter == null) {
+ batchWriter = (SupportsRecordBatchWrite) lakeWriterSupplier.get();
+ }
+ long batchEndOffset = batchBaseOffset + writableRows - 1L;
+ long batchSizeInBytes = batchToWrite.getSizeInBytes();
+ try (ArrowRecordBatch arrowRecordBatch = new ArrowRecordBatch(batchToWrite)) {
+ batchWriter.write(arrowRecordBatch);
+ }
+ if (batchSizeInBytes > 0) {
+ tieringMetrics.recordBytesRead(batchSizeInBytes);
+ }
+ lastWrittenBatchEndOffset = batchEndOffset;
+ lastWrittenTimestamp = batchToWrite.getTimestamp();
+ }
+ ArrowBatchData lastBatch = batches.get(batches.size() - 1);
+ long lastScannedOffset = lastBatch.getBaseLogOffset() + lastBatch.getRecordCount() - 1L;
+ return new LogOffsetAndTimestamp(
+ lastWrittenBatchEndOffset, lastWrittenTimestamp, lastScannedOffset);
Review Comment:
This computes lastScannedOffset from the last batch after the loop, but the last batch has already been closed during processing (try-with-resources / explicit close). After capturing lastScannedOffset before the loop, return that value here instead of reading from a potentially-closed ArrowBatchData.
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -421,6 +486,88 @@ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> forLogRecords(
return new TableBucketWriteResultWithSplitIds(writeResults, finishedSplitIds);
}
+ /** Handles row-based ScanRecord writing for the log path. */
+ private LogOffsetAndTimestamp handleLogRecords(
+ List<ScanRecord> records,
+ SupplierWithException<LakeWriter<?>, IOException> lakeWriterSupplier,
+ long stoppingOffset)
+ throws IOException {
+ long lastWrittenOffset = -1;
+ long lastWrittenTimestamp = -1;
+ LakeWriter<?> lakeWriter = null;
+ for (ScanRecord record : records) {
+ if (record.logOffset() < stoppingOffset) {
+ if (lakeWriter == null) {
+ lakeWriter = lakeWriterSupplier.get();
+ }
+ lakeWriter.write(record);
+ lastWrittenOffset = record.logOffset();
+ lastWrittenTimestamp = record.timestamp();
+ if (record.getSizeInBytes() > 0) {
+ tieringMetrics.recordBytesRead(record.getSizeInBytes());
+ }
+ }
+ }
+ ScanRecord lastRecord = records.get(records.size() - 1);
+ return new LogOffsetAndTimestamp(
+ lastWrittenOffset, lastWrittenTimestamp, lastRecord.logOffset());
+ }
+
+ /** Handles Arrow batch writing for the record batch path. */
+ private LogOffsetAndTimestamp handleArrowBatchRecords(
+ List<ArrowBatchData> batches,
+ SupplierWithException<LakeWriter<?>, IOException> lakeWriterSupplier,
+ long stoppingOffset)
+ throws IOException {
+ SupportsRecordBatchWrite batchWriter = null;
+ long lastWrittenBatchEndOffset = -1;
+ long lastWrittenTimestamp = -1;
+ for (ArrowBatchData batch : batches) {
+ long batchBaseOffset = batch.getBaseLogOffset();
+ if (batchBaseOffset >= stoppingOffset) {
+ batch.close();
+ continue;
+ }
+
+ long writableRowCount = stoppingOffset - batchBaseOffset;
+ int writableRows = (int) Math.min(batch.getRecordCount(), writableRowCount);
+ if (writableRows <= 0) {
+ batch.close();
+ continue;
+ }
+
+ ArrowBatchData batchToWrite = batch;
+ if (writableRows < batch.getRecordCount()) {
+ batchToWrite = batch.truncateAndTransferOwnership(writableRows);
+ }
+
+ if (batchWriter == null) {
+ batchWriter = (SupportsRecordBatchWrite) lakeWriterSupplier.get();
+ }
Review Comment:
batchWriter is obtained via an unchecked cast. If the LakeTieringFactory returns a LakeWriter that doesn’t implement SupportsRecordBatchWrite (e.g., custom PAIMON tiering implementation), this will fail with a ClassCastException. Prefer an instanceof check and throw an IOException/IllegalStateException with a clear message (or fall back to row-based writes).
##########
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java:
##########
@@ -421,6 +486,88 @@ private RecordsWithSplitIds<TableBucketWriteResult<WriteResult>> forLogRecords(
return new TableBucketWriteResultWithSplitIds(writeResults, finishedSplitIds);
}
+ /** Handles row-based ScanRecord writing for the log path. */
+ private LogOffsetAndTimestamp handleLogRecords(
+ List<ScanRecord> records,
+ SupplierWithException<LakeWriter<?>, IOException> lakeWriterSupplier,
+ long stoppingOffset)
+ throws IOException {
+ long lastWrittenOffset = -1;
+ long lastWrittenTimestamp = -1;
+ LakeWriter<?> lakeWriter = null;
+ for (ScanRecord record : records) {
+ if (record.logOffset() < stoppingOffset) {
+ if (lakeWriter == null) {
+ lakeWriter = lakeWriterSupplier.get();
+ }
+ lakeWriter.write(record);
+ lastWrittenOffset = record.logOffset();
+ lastWrittenTimestamp = record.timestamp();
+ if (record.getSizeInBytes() > 0) {
+ tieringMetrics.recordBytesRead(record.getSizeInBytes());
+ }
+ }
+ }
+ ScanRecord lastRecord = records.get(records.size() - 1);
+ return new LogOffsetAndTimestamp(
+ lastWrittenOffset, lastWrittenTimestamp, lastRecord.logOffset());
+ }
+
+ /** Handles Arrow batch writing for the record batch path. */
+ private LogOffsetAndTimestamp handleArrowBatchRecords(
+ List<ArrowBatchData> batches,
+ SupplierWithException<LakeWriter<?>, IOException> lakeWriterSupplier,
+ long stoppingOffset)
+ throws IOException {
+ SupportsRecordBatchWrite batchWriter = null;
+ long lastWrittenBatchEndOffset = -1;
+ long lastWrittenTimestamp = -1;
+ for (ArrowBatchData batch : batches) {
Review Comment:
handleArrowBatchRecords() later needs lastScannedOffset, but all ArrowBatchData instances are closed during iteration. Capture lastScannedOffset before the loop (while the last batch is still open) so the method doesn’t have to read VectorSchemaRoot state after close().
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyArrowBatchHelper.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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.lake.paimon.tiering.append;
+
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.record.ArrowBatchData;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.TimeStampMilliVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.TimeUnit;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.paimon.arrow.ArrowBundleRecords;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Helper class that encapsulates Arrow-dependent batch writing logic for append-only tables.
+ *
+ * <p>This class is separated from {@link AppendOnlyWriter} to avoid loading Arrow classes when
+ * Arrow is not on the classpath. It is lazily loaded only when Arrow batch writing is actually
+ * needed.
+ */
+class AppendOnlyArrowBatchHelper implements AutoCloseable {
+
+ private final FileStoreTable fileStoreTable;
+ private final TableWriteImpl<InternalRow> tableWrite;
+ private final RowType tableRowType;
+ private final int bucket;
+
+ // Child allocator for system column vectors, sharing the same root as the batch allocator
+ @Nullable private BufferAllocator systemColumnAllocator;
+
+ // Reusable resources for enriched VectorSchemaRoot with system columns
+ @Nullable private VectorSchemaRoot enrichedRoot;
+ @Nullable private Schema enrichedSchema;
+ @Nullable private IntVector bucketVector;
+ @Nullable private BigIntVector offsetVector;
+ @Nullable private TimeStampMilliVector timestampVector;
+
+ AppendOnlyArrowBatchHelper(
+ FileStoreTable fileStoreTable,
+ TableWriteImpl<InternalRow> tableWrite,
+ RowType tableRowType,
+ int bucket) {
+ this.fileStoreTable = fileStoreTable;
+ this.tableWrite = tableWrite;
+ this.tableRowType = tableRowType;
+ this.bucket = bucket;
+ }
+
+ /**
+ * Writes an Arrow batch directly to Paimon Parquet files. Enriches the VectorSchemaRoot with
+ * system columns (__bucket, __offset, __timestamp) and uses Paimon's {@link ArrowBundleRecords}
+ * for efficient batch writing.
+ */
+ void writeArrowBatch(ArrowBatchData arrowBatchData, BinaryRow partition) throws Exception {
+ int writtenBucket = bucket;
+ if (fileStoreTable.store().bucketMode() == BucketMode.BUCKET_UNAWARE) {
+ writtenBucket = 0;
+ }
+
+ VectorSchemaRoot originalRoot = arrowBatchData.getVectorSchemaRoot();
+ long baseOffset = arrowBatchData.getBaseLogOffset();
+ long timestamp = arrowBatchData.getTimestamp();
+ int rowCount = originalRoot.getRowCount();
+
+ ensureEnrichedRootInitialized(originalRoot, originalRoot.getVector(0).getAllocator());
+ updateEnrichedVectorSchemaRoot(writtenBucket, baseOffset, timestamp, rowCount);
+
+ ArrowBundleRecords arrowBundleRecords =
+ new ArrowBundleRecords(enrichedRoot, tableRowType, false);
+
+ tableWrite.writeBundle(partition, writtenBucket, arrowBundleRecords);
+ }
+
+ /**
+ * Ensures the enriched VectorSchemaRoot is initialized with system column vectors. Reuses
+ * system column vectors if schema matches. The enrichedRoot references the current
+ * originalRoot's data vectors plus the system column vectors.
+ */
+ private void ensureEnrichedRootInitialized(
+ VectorSchemaRoot originalRoot, BufferAllocator batchAllocator) {
+ Schema originalSchema = originalRoot.getSchema();
+ List<Field> originalFields = originalSchema.getFields();
+ int currentFieldCount = originalFields.size();
+
+ // initialize system column vectors on first call, using a child allocator that
+ // shares the same root as the batch allocator so all vectors are compatible
+ if (bucketVector == null) {
+ Field bucketField =
+ new Field(
+ TableDescriptor.BUCKET_COLUMN_NAME,
+ new FieldType(false, new ArrowType.Int(32, true), null),
+ null);
+ Field offsetField =
+ new Field(
+ TableDescriptor.OFFSET_COLUMN_NAME,
+ new FieldType(false, new ArrowType.Int(64, true), null),
+ null);
+ Field timestampField =
+ new Field(
+ TableDescriptor.TIMESTAMP_COLUMN_NAME,
+ new FieldType(
+ false,
+ new ArrowType.Timestamp(TimeUnit.MILLISECOND, null),
+ null),
+ null);
+
+ List<Field> enrichedFields = new ArrayList<>(originalFields);
+ enrichedFields.add(bucketField);
+ enrichedFields.add(offsetField);
+ enrichedFields.add(timestampField);
+ enrichedSchema = new Schema(enrichedFields);
+
+ if (systemColumnAllocator == null) {
+ systemColumnAllocator =
+ batchAllocator
+ .getRoot()
+ .newChildAllocator("system-column-allocator", 0, Long.MAX_VALUE);
+ }
+ bucketVector = new IntVector(bucketField, systemColumnAllocator);
+ offsetVector = new BigIntVector(offsetField, systemColumnAllocator);
+ timestampVector = new TimeStampMilliVector(timestampField, systemColumnAllocator);
+ }
+
+ // recreate enrichedRoot to reference the current originalRoot's data vectors
+ List<FieldVector> allVectors = new ArrayList<>();
+ for (int i = 0; i < currentFieldCount; i++) {
+ allVectors.add(originalRoot.getVector(i));
+ }
+ allVectors.add(bucketVector);
+ allVectors.add(offsetVector);
+ allVectors.add(timestampVector);
+
+ enrichedRoot = new VectorSchemaRoot(enrichedSchema, allVectors, originalRoot.getRowCount());
+ }
Review Comment:
ensureEnrichedRootInitialized() only builds enrichedSchema on the first batch (when bucketVector == null). If the Arrow batch schema evolves (e.g., ALTER TABLE adds/drops columns), currentFieldCount changes but enrichedSchema remains the old field list, and new VectorSchemaRoot(enrichedSchema, allVectors, ...) can fail due to field/vector count mismatch or wrong field mapping. Rebuild enrichedSchema when originalFields change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/paimon] TieringSourceReader adjust to scan arrow record batch and write arrow record batch to lake [fluss]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on code in PR #3430:
URL: https://github.com/apache/fluss/pull/3430#discussion_r3353218839
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/tiering/append/AppendOnlyArrowBatchHelper.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * 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.lake.paimon.tiering.append;
+
+import org.apache.fluss.metadata.TableDescriptor;
+import org.apache.fluss.record.ArrowBatchData;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.vector.BigIntVector;
+import org.apache.arrow.vector.FieldVector;
+import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.TimeStampMilliVector;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.TimeUnit;
+import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.apache.paimon.arrow.ArrowBundleRecords;
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.table.BucketMode;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Helper class that encapsulates Arrow-dependent batch writing logic for append-only tables.
+ *
+ * <p>This class is separated from {@link AppendOnlyWriter} to avoid loading Arrow classes when
+ * Arrow is not on the classpath. It is lazily loaded only when Arrow batch writing is actually
+ * needed.
+ */
+class AppendOnlyArrowBatchHelper implements AutoCloseable {
+
+ private final FileStoreTable fileStoreTable;
+ private final TableWriteImpl<InternalRow> tableWrite;
+ private final RowType tableRowType;
+ private final int bucket;
+
+ // Child allocator for system column vectors, sharing the same root as the batch allocator
+ @Nullable private BufferAllocator systemColumnAllocator;
+
+ // Reusable resources for enriched VectorSchemaRoot with system columns
+ @Nullable private VectorSchemaRoot enrichedRoot;
+ @Nullable private Schema enrichedSchema;
+ @Nullable private IntVector bucketVector;
+ @Nullable private BigIntVector offsetVector;
+ @Nullable private TimeStampMilliVector timestampVector;
+
+ AppendOnlyArrowBatchHelper(
+ FileStoreTable fileStoreTable,
+ TableWriteImpl<InternalRow> tableWrite,
+ RowType tableRowType,
+ int bucket) {
+ this.fileStoreTable = fileStoreTable;
+ this.tableWrite = tableWrite;
+ this.tableRowType = tableRowType;
+ this.bucket = bucket;
+ }
+
+ /**
+ * Writes an Arrow batch directly to Paimon Parquet files. Enriches the VectorSchemaRoot with
+ * system columns (__bucket, __offset, __timestamp) and uses Paimon's {@link ArrowBundleRecords}
+ * for efficient batch writing.
+ */
+ void writeArrowBatch(ArrowBatchData arrowBatchData, BinaryRow partition) throws Exception {
+ int writtenBucket = bucket;
+ if (fileStoreTable.store().bucketMode() == BucketMode.BUCKET_UNAWARE) {
+ writtenBucket = 0;
+ }
+
+ VectorSchemaRoot originalRoot = arrowBatchData.getVectorSchemaRoot();
+ long baseOffset = arrowBatchData.getBaseLogOffset();
+ long timestamp = arrowBatchData.getTimestamp();
+ int rowCount = originalRoot.getRowCount();
+
+ ensureEnrichedRootInitialized(originalRoot, originalRoot.getVector(0).getAllocator());
+ updateEnrichedVectorSchemaRoot(writtenBucket, baseOffset, timestamp, rowCount);
+
+ ArrowBundleRecords arrowBundleRecords =
+ new ArrowBundleRecords(enrichedRoot, tableRowType, false);
+
+ tableWrite.writeBundle(partition, writtenBucket, arrowBundleRecords);
+ }
+
+ /**
+ * Ensures the enriched VectorSchemaRoot is initialized with system column vectors. Reuses
+ * system column vectors if schema matches. The enrichedRoot references the current
+ * originalRoot's data vectors plus the system column vectors.
+ */
+ private void ensureEnrichedRootInitialized(
+ VectorSchemaRoot originalRoot, BufferAllocator batchAllocator) {
+ Schema originalSchema = originalRoot.getSchema();
+ List<Field> originalFields = originalSchema.getFields();
+ int currentFieldCount = originalFields.size();
+
+ // initialize system column vectors on first call, using a child allocator that
+ // shares the same root as the batch allocator so all vectors are compatible
+ if (bucketVector == null) {
+ Field bucketField =
+ new Field(
+ TableDescriptor.BUCKET_COLUMN_NAME,
+ new FieldType(false, new ArrowType.Int(32, true), null),
+ null);
+ Field offsetField =
+ new Field(
+ TableDescriptor.OFFSET_COLUMN_NAME,
+ new FieldType(false, new ArrowType.Int(64, true), null),
+ null);
+ Field timestampField =
+ new Field(
+ TableDescriptor.TIMESTAMP_COLUMN_NAME,
+ new FieldType(
+ false,
+ new ArrowType.Timestamp(TimeUnit.MILLISECOND, null),
+ null),
+ null);
+
+ List<Field> enrichedFields = new ArrayList<>(originalFields);
+ enrichedFields.add(bucketField);
+ enrichedFields.add(offsetField);
+ enrichedFields.add(timestampField);
+ enrichedSchema = new Schema(enrichedFields);
+
+ if (systemColumnAllocator == null) {
+ systemColumnAllocator =
+ batchAllocator
+ .getRoot()
+ .newChildAllocator("system-column-allocator", 0, Long.MAX_VALUE);
+ }
+ bucketVector = new IntVector(bucketField, systemColumnAllocator);
+ offsetVector = new BigIntVector(offsetField, systemColumnAllocator);
+ timestampVector = new TimeStampMilliVector(timestampField, systemColumnAllocator);
+ }
+
+ // recreate enrichedRoot to reference the current originalRoot's data vectors
+ List<FieldVector> allVectors = new ArrayList<>();
+ for (int i = 0; i < currentFieldCount; i++) {
+ allVectors.add(originalRoot.getVector(i));
+ }
+ allVectors.add(bucketVector);
+ allVectors.add(offsetVector);
+ allVectors.add(timestampVector);
+
+ enrichedRoot = new VectorSchemaRoot(enrichedSchema, allVectors, originalRoot.getRowCount());
+ }
Review Comment:
1. The Arrow scan layer always returns ArrowBatchData with the latest table schema — old data that was written before the schema evolution gets null-padded for the newly added columns during scan. So all batches seen by the same helper instance share the same schema.
2. The AppendOnlyArrowBatchHelper is constructed with a fixed tableRowType (from FileStoreTable.rowType()). If the schema evolves mid-tiering, we'd need a new helper with the updated tableRowType anyway, because ArrowBundleRecords(enrichedRoot, tableRowType, ...) requires the enrichedRoot's field count to match the tableRowType.
Rebuilding enrichedSchema alone wouldn't be sufficient — the tableRowType mismatch would still cause Paimon to fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org