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