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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
