diff --git a/java/core/src/java/org/apache/orc/impl/RecordReaderUtils.java b/java/core/src/java/org/apache/orc/impl/RecordReaderUtils.java index 2c73970688..cc1eb6fb79 100644 --- a/java/core/src/java/org/apache/orc/impl/RecordReaderUtils.java +++ b/java/core/src/java/org/apache/orc/impl/RecordReaderUtils.java @@ -35,11 +35,11 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; -import java.util.HashMap; import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.TreeMap; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; import java.util.function.IntFunction; import java.util.function.Supplier; @@ -117,7 +117,8 @@ public BufferChunkList readFileData(BufferChunkList range, boolean doForceDirect ) throws IOException { if (supportVectoredIO && zcr == null) { - RecordReaderUtils.readDiskRangesVectored(file, range, doForceDirect); + RecordReaderUtils.readDiskRangesVectored(file, range, doForceDirect, minSeekSize, + minSeekSizeTolerance); } else { RecordReaderUtils.readDiskRanges(file, zcr, range, doForceDirect, minSeekSize, minSeekSizeTolerance); @@ -571,36 +572,394 @@ private static void readDiskRanges(FSDataInputStream file, } /** - * Read the list of ranges from the file by updating each range in the list + * Maximum length of any single coalesced {@link FileRange} we are willing to + * emit. Mirrors {@link #MAX_ARRAY_SIZE} so that downstream buffer + * allocation never silently wraps to a negative int or to a smaller positive + * value than requested. + */ + private static final int MAX_FILE_RANGE_LENGTH = MAX_ARRAY_SIZE; + + /** + * Snapshot of a {@link BufferChunk} sufficient to drive coalescing decisions + * without holding an actual chunk or backing buffer. Used to test the + * length-bounded coalescing logic without allocating multi-GB byte arrays. + */ + static final class CoalesceEntry { + final long offset; + final int length; + final boolean hasData; + + CoalesceEntry(long offset, int length, boolean hasData) { + this.offset = offset; + this.length = length; + this.hasData = hasData; + } + + long end() { + return offset + length; + } + } + + /** + * Compute the list of {@link FileRange}s that would be emitted for the given + * ordered list of coalescing entries. Honours filled-chunk boundaries, the + * gap==0/overlap merge rule, and the {@link #MAX_FILE_RANGE_LENGTH} size + * cap; the merged span is computed in {@code long} arithmetic so very large + * offsets never wrap. + *

+ * Exposed package-private so unit tests can drive coalescing with controlled + * inputs (including offsets near {@link Integer#MAX_VALUE}) without + * allocating multi-gigabyte byte buffers. + */ + static List coalesceRanges(List entries) { + return coalesceRanges(entries, 0); + } + + /** + * Variant of {@link #coalesceRanges(List)} that additionally honours a + * bounded small-gap merge: a pending chunk is folded into the in-flight + * FileRange when {@code nextOffset - currentEnd <= minSeekSize}, mirroring + * the sync-path semantics of {@link #findSingleRead}. Existing invariants + * are preserved: + *

+ * With {@code minSeekSize == 0} the behaviour is identical to the no-gap + * variant (touching/overlap merge only). The vectored path bounds over-read + * by {@code minSeekSize} alone; the on-disk read is never enlarged by the + * {@code minSeekSizeTolerance} parameter. The tolerance is applied + * after the read by {@link #readDiskRangesVectored} as a memory + * bound: each merged range that exceeds the tolerance is compacted into + * a tightly-sized buffer mirroring the sync-path + * {@code ChunkReader.populateChunksReduceSize} semantics. + * + * @param entries ordered list of coalescing entries + * @param minSeekSize non-negative tolerance (in bytes) for small gaps that + * may still be merged into a single FileRange + */ + static List coalesceRanges(List entries, int minSeekSize) { + if (minSeekSize < 0) { + throw new IllegalArgumentException( + "minSeekSize must be non-negative, got " + minSeekSize); + } + long minSeekSizeL = (long) minSeekSize; + List fileRanges = new ArrayList<>(); + FileRange currentRange = null; + long currentEnd = 0L; + for (CoalesceEntry e : entries) { + if (e.hasData) { + // Already-filled chunks are an immovable boundary. + currentRange = null; + continue; + } + // Each individual pending chunk must fit inside a single FileRange: + // FileRange lengths are int-typed and capped at MAX_FILE_RANGE_LENGTH + // so a caller that requests more than the cap for a single chunk + // cannot be served. Reject explicitly instead of letting the merged + // FileRange or the downstream allocator silently overflow. + if (e.length > MAX_FILE_RANGE_LENGTH) { + throw new IllegalArgumentException( + "BufferChunk length " + e.length + " exceeds MAX_ARRAY_SIZE " + + MAX_FILE_RANGE_LENGTH + " for offset " + e.offset); + } + long chunkOffset = e.offset; + long chunkEnd = e.end(); + // Merge into the in-flight range when either the chunk touches/overlaps + // the current end (gap == 0, i.e. chunkOffset <= currentEnd) OR the + // small gap is within the minSeekSize tolerance. Filling across a small + // gap mirrors the sync path's findSingleRead semantics. + boolean overlaps = (currentRange != null) && (chunkOffset <= currentEnd); + boolean smallGap = (currentRange != null) && + (chunkOffset - currentEnd <= minSeekSizeL); + if (overlaps || smallGap) { + long newEnd = Math.max(currentEnd, chunkEnd); + long mergedLength = newEnd - currentRange.getOffset(); + if (mergedLength > MAX_FILE_RANGE_LENGTH) { + // Adding this chunk (including its bounded over-read) would + // overflow the safe single-range length bound; flush and start a + // fresh range for the current chunk. + currentRange = FileRange.createFileRange(chunkOffset, e.length); + fileRanges.add(currentRange); + currentEnd = chunkEnd; + } else { + // Replace the in-flight range with one whose length is the final + // merged span. Building a fresh FileRange keeps coalesceRanges free + // of any FileRangeImpl-specific mutation; the cap above guarantees + // the cast back to int is safe. + long baseOffset = currentRange.getOffset(); + currentRange = FileRange.createFileRange(baseOffset, (int) mergedLength); + fileRanges.set(fileRanges.size() - 1, currentRange); + currentEnd = newEnd; + } + } else { + // Non-contiguous (gap > minSeekSize) or first range: start a fresh + // FileRange. + currentRange = FileRange.createFileRange(chunkOffset, e.length); + fileRanges.add(currentRange); + currentEnd = chunkEnd; + } + } + return fileRanges; + } + + /** + * Read the list of ranges from the file by updating each range in the list. + *

+ * Phase 1 coalescing: ranges within the same batch that are contiguous or + * overlapping (offset ≤ current aggregated end, i.e. gap = 0) are merged + * into a single {@link FileRange}. Already-filled chunks + * ({@link BufferChunk#hasData()} returns true) are never crossed - they act + * as a boundary. There is no over-read across gaps, and coalescing never + * spans past the current batch because we iterate the list once in order + * and only look at the immediately preceding pending chunk. + *

+ * Phase 3 coalescing: in addition to Phase 1, a bounded small-gap merge is + * honoured - a chunk is folded into the in-flight FileRange when + * {@code nextOffset - currentEnd <= minSeekSize}. This mirrors the sync + * path's {@link #findSingleRead} rule so callers using vectored I/O do not + * regress on minSeekSize-bounded workloads (e.g. Hadoop's TestMinSeekSize + * which expects ~106% reads with minSeekSize=1MB). The tolerance is + * applied in {@code long} arithmetic; filled-chunk boundaries and the + * {@link #MAX_FILE_RANGE_LENGTH} cap remain hard limits. The on-disk read + * is bounded by {@code minSeekSize} alone; the {@code minSeekSizeTolerance} + * parameter is applied as a post-read memory bound: for every merged + * range whose extra-bytes fraction (gap bytes / requested bytes) exceeds + * the tolerance, the backing buffer is compacted into a tightly-sized + * buffer that holds only the requested chunk bytes - mirroring the sync + * path's {@code ChunkReader.populateChunksReduceSize} semantics. With + * {@code minSeekSize == 0} the behaviour is exactly the Phase 1 + * gap==0 / overlap-only merge. + *

+ * The merged span is computed in {@code long} arithmetic; if growing the + * in-flight range would push its length past + * {@link #MAX_FILE_RANGE_LENGTH} the range is flushed and a new one is + * started for the current chunk. This avoids the silent {@code (int)} wrap + * that would otherwise produce a negative or truncated FileRange when caller + * offsets approach or exceed {@link Integer#MAX_VALUE}. + * + * @param fileInputStream the input stream to read from + * @param range the ordered list of disk ranges within the + * file to read + * @param doForceDirect allocate direct buffers + * @param minSeekSize non-negative minimum gap (in bytes) that is + * still merged into a single FileRange. + * {@code 0} disables the bounded gap merge + * and restores the Phase 1 behaviour exactly. + * @param minSeekSizeTolerance non-negative extra-bytes fraction above + * which a merged range's buffer is compacted + * to exactly the requested chunk bytes. The + * compaction is a post-read memory bound only; + * it does NOT change how many bytes are read + * from disk (the read is bounded by + * {@code minSeekSize}). Mirrors the sync + * path's {@code ChunkReader.populateChunks} + * semantics so a vectored reader pays no + * extra memory for small-gap merges. */ private static void readDiskRangesVectored( FSDataInputStream fileInputStream, BufferChunkList range, - boolean doForceDirect) throws IOException { + boolean doForceDirect, + int minSeekSize, + double minSeekSizeTolerance) throws IOException { if (range == null) return; IntFunction allocate = doForceDirect ? ByteBuffer::allocateDirect : ByteBuffer::allocate; - var fileRanges = new ArrayList(); - var map = new HashMap(); - var cur = range.get(); + // Build the snapshot of BufferChunks we want to coalesce. + List entries = new ArrayList<>(); + for (BufferChunk c = range.get(); c != null; c = (BufferChunk) c.next) { + entries.add(new CoalesceEntry(c.getOffset(), c.getLength(), c.hasData())); + } + List fileRanges = coalesceRanges(entries, minSeekSize); + + // Single-range short-circuit: a single coalesced range whose data the + // caller will block on immediately is pure async overhead when routed + // through readVectored (executor task + CompletableFuture with no + // parallelism benefit). Read it synchronously via the positioned + // readFully API and hand the backfill loop a FileRange whose future is + // already completed. The backfill/slice logic below is source-agnostic + // and works on either path. + if (fileRanges.size() == 1) { + FileRange only = fileRanges.get(0); + long onlyOffset = only.getOffset(); + int onlyLength = only.getLength(); + ByteBuffer full = allocate.apply(onlyLength); + try { + if (doForceDirect) { + // Direct buffers cannot be backed by readFully(long, ByteBuffer, + // ...). Read into a heap byte[] then copy into the direct buffer, + // mirroring the pattern used by readRanges(). + byte[] tmp = new byte[onlyLength]; + fileInputStream.readFully(onlyOffset, tmp, 0, onlyLength); + full.put(tmp); + full.flip(); + } else { + // Heap-backed ByteBuffer: readFully can target its backing array + // directly when there is no offset (allocate(...) returns a + // position-0, arrayOffset-0 buffer). + fileInputStream.readFully(onlyOffset, full.array(), full.arrayOffset(), + onlyLength); + full.limit(onlyLength); + } + } catch (IOException e) { + throw new IOException(String.format( + "Failed while reading %s %d:%d", fileInputStream, onlyOffset, onlyLength), e); + } + CompletableFuture done = new CompletableFuture<>(); + done.complete(full); + only.setData(done); + } else { + fileInputStream.readVectored(fileRanges, allocate); + } + + // Walk the original BufferChunk list and, for every empty chunk, slice the + // coalesced FileRange's buffer to the chunk's exact [offset, end). + // + // BufferChunks are visited in their original sorted order; fileRanges is + // also strictly increasing in offset. We use a single forward-moving + // pointer so the total backfill cost is O(chunks + ranges) instead of + // O(chunks * ranges). We also remember which BufferChunks each FileRange + // owns (in chunk order) so the post-slicing compaction pass can decide + // per-range whether to drop the over-read bytes from memory. + List> chunksByRange = new java.util.ArrayList<>(fileRanges.size()); + for (int i = 0; i < fileRanges.size(); i++) { + chunksByRange.add(new java.util.ArrayList<>()); + } + BufferChunk cur = range.get(); + int rangeIdx = 0; while (cur != null) { if (!cur.hasData()) { - var fileRange = FileRange.createFileRange(cur.getOffset(), cur.getLength()); - fileRanges.add(fileRange); - map.put(fileRange, cur); + long chunkOffset = cur.getOffset(); + long chunkEnd = chunkOffset + cur.getLength(); + // Skip past any fileRange that ends at or before the chunk start. + while (rangeIdx < fileRanges.size() && + (long) fileRanges.get(rangeIdx).getOffset() + + fileRanges.get(rangeIdx).getLength() <= chunkOffset) { + rangeIdx++; + } + // The current range may start at or before chunkOffset but end before + // chunkEnd (e.g. a pending chunk on each side of an already-filled + // boundary chunk whose spans overlap the pending ones). In that case + // the range cannot own the chunk; keep scanning forward for a later + // range whose start is still <= chunkOffset and whose end reaches + // chunkEnd. Because ranges are sorted by offset and non-overlapping, + // any such later range is the unique owner. rangeIdx only moves + // forward, so the total work across all chunks stays O(chunks + + // ranges). + FileRange owner = null; + int ownerIdx = -1; + while (rangeIdx < fileRanges.size() && + fileRanges.get(rangeIdx).getOffset() <= chunkOffset) { + FileRange candidate = fileRanges.get(rangeIdx); + if ((long) candidate.getOffset() + candidate.getLength() >= chunkEnd) { + owner = candidate; + ownerIdx = rangeIdx; + break; + } + rangeIdx++; + } + if (owner == null) { + throw new IllegalStateException("No FileRange covers BufferChunk " + cur); + } + try { + ByteBuffer full = owner.getData().get(); + // readVectored may return a buffer whose position() is non-zero + // (e.g. ChecksumFileSystem hands back a pre-sliced view). Compute + // the absolute slice window using both the buffer's position and + // the in-range offset, and reset the slice's position to 0 so + // consumers see a clean [0, chunkLen) view of the chunk bytes. + int absPos = full.position() + (int) (chunkOffset - owner.getOffset()); + int absLimit = absPos + cur.getLength(); + if (absLimit > full.limit()) { + throw new IllegalStateException("ByteBuffer underflow for BufferChunk " + cur + + ": requested limit " + absLimit + " > buffer limit " + full.limit()); + } + ByteBuffer slice = full.duplicate(); + slice.position(absPos); + slice.limit(absLimit); + // Hand the consumer a chunk-sized view starting at 0. + slice = slice.slice(); + cur.setChunk(slice); + chunksByRange.get(ownerIdx).add(cur); + } catch (InterruptedException | ExecutionException e) { + throw new RuntimeException(e); + } } cur = (BufferChunk) cur.next; } - fileInputStream.readVectored(fileRanges, allocate); - for (FileRange r : fileRanges) { - cur = map.get(r); - try { - cur.setChunk(r.getData().get()); - } catch (InterruptedException | ExecutionException e) { - throw new RuntimeException(e); + // Post-read memory compaction. For each merged FileRange, compute the + // extra-bytes fraction as (rangeLength - reqBytes) / reqBytes. If the + // fraction exceeds minSeekSizeTolerance, the over-read gap bytes are + // dropped from memory by allocating a tightly-sized buffer and copying + // the requested chunk bytes contiguously in chunk order; each owning + // BufferChunk is then handed a slice of that compacted buffer. This + // mirrors the sync path's ChunkReader.populateChunksReduceSize + // semantics so a vectored reader pays no extra memory for small-gap + // merges. Ranges with reqBytes == 0 cannot be compacted; ranges that + // own exactly one chunk have no gap bytes and are skipped cheaply. + for (int i = 0; i < fileRanges.size(); i++) { + FileRange range_i = fileRanges.get(i); + List owned = chunksByRange.get(i); + if (owned.size() <= 1) { + continue; + } + long reqBytesL = 0L; + for (BufferChunk c : owned) { + reqBytesL += c.getLength(); + } + if (reqBytesL <= 0L) { + continue; + } + long rangeLength = range_i.getLength(); + long extraBytesFractionNum = rangeLength - reqBytesL; + if (extraBytesFractionNum <= 0L) { + continue; + } + double extraBytesFraction = extraBytesFractionNum / (double) reqBytesL; + if (extraBytesFraction <= minSeekSizeTolerance) { + continue; + } + int reqBytes = (int) reqBytesL; + ByteBuffer compacted; + if (doForceDirect) { + compacted = ByteBuffer.allocateDirect(reqBytes); + } else { + compacted = ByteBuffer.allocate(reqBytes); + } + for (BufferChunk c : owned) { + // Copy exactly c.getLength() bytes from the chunk's existing slice + // (which currently points into the full range buffer) into the next + // position in the compacted buffer. + ByteBuffer src = c.getData(); + int copyLen = c.getLength(); + compacted.put(src.duplicate().position(0).limit(copyLen)); + } + compacted.flip(); + // Re-slice each BufferChunk to its window in the compacted buffer. + int copyStart = 0; + for (BufferChunk c : owned) { + int copyLen = c.getLength(); + ByteBuffer slice = compacted.duplicate(); + slice.position(copyStart); + slice.limit(copyStart + copyLen); + c.setChunk(slice.slice()); + copyStart += copyLen; + assert copyStart <= reqBytes; + } + if (LOG.isDebugEnabled()) { + LOG.debug("Vectored compaction: range [{}, {}), reqBytes={}, extraFraction={}, " + + "tolerance={}", range_i.getOffset(), + range_i.getOffset() + rangeLength, reqBytes, extraBytesFraction, + minSeekSizeTolerance); } } } diff --git a/java/core/src/test/org/apache/orc/impl/MockDFSDataInputStream.java b/java/core/src/test/org/apache/orc/impl/MockDFSDataInputStream.java index d5135c0bf2..536e9bf099 100644 --- a/java/core/src/test/org/apache/orc/impl/MockDFSDataInputStream.java +++ b/java/core/src/test/org/apache/orc/impl/MockDFSDataInputStream.java @@ -18,6 +18,7 @@ package org.apache.orc.impl; +import org.apache.hadoop.fs.FileRange; import org.apache.hadoop.fs.HasEnhancedByteBufferAccess; import org.apache.hadoop.fs.PositionedReadable; import org.apache.hadoop.fs.ReadOption; @@ -27,8 +28,13 @@ import java.io.IOException; import java.io.InputStream; import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; import java.util.EnumSet; import java.util.IdentityHashMap; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.function.IntFunction; public class MockDFSDataInputStream extends InputStream implements Seekable, PositionedReadable, HasEnhancedByteBufferAccess { @@ -36,6 +42,18 @@ public class MockDFSDataInputStream extends InputStream implements Seekable, Pos private int startPosition; private int currentPosition; private IdentityHashMap bufferStore = new IdentityHashMap(0); + // Tracks FileRange objects issued via readVectored for test assertions. + private final List lastVectoredRanges = new ArrayList<>(); + // Counters distinguishing the sync (single-range short-circuit) path from + // the readVectored path. Tests assert that batches coalescing to a single + // range do NOT invoke readVectored. + private int vectoredCallCount = 0; + private int readFullyCallCount = 0; + // When true, readVectored hands back a ByteBuffer with position() == some + // non-zero pre-set value, mimicking filesystems that pre-slice (e.g. + // ChecksumFileSystem). This drives the position()/remaining() code path + // in RecordReaderUtils. + private boolean handBackPreSlicedBuffer = false; public MockDFSDataInputStream(ByteBuffer hdfsBlockBuffer, int startPosition) { this.hdfsBlockBuffer = hdfsBlockBuffer; @@ -43,6 +61,17 @@ public MockDFSDataInputStream(ByteBuffer hdfsBlockBuffer, int startPosition) { this.currentPosition = startPosition; } + /** + * Toggle a mode where {@link #readVectored} hands back a ByteBuffer whose + * {@code position()} is non-zero (mimicking pre-slicing filesystems such + * as ChecksumFileSystem). Used to exercise the position-aware slice path + * in {@code RecordReaderUtils.readDiskRangesVectored}. + */ + public MockDFSDataInputStream withPreSlicedBufferMode() { + this.handBackPreSlicedBuffer = true; + return this; + } + @Override public int read() throws IOException { currentPosition++; @@ -92,13 +121,87 @@ public int read(long l, byte[] bytes, int i, int i1) throws IOException { throw new RuntimeException("unsupported"); } + /** + * Synchronous positioned read used by the single-range short-circuit in + * {@code RecordReaderUtils.readDiskRangesVectored}. Copies bytes from the + * backing buffer into {@code bytes} starting at offset {@code i}, for + * {@code i1} bytes. Mirrors the semantics of FSDataInputStream.readFully + * for the contract the short-circuit depends on (throws on short read). + */ @Override - public void readFully(long l, byte[] bytes, int i, int i1) throws IOException { - throw new RuntimeException("unsupported"); + public void readFully(long offset, byte[] bytes, int dstArrayOffset, int length) + throws IOException { + readFullyCallCount++; + int srcPos = (int) (offset - startPosition); + if (srcPos < 0 || length < 0 + || srcPos + length > hdfsBlockBuffer.capacity()) { + throw new IOException("readFully out of bounds: offset=" + offset + + " length=" + length + " backing capacity=" + hdfsBlockBuffer.capacity()); + } + ByteBuffer dup = hdfsBlockBuffer.duplicate(); + dup.position(srcPos); + dup.limit(srcPos + length); + dup.get(bytes, dstArrayOffset, length); + currentPosition = srcPos + length + startPosition; } @Override public void readFully(long l, byte[] bytes) throws IOException { - throw new RuntimeException("unsupported"); + readFully(l, bytes, 0, bytes.length); + } + + /** + * Vectored read test entry point: for every FileRange in {@code ranges} + * synchronously copy the requested bytes from the backing buffer into a + * freshly allocated ByteBuffer (per the {@code allocate} factory) and + * complete the FileRange's future with that buffer. Also records the ranges + * for {@link #getLastVectoredRanges()}. + */ + @Override + public void readVectored(List ranges, + IntFunction allocate) throws IOException { + vectoredCallCount++; + lastVectoredRanges.clear(); + for (FileRange range : ranges) { + lastVectoredRanges.add(range); + ByteBuffer buf = allocate.apply(range.getLength()); + // Position the backing buffer at the FileRange's offset. + int srcPos = (int) (range.getOffset() - startPosition); + buf.put(hdfsBlockBuffer.duplicate().position(srcPos).limit(srcPos + range.getLength())); + buf.flip(); + if (handBackPreSlicedBuffer && buf.capacity() > 1) { + // Simulate a pre-sliced view (e.g. ChecksumFileSystem): leave some + // bytes before position() and trim the limit. We pre-allocate a + // larger buffer, set position to a non-zero value, then put the + // bytes and flip. + int pre = Math.min(buf.position(), 7); + int extra = pre; + ByteBuffer larger = allocate.apply(range.getLength() + extra); + larger.position(extra); + larger.put(buf); + larger.flip(); + // The buffer's "data" now starts at position() == extra; the caller + // must respect position() to find the bytes. + buf = larger; + } + CompletableFuture future = new CompletableFuture<>(); + future.complete(buf); + range.setData(future); + } + } + + /** Returns the list of FileRanges most recently passed to readVectored. */ + public List getLastVectoredRanges() { + return Collections.unmodifiableList(lastVectoredRanges); + } + + /** Number of times {@link #readVectored} has been invoked on this stream. */ + public int getVectoredCallCount() { + return vectoredCallCount; + } + + /** Number of times {@link #readFully(long, byte[], int, int)} has been invoked. */ + public int getReadFullyCallCount() { + return readFullyCallCount; } } diff --git a/java/core/src/test/org/apache/orc/impl/TestRecordReaderImpl.java b/java/core/src/test/org/apache/orc/impl/TestRecordReaderImpl.java index 3c70b7284a..41c17cf89c 100644 --- a/java/core/src/test/org/apache/orc/impl/TestRecordReaderImpl.java +++ b/java/core/src/test/org/apache/orc/impl/TestRecordReaderImpl.java @@ -94,11 +94,14 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.fail; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.doNothing; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -2746,12 +2749,48 @@ public void testHadoopVectoredIO() throws Exception { Reader reader = OrcFile.createReader(filePath, OrcFile.readerOptions(conf).filesystem(spyLocalFileSystem)); - RecordReader recordReader = reader.rows(reader.options()); + // Project the first and last top-level columns so the stripe's + // BufferChunks form disjoint ranges. Vectored reads are only used for + // batches that still have at least two ranges after coalescing; a + // single contiguous range is read synchronously instead. + TypeDescription schema = reader.getSchema(); + List fieldNames = schema.getFieldNames(); + List children = schema.getChildren(); + TypeDescription projection = TypeDescription.fromString("struct<>"); + projection.addField(fieldNames.get(0), children.get(0).clone()); + projection.addField(fieldNames.get(fieldNames.size() - 1), + children.get(children.size() - 1).clone()); + RecordReader recordReader = reader.rows(reader.options().schema(projection)); recordReader.close(); verify(spyFSDataInputStream, atLeastOnce()).readVectored(any(), any()); } + @Test + public void testHadoopVectoredIOSingleRangeUsesSyncRead() throws Exception { + Configuration conf = new Configuration(); + Path filePath = new Path(TestVectorOrcFile.getFileFromClasspath("orc-file-11-format.orc")); + + FileSystem localFileSystem = FileSystem.getLocal(conf); + FSDataInputStream fsDataInputStream = localFileSystem.open(filePath); + + FileSystem spyLocalFileSystem = spy(localFileSystem); + FSDataInputStream spyFSDataInputStream = spy(fsDataInputStream); + when(spyLocalFileSystem.open(filePath)).thenReturn(spyFSDataInputStream); + + Reader reader = OrcFile.createReader(filePath, + OrcFile.readerOptions(conf).filesystem(spyLocalFileSystem)); + // A full scan produces one contiguous BufferChunk range per stripe. + // Coalescing reduces the batch to a single FileRange, which is fetched + // with a synchronous positioned read instead of readVectored. + RecordReader recordReader = reader.rows(reader.options()); + recordReader.close(); + + verify(spyFSDataInputStream, never()).readVectored(any(), any()); + verify(spyFSDataInputStream, atLeastOnce()) + .readFully(anyLong(), any(byte[].class), anyInt(), anyInt()); + } + @Test public void testDecimalIsRepeatingFlag() throws IOException { FileSystem fs = FileSystem.get(conf); diff --git a/java/core/src/test/org/apache/orc/impl/TestRecordReaderUtils.java b/java/core/src/test/org/apache/orc/impl/TestRecordReaderUtils.java index 28e161bd31..a4851ab2d8 100644 --- a/java/core/src/test/org/apache/orc/impl/TestRecordReaderUtils.java +++ b/java/core/src/test/org/apache/orc/impl/TestRecordReaderUtils.java @@ -19,10 +19,12 @@ package org.apache.orc.impl; import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FileRange; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.function.Executable; import java.io.IOException; +import java.lang.reflect.Method; import java.nio.ByteBuffer; import java.util.Arrays; import java.util.List; @@ -30,6 +32,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertThrowsExactly; @@ -279,4 +282,1371 @@ private void validateChunks(RecordReaderUtils.ChunkReader chunkReader) { current = (BufferChunk) current.next; } } + + // --------------------------------------------------------------------- + // Phase 1 vectored-IO coalescing tests (ORC-2223). + // These cover the readDiskRangesVectored path. We invoke the private + // static method directly via reflection so we can pass our own + // MockDFSDataInputStream that records the FileRanges it receives. + // --------------------------------------------------------------------- + + private static void invokeReadDiskRangesVectored(FSDataInputStream in, + BufferChunkList ranges, + boolean doForceDirect) + throws Exception { + invokeReadDiskRangesVectored(in, ranges, doForceDirect, 0, 0.0); + } + + private static void invokeReadDiskRangesVectored(FSDataInputStream in, + BufferChunkList ranges, + boolean doForceDirect, + int minSeekSize) + throws Exception { + invokeReadDiskRangesVectored(in, ranges, doForceDirect, minSeekSize, 0.0); + } + + private static void invokeReadDiskRangesVectored(FSDataInputStream in, + BufferChunkList ranges, + boolean doForceDirect, + int minSeekSize, + double minSeekSizeTolerance) + throws Exception { + Method m = RecordReaderUtils.class.getDeclaredMethod( + "readDiskRangesVectored", FSDataInputStream.class, BufferChunkList.class, + boolean.class, int.class, double.class); + m.setAccessible(true); + m.invoke(null, in, ranges, doForceDirect, minSeekSize, minSeekSizeTolerance); + } + + private static byte[] makePatternArray(int size, long baseOffset) { + byte[] bytes = new byte[size]; + for (int i = 0; i < size; i++) { + bytes[i] = (byte) ((i + baseOffset) % Byte.MAX_VALUE); + } + return bytes; + } + + private static MockDFSDataInputStream newBackingStream(byte[] backing) { + return new MockDFSDataInputStream(ByteBuffer.wrap(backing), 0); + } + + /** + * Zero-gap adjacent ranges must be coalesced into a single FileRange and + * each resulting BufferChunk must hold the correct slice of the underlying + * bytes. + */ + @Test + public void testVectoredCoalesceAdjacentGapZero() throws Exception { + // Backing file is 4096 bytes; pattern is byte i == (i % 127). + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 200) // [100, 300) + .range(300, 200) // [300, 500) - touches, gap = 0 + .range(500, 200).build(); // [500, 700) - touches, gap = 0 + + invokeReadDiskRangesVectored(in, ranges, false); + + // Single-range short-circuit: the batch coalesces to one FileRange so + // readVectored is NOT invoked; instead readFully is used. + assertEquals(0, dis.getVectoredCallCount(), + "Single-range batch must NOT invoke readVectored"); + assertEquals(1, dis.getReadFullyCallCount(), + "Single-range batch must invoke readFully exactly once"); + + // All three BufferChunks now have data and contain the right bytes. + assertArrayEquals( + Arrays.copyOfRange(backing, 100, 300), + byteBufferToArray(ranges.get(0).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 300, 500), + byteBufferToArray(ranges.get(1).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 500, 700), + byteBufferToArray(ranges.get(2).getData())); + + // Each slice must cover exactly the requested length. + for (int i = 0; i < 3; i++) { + assertEquals(ranges.get(i).getLength(), ranges.get(i).getData().remaining(), + "slice length mismatch for chunk " + i); + } + } + + /** + * Overlapping ranges (gap < 0 effectively) must also coalesce without + * expanding the FileRange beyond the union. + */ + @Test + public void testVectoredCoalesceOverlapping() throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(200, 300) // [200, 500) + .range(450, 200) // [450, 650) - overlaps the prior end + .range(650, 100).build(); // [650, 750) - touches + + invokeReadDiskRangesVectored(in, ranges, false); + + // Single-range short-circuit: readVectored is NOT invoked. + assertEquals(0, dis.getVectoredCallCount(), + "Single-range batch must NOT invoke readVectored"); + assertEquals(1, dis.getReadFullyCallCount(), + "Single-range batch must invoke readFully exactly once"); + + assertArrayEquals( + Arrays.copyOfRange(backing, 200, 500), + byteBufferToArray(ranges.get(0).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 450, 650), + byteBufferToArray(ranges.get(1).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 650, 750), + byteBufferToArray(ranges.get(2).getData())); + } + + /** + * A non-zero gap between ranges must keep them as separate FileRanges. No + * gap over-read is permitted. + */ + @Test + public void testVectoredGapNotMerged() throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 100) // [100, 200) + .range(400, 100) // [400, 500) - gap of 200 + .range(700, 100).build(); // [700, 800) - gap of 200 + + invokeReadDiskRangesVectored(in, ranges, false); + + List issued = dis.getLastVectoredRanges(); + assertEquals(3, issued.size(), "Non-zero gap must NOT coalesce"); + assertEquals(100L, issued.get(0).getOffset()); + assertEquals(100, issued.get(0).getLength()); + assertEquals(400L, issued.get(1).getOffset()); + assertEquals(100, issued.get(1).getLength()); + assertEquals(700L, issued.get(2).getOffset()); + assertEquals(100, issued.get(2).getLength()); + + for (int i = 0; i < 3; i++) { + assertArrayEquals( + Arrays.copyOfRange(backing, (int) ranges.get(i).getOffset(), + (int) (ranges.get(i).getOffset() + ranges.get(i).getLength())), + byteBufferToArray(ranges.get(i).getData())); + } + } + + /** + * A BufferChunk that already has data is an immovable boundary; even if it + * is contiguous with surrounding pending chunks it must not be merged into + * them. + */ + @Test + public void testVectoredFilledChunkIsBoundary() throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 100) + // Pre-filled chunk: same span as the empty one would request. + .range(400, 100) + .range(500, 100).build(); + + // Mark the middle chunk as already filled with stub bytes; coalescing must + // never touch it. + byte[] preFilled = makePatternArray(100, 9999); + ranges.get(1).setChunk(ByteBuffer.wrap(preFilled)); + + invokeReadDiskRangesVectored(in, ranges, false); + + List issued = dis.getLastVectoredRanges(); + // Two ranges: [100, 200) and [500, 600). The filled [400, 500) is its + // own non-issued block. + assertEquals(2, issued.size(), + "Filled chunk must be a hard boundary between pending ranges"); + assertEquals(100L, issued.get(0).getOffset()); + assertEquals(100, issued.get(0).getLength()); + assertEquals(500L, issued.get(1).getOffset()); + assertEquals(100, issued.get(1).getLength()); + + // The pre-filled chunk must be untouched. + assertArrayEquals(preFilled, byteBufferToArray(ranges.get(1).getData())); + // The two pending chunks have their correct data. + assertArrayEquals(Arrays.copyOfRange(backing, 100, 200), + byteBufferToArray(ranges.get(0).getData())); + assertArrayEquals(Arrays.copyOfRange(backing, 500, 600), + byteBufferToArray(ranges.get(2).getData())); + } + + /** + * Mixed pattern: some ranges coalesce, others do not; ensures the loop + * correctly handles transitions between mergeable and non-mergeable chunks. + */ + @Test + public void testVectoredMixedCoalesceAndGap() throws Exception { + byte[] backing = makePatternArray(8192, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 100) // group A start [100, 200) + .range(200, 100) // A [200, 300) merge + .range(300, 100) // A [300, 400) merge + .range(800, 100) // group B start [800, 900) + .range(900, 100) // B [900, 1000) merge + .range(2000, 100).build(); // group C [2000, 2100) + + invokeReadDiskRangesVectored(in, ranges, false); + + List issued = dis.getLastVectoredRanges(); + assertEquals(3, issued.size(), "Mixed pattern must produce 3 ranges"); + assertEquals(100L, issued.get(0).getOffset()); + assertEquals(300, issued.get(0).getLength()); + assertEquals(800L, issued.get(1).getOffset()); + assertEquals(200, issued.get(1).getLength()); + assertEquals(2000L, issued.get(2).getOffset()); + assertEquals(100, issued.get(2).getLength()); + + for (int i = 0; i < 6; i++) { + long off = ranges.get(i).getOffset(); + int len = ranges.get(i).getLength(); + assertArrayEquals( + Arrays.copyOfRange(backing, (int) off, (int) (off + len)), + byteBufferToArray(ranges.get(i).getData())); + } + } + + /** + * With doForceDirect=true the allocated buffers must be direct. The same + * coalescing and slicing behaviour must hold. + */ + @Test + public void testVectoredCoalesceDirectBuffer() throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 200) + .range(300, 200) + .range(500, 200).build(); + + invokeReadDiskRangesVectored(in, ranges, true); + + // Single-range short-circuit: readVectored is NOT invoked. + assertEquals(0, dis.getVectoredCallCount(), + "Single-range batch must NOT invoke readVectored"); + assertEquals(1, dis.getReadFullyCallCount(), + "Single-range batch must invoke readFully exactly once"); + // Each chunk's slice must be direct and contain correct bytes. + for (int i = 0; i < 3; i++) { + assertTrue(ranges.get(i).getData().isDirect(), + "Chunk slice must be direct for chunk " + i); + long off = ranges.get(i).getOffset(); + int len = ranges.get(i).getLength(); + assertArrayEquals( + Arrays.copyOfRange(backing, (int) off, (int) (off + len)), + byteBufferToArray(ranges.get(i).getData())); + } + } + + /** + * An empty BufferChunkList must be a no-op (no FileRange issued, no + * exception). + */ + @Test + public void testVectoredEmptyRangeList() throws Exception { + byte[] backing = makePatternArray(64, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + invokeReadDiskRangesVectored(in, new BufferChunkList(), false); + assertTrue(dis.getLastVectoredRanges().isEmpty(), + "Empty input must not produce any FileRange"); + } + + // --------------------------------------------------------------------- + // Phase 1 vectored-IO: long-safe span upper bound (no giant allocation). + // + // We exercise the package-private coalesceRanges helper directly so we can + // drive chunk shapes with multi-gigabyte lengths / offsets without + // materialising a backing buffer of that size. + // --------------------------------------------------------------------- + + private static List entries( + long[][] offsetsAndLengths, boolean[] hasData) { + List out = + new java.util.ArrayList<>(offsetsAndLengths.length); + for (int i = 0; i < offsetsAndLengths.length; i++) { + out.add(new RecordReaderUtils.CoalesceEntry( + offsetsAndLengths[i][0], (int) offsetsAndLengths[i][1], hasData[i])); + } + return out; + } + + /** Convert a list of FileRanges to a long[][] of {offset, length}. */ + private static long[][] fileRangesAsPairs(List ranges) { + long[][] pairs = new long[ranges.size()][]; + for (int i = 0; i < ranges.size(); i++) { + pairs[i] = new long[]{ ranges.get(i).getOffset(), + ranges.get(i).getLength() }; + } + return pairs; + } + + /** + * Two adjacent chunks whose merged span equals exactly + * RecordReaderUtils.MAX_ARRAY_SIZE must coalesce into a single FileRange. + */ + @Test + public void testCoalesceRangesAtMaxArraySize() { + int max = RecordReaderUtils.MAX_ARRAY_SIZE; + // First chunk of length max - 100, second chunk of length 100 that + // touches it. Merged span = max exactly. + long[][] in = new long[][]{ + {0L, max - 100L}, + {(long) (max - 100), 100L}}; + List out = RecordReaderUtils.coalesceRanges( + entries(in, new boolean[]{false, false})); + assertEquals(1, out.size(), + "Merged span equal to MAX_ARRAY_SIZE must stay in a single FileRange"); + assertEquals(0L, out.get(0).getOffset()); + assertEquals(max, out.get(0).getLength()); + assertArrayEquals(new long[][]{{0L, max}}, fileRangesAsPairs(out)); + } + + /** + * Two adjacent chunks whose merged span would push the FileRange length + * past RecordReaderUtils.MAX_ARRAY_SIZE must NOT silently truncate: the in-flight + * range must be flushed and a new range started for the over-cap chunk. + */ + @Test + public void testCoalesceRangesFlushesAtMaxArraySize() { + int max = RecordReaderUtils.MAX_ARRAY_SIZE; + long maxL = (long) max; + // Three adjacent small chunks. The cumulative span between chunk 1 + // and chunk 2 would push merged length past max, so chunk 2 must start + // a fresh range. Chunk 3 is adjacent to chunk 2 and merges into R2, + // giving R2 a final length of exactly max. + long[][] in = new long[][]{ + {0L, maxL - 100L}, + {maxL - 100L, 200L}, // cap flush -> new R2 + {maxL + 100L, maxL - 200L}}; // grows R2 to length max + List out = RecordReaderUtils.coalesceRanges( + entries(in, new boolean[]{false, false, false})); + assertEquals(2, out.size(), + "Span exceeding MAX_ARRAY_SIZE must split into multiple FileRanges"); + long[][] expected = new long[][]{ + {0L, maxL - 100L}, + {maxL - 100L, maxL}}; + assertArrayEquals(expected, fileRangesAsPairs(out)); + for (FileRange r : out) { + assertTrue(r.getLength() > 0 && r.getLength() <= max, + "FileRange length out of bounds: " + r.getLength()); + } + } + + /** + * Offsets themselves larger than Integer.MAX_VALUE must be supported: a + * {@code long}-offset chunk followed by an adjacent chunk must coalesce + * without overflow when computing the merged length. + */ + @Test + public void testCoalesceRangesWithLargeLongOffset() { + long hugeOffset = (long) Integer.MAX_VALUE + 1_000_000L; + long[][] in = new long[][]{ + {hugeOffset, 100L}, + {hugeOffset + 100L, 50L}}; + List out = RecordReaderUtils.coalesceRanges( + entries(in, new boolean[]{false, false})); + assertEquals(1, out.size(), + "Adjacent chunks past Integer.MAX_VALUE must coalesce"); + assertEquals(hugeOffset, out.get(0).getOffset()); + assertEquals(150, out.get(0).getLength()); + } + + /** + * A coalesce where adding the next chunk would push the merged length + * past {@code MAX_FILE_RANGE_LENGTH} must flush the current range and + * start a new one, with the cap enforced in {@code long} arithmetic. + */ + @Test + public void testCoalesceRangesSplitAcrossCapBoundary() { + int max = RecordReaderUtils.MAX_ARRAY_SIZE; + long maxL = (long) max; + // Group A: three adjacent chunks whose total cumulative span is + // 2*max - 50. The first three fit in a single capped span (max). + // Chunk 4 (adjacent to chunk 3) would push the merged span past max, + // so it must start a new range; chunk 5 then grows that second range. + long[][] in = new long[][]{ + {0L, maxL - 100L}, // [0, max - 100) + {maxL - 100L, 50L}, // -> [0, max - 50) + {maxL - 50L, 50L}, // -> fills R1 to max + {maxL, maxL - 100L}, // adjacent, fits cap + {2L * maxL - 100L, 50L}}; // grows R2 by 50 + List out = RecordReaderUtils.coalesceRanges( + entries(in, new boolean[]{false, false, false, false, false})); + long[][] expected = new long[][]{ + {0L, maxL}, + {maxL, maxL - 50L}}; + assertArrayEquals(expected, fileRangesAsPairs(out)); + // Every emitted length must remain positive and within the cap. + for (FileRange r : out) { + assertTrue(r.getLength() > 0 && r.getLength() <= max, + "FileRange length out of bounds: " + r.getLength()); + } + } + + /** + * Filled-chunk boundaries inside an over-cap region still split ranges + * correctly. Verifies the boundary reset coexists with the size-cap rule. + */ + @Test + public void testCoalesceRangesFilledBoundaryAndCap() { + int max = RecordReaderUtils.MAX_ARRAY_SIZE; + long maxL = (long) max; + // Two pending chunks that fill R1 to length max - 50, then a filled + // boundary chunk, then two adjacent pending chunks that grow R2 from + // initial length 200 up to 250. + long[][] in = new long[][]{ + {0L, maxL - 100L}, // R1 starts + {maxL - 100L, 50L}, // -> fills R1 to max-50 + {maxL - 50L, 0L}, // zero-length boundary chunk + {maxL - 50L, 200L}, // starts R2 + {maxL + 150L, 50L}}; // adjacent to last, grows R2 + boolean[] hasData = new boolean[]{false, false, true, false, false}; + List out = RecordReaderUtils.coalesceRanges(entries(in, hasData)); + long[][] expected = new long[][]{ + {0L, maxL - 50L}, + {maxL - 50L, 250L}}; + assertArrayEquals(expected, fileRangesAsPairs(out)); + } + + // --------------------------------------------------------------------- + // Phase 1 vectored-IO: existing backfill / byte-slice correctness with + // many small chunks per coalesced range. This exercises the monotonic + // pointer that walks BufferChunks and fileRanges in O(chunks + ranges). + // --------------------------------------------------------------------- + + /** + * Many small chunks spread across multiple coalesced groups. Each chunk + * must receive the correct byte-slice from its owning fileRange; the + * monotonic pointer must correctly advance across groups and within a + * group even with a long tail of chunk-per-byte-style requests. + */ + @Test + public void testVectoredManySmallChunksAcrossGroups() throws Exception { + int totalLen = 8192; + byte[] backing = makePatternArray(totalLen, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + TestOrcLargeStripe.RangeBuilder rb = new TestOrcLargeStripe.RangeBuilder(); + // Group A: 10 adjacent 16-byte chunks at [0, 16), [16, 32), ... + for (int i = 0; i < 10; i++) { + rb.range(i * 16L, 16); + } + // Group B: 10 adjacent 16-byte chunks at [1000, 1016), [1016, 1032), ... + for (int i = 0; i < 10; i++) { + rb.range(1000L + i * 16L, 16); + } + // Group C: 5 adjacent 8-byte chunks at [4000, 4008), [4008, 4016), ... + for (int i = 0; i < 5; i++) { + rb.range(4000L + i * 8L, 8); + } + BufferChunkList ranges = rb.build(); + + invokeReadDiskRangesVectored(in, ranges, false); + + List issued = dis.getLastVectoredRanges(); + assertEquals(3, issued.size(), + "Three contiguous groups should yield exactly three FileRanges"); + assertEquals(0L, issued.get(0).getOffset()); + assertEquals(160, issued.get(0).getLength()); + assertEquals(1000L, issued.get(1).getOffset()); + assertEquals(160, issued.get(1).getLength()); + assertEquals(4000L, issued.get(2).getOffset()); + assertEquals(40, issued.get(2).getLength()); + + // Every chunk must carry the bytes it requested. + int total = 25; + for (int i = 0; i < total; i++) { + long off = ranges.get(i).getOffset(); + int len = ranges.get(i).getLength(); + assertArrayEquals( + Arrays.copyOfRange(backing, (int) off, (int) (off + len)), + byteBufferToArray(ranges.get(i).getData()), + "slice mismatch for chunk " + i); + assertEquals(len, ranges.get(i).getData().remaining(), + "slice length mismatch for chunk " + i); + } + } + + /** + * Alternating long and short chunks across a non-trivial number of + * groups verify the backfill pointer correctly skips fileRange entries + * between groups while still matching each chunk to the right range. + */ + @Test + public void testVectoredInterleavedGroupMembership() throws Exception { + int totalLen = 8192; + byte[] backing = makePatternArray(totalLen, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + // 4 groups, each with 3 adjacent chunks of varying length, separated by + // gaps. + TestOrcLargeStripe.RangeBuilder rb = new TestOrcLargeStripe.RangeBuilder(); + long base = 0L; + int[][] lens = new int[][]{ + {50, 100, 200}, // group 1 + {30, 70, 90}, // group 2 + {200, 25, 25}, // group 3 + {10, 1000, 5}}; // group 4 + long[] gaps = new long[]{300L, 500L, 200L}; // gap between groups + long[][] specs = new long[12][]; + int idx = 0; + for (int g = 0; g < 4; g++) { + long pos = base; + for (int j = 0; j < 3; j++) { + specs[idx++] = new long[]{pos, lens[g][j]}; + pos += lens[g][j]; + } + if (g < 3) base = pos + gaps[g]; + } + for (long[] s : specs) { + rb.range(s[0], (int) s[1]); + } + BufferChunkList ranges = rb.build(); + + invokeReadDiskRangesVectored(in, ranges, false); + + List issued = dis.getLastVectoredRanges(); + assertEquals(4, issued.size(), + "Four groups separated by gaps should yield four FileRanges"); + // Each group's emitted FileRange covers the union of its three chunks. + int[] expectedTotalLen = new int[4]; + for (int g = 0; g < 4; g++) { + int sum = 0; + for (int j = 0; j < 3; j++) sum += lens[g][j]; + expectedTotalLen[g] = sum; + } + for (int g = 0; g < 4; g++) { + assertEquals(specs[g * 3][0], issued.get(g).getOffset(), + "group " + g + " offset"); + assertEquals(expectedTotalLen[g], issued.get(g).getLength(), + "group " + g + " length"); + } + + // Every chunk must carry the bytes it requested, regardless of group. + for (int i = 0; i < 12; i++) { + long off = ranges.get(i).getOffset(); + int len = ranges.get(i).getLength(); + assertArrayEquals( + Arrays.copyOfRange(backing, (int) off, (int) (off + len)), + byteBufferToArray(ranges.get(i).getData()), + "slice mismatch for chunk " + i); + } + } + + /** + * When {@code readVectored} returns a ByteBuffer whose {@code position()} + * is non-zero (e.g. ChecksumFileSystem hands back a pre-sliced view), + * the per-chunk slice handed to {@link BufferChunk#setChunk} must + * still start at {@code position() == 0} and have + * {@code remaining() == chunk.getLength()}, and its bytes must match + * the requested window of the file. + */ + @Test + public void testVectoredRespectsBufferPositionAndRemaining() throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing).withPreSlicedBufferMode(); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 200) // [100, 300) + .range(300, 200) // [300, 500) - coalesces with above + .range(900, 50).build(); // [900, 950) - separate group + + invokeReadDiskRangesVectored(in, ranges, false); + + List issued = dis.getLastVectoredRanges(); + assertEquals(2, issued.size()); + + // Each chunk must have a clean [0, chunkLen) slice with the right bytes. + for (int i = 0; i < 3; i++) { + BufferChunk c = ranges.get(i); + ByteBuffer data = c.getData(); + assertNotNull(data, "chunk " + i + " should have data"); + assertEquals(0, data.position(), + "chunk " + i + " slice must start at position 0"); + assertEquals(c.getLength(), data.remaining(), + "chunk " + i + " slice must have remaining() == chunk length"); + assertArrayEquals( + Arrays.copyOfRange(backing, (int) c.getOffset(), + (int) (c.getOffset() + c.getLength())), + byteBufferToArray(data), + "bytes mismatch for chunk " + i); + } + } + + /** + * Regression for ORC-2223 three-chunk case: a pending chunk + * {@code [0, 10)}, a pre-filled chunk {@code [5, 15)}, and a pending + * chunk {@code [6, 16)}. The filled chunk is an immovable boundary so + * coalescing emits two FileRanges: {@code [0, 10)} and {@code [6, 16)}. + * The backfill for the third chunk must skip the first (which ends at + * 10, before the chunk end 16) and locate the second; the filled + * chunk's data must be left untouched. + */ + @Test + public void testVectoredThreeChunkFilledBoundaryRegression() throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(0, 10) // [0, 10) pending + .range(5, 10) // [5, 15) pre-filled (boundary) + .range(6, 10).build(); // [6, 16) pending + + byte[] preFilled = makePatternArray(10, 9999); + ranges.get(1).setChunk(ByteBuffer.wrap(preFilled)); + + invokeReadDiskRangesVectored(in, ranges, false); + + List issued = dis.getLastVectoredRanges(); + assertEquals(2, issued.size(), + "Filled boundary must split pending chunks into two FileRanges"); + assertEquals(0L, issued.get(0).getOffset()); + assertEquals(10, issued.get(0).getLength()); + assertEquals(6L, issued.get(1).getOffset()); + assertEquals(10, issued.get(1).getLength()); + + // The filled chunk's data must be unchanged. + assertArrayEquals(preFilled, byteBufferToArray(ranges.get(1).getData()), + "Pre-filled middle chunk must be left untouched"); + + // The first pending chunk: bytes [0, 10) from the file. + assertArrayEquals( + Arrays.copyOfRange(backing, 0, 10), + byteBufferToArray(ranges.get(0).getData()), + "First pending chunk must contain backing bytes [0, 10)"); + + // The third pending chunk: bytes [6, 16) from the file. + assertArrayEquals( + Arrays.copyOfRange(backing, 6, 16), + byteBufferToArray(ranges.get(2).getData()), + "Third pending chunk must contain backing bytes [6, 16)"); + + // Each pending slice must be exactly the requested length. + assertEquals(0, ranges.get(0).getData().position(), + "First pending chunk slice must start at position 0"); + assertEquals(10, ranges.get(0).getData().remaining()); + assertEquals(0, ranges.get(2).getData().position(), + "Third pending chunk slice must start at position 0"); + assertEquals(10, ranges.get(2).getData().remaining()); + } + + /** + * Length cap rejection: a single CoalesceEntry whose length exceeds + * {@code RecordReaderUtils.MAX_ARRAY_SIZE} must be rejected with an + * {@link IllegalArgumentException} (no oversized FileRange must be + * emitted). Verified against the package-private {@code coalesceRanges} + * helper so the test does not allocate multi-GB backing buffers. + */ + @Test + public void testCoalesceRangesRejectsOverCapSingleEntry() { + int overCap = RecordReaderUtils.MAX_ARRAY_SIZE + 1; + long[][] in = new long[][]{ + {0L, (long) overCap}}; + IllegalArgumentException ex = assertThrowsExactly( + IllegalArgumentException.class, + () -> RecordReaderUtils.coalesceRanges( + entries(in, new boolean[]{false})), + "A single CoalesceEntry longer than MAX_ARRAY_SIZE must be rejected"); + assertTrue(ex.getMessage().contains(String.valueOf(overCap)), + "Exception message must surface the offending length: " + ex.getMessage()); + } + + // --------------------------------------------------------------------- + // Single-range short-circuit: when coalesceRanges produces exactly one + // FileRange, readDiskRangesVectored must NOT invoke readVectored; instead + // it must perform a synchronous positioned read and backfill the + // BufferChunks identically. Multi-range batches must still go through + // readVectored. + // --------------------------------------------------------------------- + + /** + * A batch that coalesces to a single range must NOT invoke readVectored + * and must populate each chunk with the correct bytes via the sync path. + */ + @Test + public void testVectoredSingleRangeSyncShortCircuit() throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 200) // [100, 300) + .range(300, 200) // [300, 500) - touches + .range(500, 200).build(); // [500, 700) - touches + + int vectoredBefore = dis.getVectoredCallCount(); + int readFullyBefore = dis.getReadFullyCallCount(); + invokeReadDiskRangesVectored(in, ranges, false); + + assertEquals(0, dis.getVectoredCallCount() - vectoredBefore, + "Single-range batch must NOT invoke readVectored"); + assertEquals(1, dis.getReadFullyCallCount() - readFullyBefore, + "Single-range batch must invoke readFully exactly once"); + assertTrue(dis.getLastVectoredRanges().isEmpty(), + "readVectored must not record any ranges on the sync path"); + + // Each chunk must hold the bytes it requested. + assertArrayEquals( + Arrays.copyOfRange(backing, 100, 300), + byteBufferToArray(ranges.get(0).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 300, 500), + byteBufferToArray(ranges.get(1).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 500, 700), + byteBufferToArray(ranges.get(2).getData())); + + // Each slice must be a clean [0, chunkLen) view. + for (int i = 0; i < 3; i++) { + assertEquals(0, ranges.get(i).getData().position(), + "chunk " + i + " slice must start at position 0"); + assertEquals(ranges.get(i).getLength(), + ranges.get(i).getData().remaining(), + "chunk " + i + " slice must have remaining() == chunk length"); + // Sync path uses heap allocation; buffers must NOT be direct. + assertTrue(!ranges.get(i).getData().isDirect(), + "Heap allocation path must produce non-direct slices"); + } + } + + /** + * Single-range short-circuit must also work with direct buffers. The + * sync path must read into a temp byte[] then put into the direct + * buffer (mirroring the readRanges pattern). + */ + @Test + public void testVectoredSingleRangeSyncShortCircuitDirect() throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(50, 100) + .range(150, 100) + .range(250, 100).build(); + + int vectoredBefore = dis.getVectoredCallCount(); + invokeReadDiskRangesVectored(in, ranges, true); + + assertEquals(0, dis.getVectoredCallCount() - vectoredBefore, + "Single-range batch with doForceDirect=true must NOT invoke readVectored"); + + for (int i = 0; i < 3; i++) { + BufferChunk c = ranges.get(i); + long off = c.getOffset(); + int len = c.getLength(); + assertArrayEquals( + Arrays.copyOfRange(backing, (int) off, (int) (off + len)), + byteBufferToArray(c.getData()), + "bytes mismatch for chunk " + i); + assertTrue(c.getData().isDirect(), + "Direct path must yield direct slices for chunk " + i); + assertEquals(0, c.getData().position()); + assertEquals(len, c.getData().remaining()); + } + } + + /** + * A batch that coalesces to a single range spanning multiple chunks + * must produce correct slices for each chunk (offsets/lengths verified + * per chunk). This is the chunk-slicing correctness case for the + * sync path. + */ + @Test + public void testVectoredSingleRangeSyncSlicingCorrectness() throws Exception { + // Build a 4-chunk batch where all four coalesce into one range. + // Mix chunk sizes to exercise offset/limit arithmetic across the + // shared ByteBuffer. + int totalLen = 8192; + byte[] backing = makePatternArray(totalLen, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(1000, 250) // [1000, 1250) + .range(1250, 500) // [1250, 1750) + .range(1750, 75) // [1750, 1825) + .range(1825, 1000).build(); // [1825, 2825) + + invokeReadDiskRangesVectored(in, ranges, false); + + assertEquals(0, dis.getVectoredCallCount(), + "Single-range batch must NOT invoke readVectored"); + + for (int i = 0; i < 4; i++) { + BufferChunk c = ranges.get(i); + long off = c.getOffset(); + int len = c.getLength(); + assertArrayEquals( + Arrays.copyOfRange(backing, (int) off, (int) (off + len)), + byteBufferToArray(c.getData()), + "slice bytes mismatch for chunk " + i); + assertEquals(0, c.getData().position(), + "chunk " + i + " must start at position 0"); + assertEquals(len, c.getData().remaining(), + "chunk " + i + " must have remaining() == length"); + } + } + + /** + * A batch with 2+ non-contiguous (gap > 0) ranges must still invoke + * readVectored and must NOT take the sync short-circuit. readFully must + * NOT be invoked. + */ + @Test + public void testVectoredMultiRangeStillUsesVectoredIO() throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 100) // [100, 200) + .range(400, 100) // [400, 500) - gap of 200 -> separate range + .range(700, 100).build(); // [700, 800) - gap of 200 -> separate range + + int readFullyBefore = dis.getReadFullyCallCount(); + invokeReadDiskRangesVectored(in, ranges, false); + + assertEquals(1, dis.getVectoredCallCount(), + "Multi-range batch MUST invoke readVectored exactly once"); + assertEquals(0, dis.getReadFullyCallCount() - readFullyBefore, + "Multi-range batch must NOT invoke readFully"); + + List issued = dis.getLastVectoredRanges(); + assertEquals(3, issued.size(), "Non-contiguous ranges stay separate"); + assertEquals(100L, issued.get(0).getOffset()); + assertEquals(400L, issued.get(1).getOffset()); + assertEquals(700L, issued.get(2).getOffset()); + + // Each chunk still receives correct bytes. + for (int i = 0; i < 3; i++) { + long off = ranges.get(i).getOffset(); + int len = ranges.get(i).getLength(); + assertArrayEquals( + Arrays.copyOfRange(backing, (int) off, (int) (off + len)), + byteBufferToArray(ranges.get(i).getData())); + } + } + + /** + * A batch of 2 ranges (the boundary case for the size check) must still + * invoke readVectored: the sync short-circuit is strictly size==1. + */ + @Test + public void testVectoredTwoRangeBatchStillUsesVectoredIO() throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 100) // [100, 200) + .range(400, 100) // [400, 500) - separate range + .build(); + + invokeReadDiskRangesVectored(in, ranges, false); + + assertEquals(1, dis.getVectoredCallCount(), + "Two-range batch MUST still use readVectored (size==1 only short-circuit)"); + assertEquals(0, dis.getReadFullyCallCount(), + "Two-range batch must NOT use readFully"); + } + + // --------------------------------------------------------------------- + // Phase 3 vectored-IO: bounded small-gap merge honouring minSeekSize. + // Mirrors the sync-path semantics of findSingleRead (RecordReaderUtils:500). + // With minSeekSize > 0, a gap of at most minSeekSize bytes is treated as + // contiguous: the gap bytes are over-read into a single FileRange that + // covers both chunks. With minSeekSize == 0 the behaviour is exactly the + // Phase 1 gap==0/overlap-only merge (no extra bytes read). + // --------------------------------------------------------------------- + + /** + * A gap of exactly {@code minSeekSize} bytes (and any smaller gap) must + * merge two pending chunks into a single FileRange whose span covers both + * chunks AND the gap bytes between them. Each merged chunk must still + * receive its exact byte window. + */ + @Test + public void testVectoredGapMergedWithinMinSeekSize() throws Exception { + byte[] backing = makePatternArray(8192, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + int minSeekSize = 500; + // First chunk ends at 200, second starts at 700 -> gap = 500 == minSeekSize. + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 100) // [100, 200) + .range(700, 100).build(); // [700, 800) - gap of 500 = minSeekSize + + invokeReadDiskRangesVectored(in, ranges, false, minSeekSize); + + // The sync short-circuit fires because gap merge reduces two FileRanges + // to one; readVectored must NOT be invoked. + assertEquals(0, dis.getVectoredCallCount(), + "Gap-merged batch coalescing to one FileRange must NOT invoke readVectored"); + assertEquals(1, dis.getReadFullyCallCount(), + "Gap-merged single range must invoke readFully exactly once"); + + // The over-read consumed [100, 800) inclusive of the 500-byte gap. + // Each chunk must hold its exact window of the backing bytes. + assertArrayEquals( + Arrays.copyOfRange(backing, 100, 200), + byteBufferToArray(ranges.get(0).getData()), + "First chunk must contain backing bytes [100, 200)"); + assertArrayEquals( + Arrays.copyOfRange(backing, 700, 800), + byteBufferToArray(ranges.get(1).getData()), + "Second chunk must contain backing bytes [700, 800)"); + assertEquals(0, ranges.get(0).getData().position()); + assertEquals(100, ranges.get(0).getData().remaining()); + assertEquals(0, ranges.get(1).getData().position()); + assertEquals(100, ranges.get(1).getData().remaining()); + } + + /** + * A gap strictly greater than {@code minSeekSize} must keep the chunks as + * separate FileRanges - the over-read is bounded by minSeekSize alone. + */ + @Test + public void testVectoredGapNotMergedBeyondMinSeekSize() throws Exception { + byte[] backing = makePatternArray(8192, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + int minSeekSize = 100; + // Gap = 200 > minSeekSize=100 must remain split. + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 100) // [100, 200) + .range(400, 100) // [400, 500) - gap = 200 > minSeekSize + .build(); + + invokeReadDiskRangesVectored(in, ranges, false, minSeekSize); + + List issued = dis.getLastVectoredRanges(); + assertEquals(1, dis.getVectoredCallCount(), + "Split batch must still invoke readVectored"); + assertEquals(2, issued.size(), + "Gap > minSeekSize must NOT be merged"); + assertEquals(100L, issued.get(0).getOffset()); + assertEquals(100, issued.get(0).getLength()); + assertEquals(400L, issued.get(1).getOffset()); + assertEquals(100, issued.get(1).getLength()); + + assertArrayEquals( + Arrays.copyOfRange(backing, 100, 200), + byteBufferToArray(ranges.get(0).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 400, 500), + byteBufferToArray(ranges.get(1).getData())); + } + + /** + * With {@code minSeekSize == 0} the Phase 3 path must behave exactly like + * Phase 1 - touching/overlap-only merge, no extra bytes read across gaps. + * A non-zero gap must NOT be merged. + */ + @Test + public void testVectoredMinSeekSizeZeroMatchesPhase1() throws Exception { + byte[] backing = makePatternArray(8192, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + // Gap of 1 byte > minSeekSize=0 must NOT merge; touching (gap==0) does. + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 100) // [100, 200) + .range(200, 100) // [200, 300) - touches, gap = 0 -> merges + .range(301, 100).build(); // [301, 401) - gap = 1 > 0, stays split + + invokeReadDiskRangesVectored(in, ranges, false, 0); + + // Two FileRanges emitted (touching merge for chunks 0+1, then chunk 2 + // stays split because gap=1 > minSeekSize=0). The batch is therefore + // multi-range; it goes through readVectored, NOT readFully. This is the + // exact Phase 1 behaviour - just exercising the vectored path with the + // minSeekSize=0 code path through coalesceRanges. + assertEquals(1, dis.getVectoredCallCount(), + "minSeekSize=0 must match Phase 1: touching merge keeps gap>0 split"); + assertEquals(0, dis.getReadFullyCallCount(), + "Multi-range batch must NOT invoke readFully"); + + List issued = dis.getLastVectoredRanges(); + assertEquals(2, issued.size(), + "minSeekSize=0 must match Phase 1: exactly two FileRanges"); + assertEquals(100L, issued.get(0).getOffset()); + assertEquals(200, issued.get(0).getLength()); + assertEquals(301L, issued.get(1).getOffset()); + assertEquals(100, issued.get(1).getLength()); + + // All three chunks still receive correct byte windows. + assertArrayEquals( + Arrays.copyOfRange(backing, 100, 200), + byteBufferToArray(ranges.get(0).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 200, 300), + byteBufferToArray(ranges.get(1).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 301, 401), + byteBufferToArray(ranges.get(2).getData())); + } + + /** + * A filled chunk is an immovable boundary even when a gap inside the + * minSeekSize tolerance exists on either side; the over-read must NOT + * cross the filled chunk. + */ + @Test + public void testVectoredFilledChunkBlocksGapMerge() throws Exception { + byte[] backing = makePatternArray(8192, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + int minSeekSize = 1000; + // Layout: + // chunk 0: [100, 200) pending + // chunk 1: [400, 500) pre-filled (boundary) + // chunk 2: [700, 800) pending - gap of 200 to chunk 1, gap of 200 to chunk 2 + // minSeekSize would otherwise merge 0<->1<->2 across the gap because + // 200 <= 1000, BUT chunk 1 is hasData() and must be an immovable boundary. + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 100) + .range(400, 100) + .range(700, 100).build(); + + byte[] preFilled = makePatternArray(100, 9999); + ranges.get(1).setChunk(ByteBuffer.wrap(preFilled)); + + invokeReadDiskRangesVectored(in, ranges, false, minSeekSize); + + List issued = dis.getLastVectoredRanges(); + // chunk 0 stands alone (no merge possible to the right past chunk 1 + // boundary); chunk 2 stands alone (no merge possible to the left past + // chunk 1 boundary). Two FileRanges total, filled chunk untouched. + assertEquals(2, issued.size(), + "Filled chunk must remain an immovable boundary under gap merge"); + assertEquals(100L, issued.get(0).getOffset(), + "Pre-fill must NOT be crossed: left chunk stays at offset 100"); + assertEquals(100, issued.get(0).getLength()); + assertEquals(700L, issued.get(1).getOffset(), + "Pre-fill must NOT be crossed: right chunk stays at offset 700"); + assertEquals(100, issued.get(1).getLength()); + + assertArrayEquals(preFilled, byteBufferToArray(ranges.get(1).getData()), + "Filled chunk must remain untouched"); + assertArrayEquals(Arrays.copyOfRange(backing, 100, 200), + byteBufferToArray(ranges.get(0).getData())); + assertArrayEquals(Arrays.copyOfRange(backing, 700, 800), + byteBufferToArray(ranges.get(2).getData())); + } + + /** + * {@link RecordReaderUtils#MAX_FILE_RANGE_LENGTH} cap still applies to + * the gap-merged FileRange: if merging across a small gap would push the + * resulting span past the cap, the in-flight range must be flushed and a + * new one started for the over-cap chunk. Verified via the package-private + * {@code coalesceRanges} helper so we can drive the inputs without + * materialising a multi-GB backing buffer. + */ + @Test + public void testVectoredGapMergeRespectsMaxFileRangeLength() { + int max = RecordReaderUtils.MAX_ARRAY_SIZE; + long maxL = (long) max; + int minSeekSize = 1024; + // R1 starts as [0, max-100) (length max-100). Chunk 1 begins at max-50 + // (gap 50 <= minSeekSize) so it would normally merge; the merged span is + // (max - 0) = max, exactly at the cap, so the merge is allowed (the cap + // is ">", not ">="). After the merge R1 has length max. Chunk 2 is far + // away so it starts R2. + // + // Then exercise the cap-flush branch: chunk 3 sits a small gap inside + // R2's tail but merging would overflow. R2 starts as [2*max-100, 2*max) + // (length 100). Chunk 3 begins at 2*max-50 (gap 50 <= minSeekSize). + // Merged span would be 2*max - (2*max-100) = max, exceeding the cap, so + // the cap-flush branch must fire and chunk 3 starts R3. + long[][] in = new long[][]{ + {0L, maxL - 100L}, // R1 starts at 0, ends at max-100 + {maxL - 50L, 50L}, // gap 50 -> merges into R1, R1 length = max + {2L * maxL + 100L, 100L}, // far away -> starts R2 + {2L * maxL + 200L, maxL - 50L}}; // gap 100 <= minSeekSize, merge would + // exceed cap -> flush, starts R3 + List out = RecordReaderUtils.coalesceRanges( + entries(in, new boolean[]{false, false, false, false}), minSeekSize); + long[][] expected = new long[][]{ + {0L, maxL}, + {2L * maxL + 100L, 100L}, + {2L * maxL + 200L, maxL - 50L}}; + assertArrayEquals(expected, fileRangesAsPairs(out)); + for (FileRange r : out) { + assertTrue(r.getLength() > 0 && r.getLength() <= max, + "FileRange length out of bounds: " + r.getLength()); + } + } + + /** + * {@code coalesceRanges(entries, minSeekSize)} must reject negative + * minSeekSize values to keep the gap-arithmetic well-defined. + */ + @Test + public void testCoalesceRangesRejectsNegativeMinSeekSize() { + long[][] in = new long[][]{ + {0L, 100L}, + {200L, 100L}}; + IllegalArgumentException ex = assertThrowsExactly( + IllegalArgumentException.class, + () -> RecordReaderUtils.coalesceRanges( + entries(in, new boolean[]{false, false}), -1), + "Negative minSeekSize must be rejected"); + assertTrue(ex.getMessage().contains("minSeekSize"), + "Exception message must mention minSeekSize: " + ex.getMessage()); + } + + // --------------------------------------------------------------------- + // Vectored read path: minSeekSizeTolerance memory compaction parity + // with the sync path's ChunkReader.populateChunksReduceSize semantics. + // + // The vectored path coalesces pending BufferChunks into one or more + // FileRanges (with bounded small-gap merge via minSeekSize). When the + // gap bytes included in a merged range push the extra-bytes fraction + // (gap / reqBytes) above minSeekSizeTolerance, the over-read bytes must + // be dropped from memory by allocating a tightly-sized buffer and + // copying only the requested chunk bytes contiguously in chunk order. + // --------------------------------------------------------------------- + + /** + * When a small-gap merge produces a single FileRange whose extra-bytes + * fraction exceeds {@code minSeekSizeTolerance}, the vectored path must + * compact the merged buffer to exactly the requested chunk bytes. The + * sync-path short-circuit fires (gap-merged batch coalesces to one + * FileRange, so readVectored is NOT invoked) and the over-read must be + * dropped from memory. Chunk contents must remain correct. + */ + @Test + public void testVectoredGapMergedAndCompactedWhenToleranceExceeded() + throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + int minSeekSize = 100; + // First chunk ends at 100, second starts at 200 -> gap = 100 == minSeekSize. + // reqBytes = 100 + 100 = 200, rangeLength = 300, fraction = 100/200 = 0.5. + // With tolerance = 0.0 the fraction (0.5) > 0.0 so compaction must fire. + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(0, 100) // [0, 100) + .range(200, 100).build(); // [200, 300) - gap of 100 = minSeekSize + + invokeReadDiskRangesVectored(in, ranges, false, minSeekSize, 0.0); + + // Sync short-circuit: readVectored must NOT be invoked. + assertEquals(0, dis.getVectoredCallCount(), + "Gap-merged batch must NOT invoke readVectored"); + assertEquals(1, dis.getReadFullyCallCount(), + "Gap-merged single range must invoke readFully exactly once"); + + // Chunk contents must remain correct after compaction. + assertArrayEquals( + Arrays.copyOfRange(backing, 0, 100), + byteBufferToArray(ranges.get(0).getData()), + "First chunk contents must be preserved across compaction"); + assertArrayEquals( + Arrays.copyOfRange(backing, 200, 300), + byteBufferToArray(ranges.get(1).getData()), + "Second chunk contents must be preserved across compaction"); + + // The compacted buffer holds exactly the requested chunk bytes (200 B), + // not the 300-byte over-read range. Verify the chunk's backing buffer + // capacity is bounded to 200, and that the two chunk slices are + // contiguous views of a single 200-byte backing array. + ByteBuffer c0 = ranges.get(0).getData(); + ByteBuffer c1 = ranges.get(1).getData(); + assertEquals(100, c0.capacity(), + "Compacted chunk 0 capacity must equal its length, not the range length"); + assertEquals(100, c1.capacity(), + "Compacted chunk 1 capacity must equal its length, not the range length"); + assertEquals(0, c0.position(), + "Compacted chunk 0 must start at position 0"); + assertEquals(0, c1.position(), + "Compacted chunk 1 must start at position 0"); + assertEquals(100, c0.remaining(), + "Compacted chunk 0 must have remaining() == 100"); + assertEquals(100, c1.remaining(), + "Compacted chunk 1 must have remaining() == 100"); + // The two slices must be backed by the same underlying array; that + // array's length must be exactly reqBytes = 200, not the range's 300. + assertEquals(c0.array(), c1.array(), + "Compacted chunks must share a single backing array"); + assertEquals(200, c0.array().length, + "Compacted backing array length must equal reqBytes (200), not the range length (300)"); + } + + /** + * When the extra-bytes fraction is within the tolerance, the merged + * range buffer is kept as-is: chunks remain slices of the over-read + * range buffer (no compaction). + */ + @Test + public void testVectoredGapMergedButNotCompactedWithinTolerance() + throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + int minSeekSize = 100; + // Same layout as the compaction test: reqBytes = 200, rangeLength = 300, + // fraction = 0.5. With tolerance = 1.0 the fraction (0.5) <= 1.0 so + // compaction must NOT fire; chunk slices must still point into the + // 300-byte range buffer. + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(0, 100) + .range(200, 100).build(); + + invokeReadDiskRangesVectored(in, ranges, false, minSeekSize, 1.0); + + // Sync short-circuit (single coalesced range). + assertEquals(0, dis.getVectoredCallCount(), + "Gap-merged batch must NOT invoke readVectored"); + assertEquals(1, dis.getReadFullyCallCount(), + "Gap-merged single range must invoke readFully exactly once"); + + // Chunk contents must still be correct. + assertArrayEquals( + Arrays.copyOfRange(backing, 0, 100), + byteBufferToArray(ranges.get(0).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 200, 300), + byteBufferToArray(ranges.get(1).getData())); + + // The two slices must NOT be backed by a 200-byte compacted array; + // they must remain slices of the 300-byte range buffer. + ByteBuffer c0 = ranges.get(0).getData(); + ByteBuffer c1 = ranges.get(1).getData(); + assertEquals(100, c0.capacity(), + "Non-compacted chunk 0 capacity must equal its length"); + assertEquals(100, c1.capacity(), + "Non-compacted chunk 1 capacity must equal its length"); + assertEquals(c0.array(), c1.array(), + "Non-compacted chunks must share the range backing array"); + assertEquals(300, c0.array().length, + "Non-compacted backing array must be the full 300-byte range buffer, not compacted to 200"); + } + + /** + * The compaction pass also applies on the single-range short-circuit + * path. A coalesced single range with an internal gap (e.g. two chunks + * merging into one range because the gap is within minSeekSize) must + * still have its over-read dropped when the fraction exceeds tolerance. + * readVectored is NOT invoked (the batch coalesces to one range); the + * sync readFully path runs, then the post-read compaction pass trims + * the over-read from memory. + */ + @Test + public void testVectoredSingleRangeShortCircuitIsAlsoCompacted() + throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + int minSeekSize = 100; + // Two chunks with a 100-byte gap that minSeekSize covers; batch + // coalesces to one FileRange, sync short-circuit fires. With + // tolerance = 0.0 the compaction must still fire on the single-range + // path because (300 - 200) / 200 = 0.5 > 0.0. + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 100) // [100, 200) + .range(300, 100).build(); // [300, 400) - gap of 100 = minSeekSize + + invokeReadDiskRangesVectored(in, ranges, false, minSeekSize, 0.0); + + assertEquals(0, dis.getVectoredCallCount(), + "Single-range short-circuit must NOT invoke readVectored"); + assertEquals(1, dis.getReadFullyCallCount(), + "Single-range short-circuit must invoke readFully exactly once"); + + // Contents must be correct. + assertArrayEquals( + Arrays.copyOfRange(backing, 100, 200), + byteBufferToArray(ranges.get(0).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 300, 400), + byteBufferToArray(ranges.get(1).getData())); + + // Backing array must be the 200-byte compacted buffer, not the + // 300-byte range buffer read off disk. + ByteBuffer c0 = ranges.get(0).getData(); + ByteBuffer c1 = ranges.get(1).getData(); + assertEquals(c0.array(), c1.array(), + "Compacted chunks must share a single backing array"); + assertEquals(200, c0.array().length, + "Single-range short-circuit backing array must be compacted to reqBytes (200), " + + "not the 300-byte over-read range"); + } + + /** + * Regression guard: existing minSeekSize=0 default (Phase 1) behaviour + * must not change. Two adjacent chunks coalesce to a single range with + * no over-read (reqBytes == rangeLength), so the extra-bytes fraction + * is 0, which is <= any non-negative tolerance - compaction is a no-op + * and chunk slices remain the original 200-byte range buffer. + */ + @Test + public void testVectoredToleranceCompactionNoOpOnContiguousMerge() + throws Exception { + byte[] backing = makePatternArray(4096, 0); + MockDFSDataInputStream dis = newBackingStream(backing); + FSDataInputStream in = new FSDataInputStream(dis); + + // minSeekSize = 0 (default), tolerance = 0.0; adjacent chunks touch, + // reqBytes = rangeLength = 200, so fraction = 0 <= 0.0 and no + // compaction should fire. + BufferChunkList ranges = new TestOrcLargeStripe.RangeBuilder() + .range(100, 100) + .range(200, 100).build(); + + invokeReadDiskRangesVectored(in, ranges, false, 0, 0.0); + + assertEquals(0, dis.getVectoredCallCount(), + "Single-range batch must NOT invoke readVectored"); + assertEquals(1, dis.getReadFullyCallCount(), + "Single-range batch must invoke readFully exactly once"); + + assertArrayEquals( + Arrays.copyOfRange(backing, 100, 200), + byteBufferToArray(ranges.get(0).getData())); + assertArrayEquals( + Arrays.copyOfRange(backing, 200, 300), + byteBufferToArray(ranges.get(1).getData())); + + // reqBytes == rangeLength so no compaction; backing array is the + // 200-byte range buffer. + ByteBuffer c0 = ranges.get(0).getData(); + ByteBuffer c1 = ranges.get(1).getData(); + assertEquals(c0.array(), c1.array(), + "Contiguous-merge chunks must share a single backing array"); + assertEquals(200, c0.array().length, + "Contiguous-merge backing array must remain the 200-byte range buffer"); + } }