hudi-agent commented on code in PR #19211: URL: https://github.com/apache/hudi/pull/19211#discussion_r3770895322
########## hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/BlobMaterializingIterator.java: ########## @@ -0,0 +1,283 @@ +/* + * 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.hudi.source; + +import org.apache.hudi.common.blob.BlobRangeReader; +import org.apache.hudi.common.blob.BlobReadRequest; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.util.collection.ClosableIterator; +import org.apache.hudi.storage.HoodieStorage; + +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.Objects; + +/** + * A {@link ClosableIterator}{@code <RowData>} decorator that materializes out-of-line (OOL) + * BLOB fields in batches. + * + * <h3>Algorithm</h3> + * <ol> + * <li>Buffer up to {@code lookaheadSize} rows from the inner iterator (shallow-copied to + * prevent row-buffer aliasing in upstream iterators). + * <li>For every BLOB field in each row, check the {@code type} discriminator: + * {@code INLINE} passes through; {@code OUT_OF_LINE} is collected as a + * {@link BlobReadRequest} tagged with a {@code (rowIndex, blobFieldPosition)} key. + * <li>Delegate all I/O to {@link BlobRangeReader#readBatched}: it groups requests by file + * path, merges nearby ranges within {@code maxGapBytes}, and issues one + * seek+readFully per merged range. + * <li>Replace each materialized blob struct with {@code (INLINE, <bytes>, null)} and emit + * rows in their original order. + * </ol> + * + * <p>Rows whose BLOB field is {@code null} or already {@code INLINE} are passed through unchanged. + */ +public class BlobMaterializingIterator implements ClosableIterator<RowData> { + + private final ClosableIterator<RowData> inner; + private final HoodieStorage storage; + private final int maxGapBytes; + private final int lookaheadSize; + + // Primitive int[] follows the same convention as CopyOnWriteInputFormat.selectedFields and + // RowDataProjection.positions — avoids Integer boxing in a per-row hot path. + private final int[] blobFieldPositions; + private final RowData.FieldGetter[] fieldGetters; + private final int numFields; + + private final Deque<RowData> outputQueue = new ArrayDeque<>(); + + /** + * @param inner raw iterator from the file group reader + * @param requiredRowType Flink RowType matching the rows emitted by {@code inner} + * @param blobFieldPositions field indices in {@code requiredRowType} that are BLOB structs + * @param storage HoodieStorage instance for DFS I/O + * @param maxGapBytes max byte gap between two OOL ranges in the same file to merge + * @param lookaheadSize rows to buffer per batch + */ + public BlobMaterializingIterator( + ClosableIterator<RowData> inner, + RowType requiredRowType, + int[] blobFieldPositions, + HoodieStorage storage, + int maxGapBytes, + int lookaheadSize) { + this.inner = inner; + this.storage = storage; + this.maxGapBytes = maxGapBytes; + this.lookaheadSize = lookaheadSize; + this.blobFieldPositions = blobFieldPositions; + this.numFields = requiredRowType.getFieldCount(); + this.fieldGetters = buildFieldGetters(requiredRowType); + } + + @Override + public boolean hasNext() { + if (outputQueue.isEmpty()) { + fillBatch(); + } + return !outputQueue.isEmpty(); + } + + @Override + public RowData next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + return outputQueue.poll(); + } + + @Override + public void close() { + inner.close(); + } + + // ------------------------------------------------------------------------- + // Batch fill + // ------------------------------------------------------------------------- + + private void fillBatch() { + // Collect up to lookaheadSize rows; shallow-copy to guard against row-buffer reuse. + List<RowData> lookaheadBuffer = new ArrayList<>(lookaheadSize); + while (lookaheadBuffer.size() < lookaheadSize && inner.hasNext()) { + lookaheadBuffer.add(shallowCopyRow(inner.next())); + } + if (lookaheadBuffer.isEmpty()) { + return; + } + + // Build BlobReadRequests for every OOL blob field in the lookahead buffer. + List<BlobReadRequest<RowFieldKey>> readRequests = collectReadRequests(lookaheadBuffer); + + // Delegate all I/O to the common utility. + Map<RowFieldKey, byte[]> dataMap = BlobRangeReader.readBatched(readRequests, storage, maxGapBytes); + + // Reconstruct rows, replacing OOL structs with INLINE ones; pass others through unchanged. + for (int rowIndex = 0; rowIndex < lookaheadBuffer.size(); rowIndex++) { + RowData row = lookaheadBuffer.get(rowIndex); + if (!hasMaterializedBlob(rowIndex, dataMap)) { + outputQueue.add(row); + continue; + } + GenericRowData newRow = new GenericRowData(row.getRowKind(), numFields); + for (int i = 0; i < numFields; i++) { + newRow.setField(i, fieldGetters[i].getFieldOrNull(row)); + } + for (int blobFieldPosition : blobFieldPositions) { + byte[] data = dataMap.get(new RowFieldKey(rowIndex, blobFieldPosition)); + if (data != null) { + newRow.setField(blobFieldPosition, buildInlineBlob(data)); + } + } + outputQueue.add(newRow); + } + } + + private boolean hasMaterializedBlob(int rowIndex, Map<RowFieldKey, byte[]> dataMap) { + for (int blobFieldPosition : blobFieldPositions) { + if (dataMap.containsKey(new RowFieldKey(rowIndex, blobFieldPosition))) { + return true; + } + } + return false; + } + + private List<BlobReadRequest<RowFieldKey>> collectReadRequests(List<RowData> lookaheadBuffer) { + List<BlobReadRequest<RowFieldKey>> requests = new ArrayList<>(); + for (int rowIndex = 0; rowIndex < lookaheadBuffer.size(); rowIndex++) { + RowData row = lookaheadBuffer.get(rowIndex); + for (int blobFieldPosition : blobFieldPositions) { + if (row.isNullAt(blobFieldPosition)) { + continue; + } + RowData blobRow = row.getRow(blobFieldPosition, HoodieSchema.Blob.getFieldCount()); + if (blobRow == null || blobRow.isNullAt(0)) { Review Comment: 🤖 The blob positions come from `schema().isBlobField()`, which also returns true for `ARRAY<BLOB>` and `MAP<..., BLOB>` columns (see `HoodieSchema.isBlobField`). But here we unconditionally do `row.getRow(blobFieldPosition, 3)`, assuming a scalar blob struct. For an array/map-of-blob column that position holds ArrayData/MapData, so this would throw a ClassCastException (or misread) once `read.blob.materialize` is on. Could we restrict detection to scalar BLOB fields, or handle the array/map cases explicitly? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/BlobMaterializingIterator.java: ########## @@ -0,0 +1,283 @@ +/* + * 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.hudi.source; + +import org.apache.hudi.common.blob.BlobRangeReader; +import org.apache.hudi.common.blob.BlobReadRequest; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.util.collection.ClosableIterator; +import org.apache.hudi.storage.HoodieStorage; + +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.Objects; + +/** + * A {@link ClosableIterator}{@code <RowData>} decorator that materializes out-of-line (OOL) + * BLOB fields in batches. + * + * <h3>Algorithm</h3> + * <ol> + * <li>Buffer up to {@code lookaheadSize} rows from the inner iterator (shallow-copied to + * prevent row-buffer aliasing in upstream iterators). + * <li>For every BLOB field in each row, check the {@code type} discriminator: + * {@code INLINE} passes through; {@code OUT_OF_LINE} is collected as a + * {@link BlobReadRequest} tagged with a {@code (rowIndex, blobFieldPosition)} key. + * <li>Delegate all I/O to {@link BlobRangeReader#readBatched}: it groups requests by file + * path, merges nearby ranges within {@code maxGapBytes}, and issues one + * seek+readFully per merged range. + * <li>Replace each materialized blob struct with {@code (INLINE, <bytes>, null)} and emit + * rows in their original order. + * </ol> + * + * <p>Rows whose BLOB field is {@code null} or already {@code INLINE} are passed through unchanged. + */ +public class BlobMaterializingIterator implements ClosableIterator<RowData> { + + private final ClosableIterator<RowData> inner; + private final HoodieStorage storage; + private final int maxGapBytes; + private final int lookaheadSize; + + // Primitive int[] follows the same convention as CopyOnWriteInputFormat.selectedFields and + // RowDataProjection.positions — avoids Integer boxing in a per-row hot path. + private final int[] blobFieldPositions; + private final RowData.FieldGetter[] fieldGetters; + private final int numFields; + + private final Deque<RowData> outputQueue = new ArrayDeque<>(); + + /** + * @param inner raw iterator from the file group reader + * @param requiredRowType Flink RowType matching the rows emitted by {@code inner} + * @param blobFieldPositions field indices in {@code requiredRowType} that are BLOB structs + * @param storage HoodieStorage instance for DFS I/O + * @param maxGapBytes max byte gap between two OOL ranges in the same file to merge + * @param lookaheadSize rows to buffer per batch + */ + public BlobMaterializingIterator( + ClosableIterator<RowData> inner, + RowType requiredRowType, + int[] blobFieldPositions, + HoodieStorage storage, + int maxGapBytes, + int lookaheadSize) { + this.inner = inner; + this.storage = storage; + this.maxGapBytes = maxGapBytes; + this.lookaheadSize = lookaheadSize; + this.blobFieldPositions = blobFieldPositions; + this.numFields = requiredRowType.getFieldCount(); + this.fieldGetters = buildFieldGetters(requiredRowType); + } + + @Override + public boolean hasNext() { + if (outputQueue.isEmpty()) { + fillBatch(); + } + return !outputQueue.isEmpty(); + } + + @Override + public RowData next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + return outputQueue.poll(); + } + + @Override + public void close() { + inner.close(); + } + + // ------------------------------------------------------------------------- + // Batch fill + // ------------------------------------------------------------------------- + + private void fillBatch() { + // Collect up to lookaheadSize rows; shallow-copy to guard against row-buffer reuse. + List<RowData> lookaheadBuffer = new ArrayList<>(lookaheadSize); + while (lookaheadBuffer.size() < lookaheadSize && inner.hasNext()) { + lookaheadBuffer.add(shallowCopyRow(inner.next())); + } + if (lookaheadBuffer.isEmpty()) { + return; + } + + // Build BlobReadRequests for every OOL blob field in the lookahead buffer. + List<BlobReadRequest<RowFieldKey>> readRequests = collectReadRequests(lookaheadBuffer); + + // Delegate all I/O to the common utility. + Map<RowFieldKey, byte[]> dataMap = BlobRangeReader.readBatched(readRequests, storage, maxGapBytes); + + // Reconstruct rows, replacing OOL structs with INLINE ones; pass others through unchanged. + for (int rowIndex = 0; rowIndex < lookaheadBuffer.size(); rowIndex++) { + RowData row = lookaheadBuffer.get(rowIndex); + if (!hasMaterializedBlob(rowIndex, dataMap)) { + outputQueue.add(row); + continue; + } + GenericRowData newRow = new GenericRowData(row.getRowKind(), numFields); + for (int i = 0; i < numFields; i++) { + newRow.setField(i, fieldGetters[i].getFieldOrNull(row)); + } + for (int blobFieldPosition : blobFieldPositions) { + byte[] data = dataMap.get(new RowFieldKey(rowIndex, blobFieldPosition)); + if (data != null) { + newRow.setField(blobFieldPosition, buildInlineBlob(data)); + } + } + outputQueue.add(newRow); + } + } + + private boolean hasMaterializedBlob(int rowIndex, Map<RowFieldKey, byte[]> dataMap) { + for (int blobFieldPosition : blobFieldPositions) { + if (dataMap.containsKey(new RowFieldKey(rowIndex, blobFieldPosition))) { + return true; + } + } + return false; + } + + private List<BlobReadRequest<RowFieldKey>> collectReadRequests(List<RowData> lookaheadBuffer) { + List<BlobReadRequest<RowFieldKey>> requests = new ArrayList<>(); + for (int rowIndex = 0; rowIndex < lookaheadBuffer.size(); rowIndex++) { + RowData row = lookaheadBuffer.get(rowIndex); + for (int blobFieldPosition : blobFieldPositions) { + if (row.isNullAt(blobFieldPosition)) { + continue; + } + RowData blobRow = row.getRow(blobFieldPosition, HoodieSchema.Blob.getFieldCount()); + if (blobRow == null || blobRow.isNullAt(0)) { + continue; + } + String storageType = blobRow.getString(0).toString(); + if (!HoodieSchema.Blob.OUT_OF_LINE.equals(storageType)) { + continue; + } + if (blobRow.isNullAt(2)) { + throw new IllegalStateException( + "OUT_OF_LINE blob is missing its reference struct at row " + rowIndex); + } + RowData refRow = blobRow.getRow(2, HoodieSchema.Blob.getReferenceFieldCount()); + String externalPath = refRow.getString(0).toString(); + boolean noOffset = refRow.isNullAt(1); + boolean noLength = refRow.isNullAt(2); + RowFieldKey key = new RowFieldKey(rowIndex, blobFieldPosition); + if (noOffset && noLength) { + requests.add(BlobReadRequest.wholeFile(externalPath, key)); + } else if (noOffset || noLength) { + throw new IllegalArgumentException( Review Comment: 🤖 The blob reference schema documents `offset` and `length` as independently optional — offset absent means 0, length absent means 'the rest of the file starting from offset' (see the field comments on HoodieSchema.Blob). This branch instead rejects any reference that sets exactly one of them. Should we honor the documented partial-specification semantics (offset-only → [offset, EOF), length-only → [0, length)) rather than throwing on otherwise-valid data? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/BlobMaterializingIterator.java: ########## @@ -0,0 +1,283 @@ +/* + * 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.hudi.source; + +import org.apache.hudi.common.blob.BlobRangeReader; +import org.apache.hudi.common.blob.BlobReadRequest; +import org.apache.hudi.common.schema.HoodieSchema; +import org.apache.hudi.common.util.collection.ClosableIterator; +import org.apache.hudi.storage.HoodieStorage; + +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; + +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Deque; +import java.util.List; +import java.util.Map; +import java.util.NoSuchElementException; +import java.util.Objects; + +/** + * A {@link ClosableIterator}{@code <RowData>} decorator that materializes out-of-line (OOL) + * BLOB fields in batches. + * + * <h3>Algorithm</h3> + * <ol> + * <li>Buffer up to {@code lookaheadSize} rows from the inner iterator (shallow-copied to + * prevent row-buffer aliasing in upstream iterators). + * <li>For every BLOB field in each row, check the {@code type} discriminator: + * {@code INLINE} passes through; {@code OUT_OF_LINE} is collected as a + * {@link BlobReadRequest} tagged with a {@code (rowIndex, blobFieldPosition)} key. + * <li>Delegate all I/O to {@link BlobRangeReader#readBatched}: it groups requests by file + * path, merges nearby ranges within {@code maxGapBytes}, and issues one + * seek+readFully per merged range. + * <li>Replace each materialized blob struct with {@code (INLINE, <bytes>, null)} and emit + * rows in their original order. + * </ol> + * + * <p>Rows whose BLOB field is {@code null} or already {@code INLINE} are passed through unchanged. + */ +public class BlobMaterializingIterator implements ClosableIterator<RowData> { + + private final ClosableIterator<RowData> inner; + private final HoodieStorage storage; + private final int maxGapBytes; + private final int lookaheadSize; + + // Primitive int[] follows the same convention as CopyOnWriteInputFormat.selectedFields and + // RowDataProjection.positions — avoids Integer boxing in a per-row hot path. + private final int[] blobFieldPositions; + private final RowData.FieldGetter[] fieldGetters; + private final int numFields; + + private final Deque<RowData> outputQueue = new ArrayDeque<>(); + + /** + * @param inner raw iterator from the file group reader + * @param requiredRowType Flink RowType matching the rows emitted by {@code inner} + * @param blobFieldPositions field indices in {@code requiredRowType} that are BLOB structs + * @param storage HoodieStorage instance for DFS I/O + * @param maxGapBytes max byte gap between two OOL ranges in the same file to merge + * @param lookaheadSize rows to buffer per batch + */ + public BlobMaterializingIterator( + ClosableIterator<RowData> inner, + RowType requiredRowType, + int[] blobFieldPositions, + HoodieStorage storage, + int maxGapBytes, + int lookaheadSize) { + this.inner = inner; + this.storage = storage; + this.maxGapBytes = maxGapBytes; + this.lookaheadSize = lookaheadSize; + this.blobFieldPositions = blobFieldPositions; + this.numFields = requiredRowType.getFieldCount(); + this.fieldGetters = buildFieldGetters(requiredRowType); + } + + @Override + public boolean hasNext() { + if (outputQueue.isEmpty()) { + fillBatch(); + } + return !outputQueue.isEmpty(); + } + + @Override + public RowData next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + return outputQueue.poll(); + } + + @Override + public void close() { + inner.close(); + } + + // ------------------------------------------------------------------------- + // Batch fill + // ------------------------------------------------------------------------- + + private void fillBatch() { + // Collect up to lookaheadSize rows; shallow-copy to guard against row-buffer reuse. + List<RowData> lookaheadBuffer = new ArrayList<>(lookaheadSize); + while (lookaheadBuffer.size() < lookaheadSize && inner.hasNext()) { + lookaheadBuffer.add(shallowCopyRow(inner.next())); Review Comment: 🤖 If `read.blob.batching.lookahead.size` is set to 0 (or negative), this loop never buffers a row, so `fillBatch()` returns with an empty buffer and `hasNext()` reports false — every row is silently dropped with no error. Could we validate/clamp the lookahead to >= 1 so a misconfiguration can't cause silent data loss? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-common/src/main/java/org/apache/hudi/common/blob/BlobRangeReader.java: ########## @@ -0,0 +1,264 @@ +/* + * 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.hudi.common.blob; + +import org.apache.hudi.io.SeekableDataInputStream; +import org.apache.hudi.storage.HoodieStorage; +import org.apache.hudi.storage.StoragePath; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +/** + * Engine-agnostic utility for batched byte-range reads of out-of-line (OOL) BLOB references. + * + * <h3>Algorithm</h3> + * <ol> + * <li>Separate whole-file requests from range requests. + * <li>Group range requests by {@code filePath}, sort each group by {@code offset}, and merge + * adjacent or nearby ranges whose gap is ≤ {@code maxGapBytes} into a single + * {@link MergedRange}. + * <li>For each {@link MergedRange}: seek to {@code startOffset} and {@code readFully} the + * entire merged span in one I/O call; slice the buffer to extract each individual request's + * bytes. + * <li>For whole-file requests: open the file and read all bytes. + * </ol> + * + * <p>The caller provides opaque {@code tag} values on each {@link BlobReadRequest}; those tags are + * used as keys in the returned map so results can be correlated back to the caller's domain + * (e.g. row index + field position in Flink). Spark's {@code BatchedBlobReader} solves the same + * problem by carrying a {@code RowInfo} through {@code MergedRange} instead of a separate tag. + * + * <h3>Overlap detection</h3> + * If two requests for the same file have overlapping byte ranges an + * {@link IllegalArgumentException} is thrown (same check exists in Spark's + * {@code BatchedBlobReader.mergeRanges}). Hudi's OOL blob writer assigns disjoint + * {@code (offset, length)} spans when appending to an external file, so well-formed table data + * should never produce overlaps; this guard catches corruption, manual edits, or buggy callers. + * Adjacent or gapped ranges within {@code maxGapBytes} are merged; truly overlapping ranges are not. + */ +public final class BlobRangeReader { + + private static final Logger LOG = LoggerFactory.getLogger(BlobRangeReader.class); + + /** + * Read chunk size for whole-file streaming. Spark's {@code BatchedBlobReader.readWholeFile} + * uses {@code InputStream.readAllBytes()} (Java 9+); hudi-common targets Java 8, so we stream + * into a {@link ByteArrayOutputStream} with a standard 8 KiB buffer instead. + */ + private static final int WHOLE_FILE_READ_BUFFER_SIZE = 8192; + + private BlobRangeReader() { + } + + // ------------------------------------------------------------------------- + // Public API + // ------------------------------------------------------------------------- + + /** + * Issues batched DFS reads for all {@code requests} and returns a map from each request's + * {@code tag} to the bytes that were read. + * + * @param requests OOL blob read requests (may contain a mix of range and whole-file refs) + * @param storage HoodieStorage instance for file I/O + * @param maxGapBytes maximum byte gap between two range requests in the same file that are + * still coalesced into a single read (0 = only truly adjacent ranges are + * merged) + * @param <T> caller-supplied correlation tag type + * @return map from {@link BlobReadRequest#tag} to the bytes read for that request + */ + public static <T> Map<T, byte[]> readBatched( + List<BlobReadRequest<T>> requests, + HoodieStorage storage, + int maxGapBytes) { + if (requests == null || requests.isEmpty()) { + return Collections.emptyMap(); + } + + List<BlobReadRequest<T>> wholeFileReqs = new ArrayList<>(); + List<BlobReadRequest<T>> rangeReqs = new ArrayList<>(); + for (BlobReadRequest<T> req : requests) { + (req.isWholeFile() ? wholeFileReqs : rangeReqs).add(req); + } + + return Stream.concat( + readWholeFiles(wholeFileReqs, storage).entrySet().stream(), + readRanges(rangeReqs, storage, maxGapBytes).entrySet().stream()) + .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); + } + + // ------------------------------------------------------------------------- + // Package-private: visible for testing + // ------------------------------------------------------------------------- + + /** + * Groups {@code requests} by file path, sorts each group by offset, and merges nearby ranges. + * + * <p>Cross-file processing order does not affect read correctness — each file is opened and + * read independently. Only within-file offset order matters for range merging. We use a + * {@link LinkedHashMap} so iteration follows first-seen file path in the request list, giving + * stable I/O ordering for tests and debug logs. Sorting file paths (e.g. via {@code TreeMap}) + * would also be stable but is unnecessary for correctness. + */ + static <T> List<MergedRange<T>> groupSortMerge( + List<BlobReadRequest<T>> requests, + int maxGapBytes) { + Map<String, List<BlobReadRequest<T>>> byFile = new LinkedHashMap<>(); + for (BlobReadRequest<T> req : requests) { + byFile.computeIfAbsent(req.filePath, k -> new ArrayList<>()).add(req); + } + + List<MergedRange<T>> merged = new ArrayList<>(); + for (Map.Entry<String, List<BlobReadRequest<T>>> entry : byFile.entrySet()) { + List<BlobReadRequest<T>> fileReqs = entry.getValue(); + fileReqs.sort((a, b) -> Long.compare(a.offset, b.offset)); + merged.addAll(mergeWithinFile(fileReqs, maxGapBytes)); + } + return merged; + } + + // ------------------------------------------------------------------------- + // Private helpers + // ------------------------------------------------------------------------- + + private static <T> Map<T, byte[]> readWholeFiles( + List<BlobReadRequest<T>> reqs, + HoodieStorage storage) { + Map<T, byte[]> result = new HashMap<>(); + for (BlobReadRequest<T> req : reqs) { + try (InputStream in = storage.open(new StoragePath(req.filePath))) { + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + byte[] buf = new byte[WHOLE_FILE_READ_BUFFER_SIZE]; + int n; + while ((n = in.read(buf)) != -1) { + baos.write(buf, 0, n); + } + result.put(req.tag, baos.toByteArray()); + LOG.debug("Read whole file {} for tag {}", req.filePath, req.tag); + } catch (IOException e) { + throw new RuntimeException("Failed to read whole-file blob: " + req.filePath, e); + } + } + return result; + } + + private static <T> Map<T, byte[]> readRanges( + List<BlobReadRequest<T>> rangeReqs, + HoodieStorage storage, + int maxGapBytes) { + if (rangeReqs.isEmpty()) { + return Collections.emptyMap(); + } + Map<T, byte[]> result = new HashMap<>(); + for (MergedRange<T> range : groupSortMerge(rangeReqs, maxGapBytes)) { + try (SeekableDataInputStream in = + storage.openSeekable(new StoragePath(range.filePath), false)) { + in.seek(range.startOffset); + int totalLength = (int) (range.endOffset - range.startOffset); + byte[] buffer = new byte[totalLength]; + in.readFully(buffer, 0, totalLength); + LOG.debug("Read {} bytes from {} at offset {} for {} request(s)", + totalLength, range.filePath, range.startOffset, range.requests.size()); + for (BlobReadRequest<T> req : range.requests) { + int relOff = (int) (req.offset - range.startOffset); + result.put(req.tag, Arrays.copyOfRange(buffer, relOff, relOff + (int) req.length)); + } + } catch (IOException e) { + throw new RuntimeException( + "Failed to read batched blob ranges from: " + range.filePath, e); + } + } + return result; + } + + /** + * Merges a sorted (by offset) list of range requests from the same file into + * {@link MergedRange}s. Requests whose gap is ≤ {@code maxGapBytes} are combined into a single + * range. Truly overlapping ranges (gap < 0) are rejected. + * + * @throws IllegalArgumentException if two requests have overlapping byte ranges + */ + private static <T> List<MergedRange<T>> mergeWithinFile( + List<BlobReadRequest<T>> sortedReqs, + int maxGapBytes) { + List<MergedRange<T>> merged = new ArrayList<>(); + MergedRange<T> current = null; + for (BlobReadRequest<T> req : sortedReqs) { + if (current == null) { + current = new MergedRange<>(req.filePath, req.offset, req.offset + req.length); + current.requests.add(req); + } else { + long gap = req.offset - current.endOffset; Review Comment: 🤖 This throws on any two ranges in the same file whose spans touch/overlap (gap < 0). Are we certain two different rows can never reference the same or an overlapping byte range in one external file — e.g. shared/copied blob references via INSERT ... SELECT, or an unchanged blob carried across an update? If that can happen, a valid read within one lookahead window would fail here with IllegalArgumentException. The disjoint-span guarantee holds within a single writer append, but the Flink lookahead groups references across rows, so it's worth confirming it holds there too. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieSplitReaderFunction.java: ########## @@ -91,6 +98,45 @@ protected ClosableIterator<RowData> createRecordIterator(HoodieSourceSplit split } } + /** + * Wraps the raw iterator with {@link BlobMaterializingIterator} when + * {@link FlinkOptions#BLOB_READ_MATERIALIZE} is enabled and the required schema contains + * at least one BLOB field. + */ + private ClosableIterator<RowData> maybeMaterialize(ClosableIterator<RowData> iter) { + if (!conf.get(FlinkOptions.BLOB_READ_MATERIALIZE)) { + return iter; + } + int[] blobPositions = detectBlobFieldPositions(requiredSchema); + if (blobPositions.length == 0) { + return iter; + } + RowType requiredRowType = HoodieSchemaConverter.convertToRowType(requiredSchema); + HoodieStorage storage = HoodieStorageUtils.getStorage( + conf.get(FlinkOptions.PATH), HadoopFSUtils.getStorageConf(getHadoopConf())); + return new BlobMaterializingIterator( + iter, + requiredRowType, + blobPositions, + storage, + conf.get(FlinkOptions.BLOB_BATCHING_MAX_GAP_BYTES), + conf.get(FlinkOptions.BLOB_BATCHING_LOOKAHEAD_SIZE)); + } + + /** + * Returns the field positions in {@code schema} whose type is a BLOB field. + */ Review Comment: 🤖 nit: `detectBlobFieldPositions` is identical to the copy in `MergeOnReadInputFormat` — could you move it to a shared utility (e.g. a new `BlobSchemaUtils` helper or an existing utils class) so future changes don't need to be applied in two places? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/format/cow/CopyOnWriteInputFormat.java: ########## @@ -154,6 +173,18 @@ public void open(FileInputSplit fileSplit) throws IOException { predicates); Review Comment: 🤖 nit: the `flinkConf != null` guard is surprising since the field is always set in the constructor — could you default to an empty `Configuration()` when the caller passes null (or make the parameter non-nullable) to avoid this defensive check in `open()`? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- 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]
