sunchao commented on code in PR #3726:
URL: https://github.com/apache/parquet-java/pull/3726#discussion_r4084776553
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -1357,32 +1364,153 @@ private boolean
arePartsValidForVectoredIo(List<ConsecutivePartList> allParts) {
* If directly implemented by a Filesystem then it is likely to be a more
efficient
* operation such as a scatter-gather read (native IO) or set of parallel
* GET requests against an object store.
+ * Submission and all requested ranges share the vectored-read timeout.
Failed
+ * operations invalidate the stream; cleanup waits for submission to exit
before
+ * closing it, even if the backend does not respond promptly to interruption.
+ * The allocation limit applies to filesystem buffers; decoders can still
require a
+ * contiguous buffer for an individual logical value larger than that limit.
* @param allParts all parts to be read.
* @param builder used to build chunk list to read the pages for the
different columns.
- * @throws IOException any IOE.
- * @throws IllegalArgumentException arguments are invalid.
- * @throws UnsupportedOperationException if the filesystem does not support
vectored IO.
+ * @throws IOException if submitting or consuming the vectored reads fails.
+ * @throws IllegalArgumentException if range preparation fails before any
reads are submitted.
*/
private void readVectored(List<ConsecutivePartList> allParts,
ChunkListBuilder builder) throws IOException {
-
+ final int maximumAllocation = options.getMaxAllocationSize();
+ Preconditions.checkArgument(maximumAllocation > 0, "Invalid maximum
allocation size %s", maximumAllocation);
+ final long fileLength = vectoredReadFileLength;
List<ParquetFileRange> ranges = new ArrayList<>(allParts.size());
+ List<Integer> partRangeCounts = new ArrayList<>(allParts.size());
long totalSize = 0;
for (ConsecutivePartList consecutiveChunks : allParts) {
final long len = consecutiveChunks.length;
- Preconditions.checkArgument(
- len < Integer.MAX_VALUE,
- "Invalid length %s for vectored read operation. It must be less than
max integer value.",
- len);
- ranges.add(new ParquetFileRange(consecutiveChunks.offset, (int) len));
+ final long start = consecutiveChunks.offset;
+ if (start < 0 || len < 0 || start > fileLength || len > fileLength -
start) {
+ throw new IOException(String.format(
+ "Invalid vectored read range (offset %d, length %d) for file
length %d",
+ start, len, fileLength));
+ }
+ final int firstRange = ranges.size();
+ long remaining = len;
+ long offset = start;
+ do {
+ int rangeLength = (int) Math.min(remaining, maximumAllocation);
Review Comment:
Yes: the strict bound is on the length of each range Parquet requests, not
on every buffer the filesystem allocates. Checksum alignment and range merging
can produce larger allocations. This is now explicit in the configuration
README, `ParquetReadOptions` API documentation, and reader documentation in
[5d3d6f3e](https://github.com/apache/parquet-java/commit/5d3d6f3eb37f48cb46f28d4f040b2dc2ecfb6828).
Footer/decoder allocations and total memory are outside this setting as well.
Strengthened the merged-buffer regression to verify that requested ranges
stay within 128 bytes while the backend allocates a larger merged buffer, all
rows decode correctly, and the original merged/checksum buffers are released.
The PR description now states this limitation rather than implying a hard
filesystem allocation cap.
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -1951,10 +2096,12 @@ protected PageHeader
readPageHeader(BlockCipher.Decryptor blockDecryptor, byte[]
* Calculate checksum of input bytes, throw decoding exception if it does
not match the provided
* reference crc
*/
- private void verifyCrc(int referenceCrc, BytesInput bytes, String
exceptionMsg) {
+ private void verifyCrc(int referenceCrc, String exceptionMsg,
BytesInput... inputs) throws IOException {
crc.reset();
- try (ByteBufferReleaser releaser = crcAllocator.getReleaser()) {
- crc.update(bytes.toByteBuffer(releaser));
+ for (BytesInput input : inputs) {
+ for (ByteBuffer buffer : input.toInputStream().remainingBuffers()) {
Review Comment:
Removed in
[5d3d6f3e](https://github.com/apache/parquet-java/commit/5d3d6f3eb37f48cb46f28d4f040b2dc2ecfb6828),
including the field, import, all constructor initialization, and close
argument. Incremental checksum verification already updates CRC directly from
the existing buffer views, so the scratch allocator is unnecessary.
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -1357,32 +1335,100 @@ private boolean
arePartsValidForVectoredIo(List<ConsecutivePartList> allParts) {
* If directly implemented by a Filesystem then it is likely to be a more
efficient
* operation such as a scatter-gather read (native IO) or set of parallel
* GET requests against an object store.
+ * The allocation limit applies to filesystem buffers; decoders can still
require a
+ * contiguous buffer for an individual logical value larger than that limit.
* @param allParts all parts to be read.
* @param builder used to build chunk list to read the pages for the
different columns.
- * @throws IOException any IOE.
- * @throws IllegalArgumentException arguments are invalid.
- * @throws UnsupportedOperationException if the filesystem does not support
vectored IO.
+ * @throws IOException if submitting or consuming the vectored reads fails.
+ * @throws IllegalArgumentException if range preparation fails before any
reads are submitted.
*/
private void readVectored(List<ConsecutivePartList> allParts,
ChunkListBuilder builder) throws IOException {
-
+ final int maximumAllocation = options.getMaxAllocationSize();
+ Preconditions.checkArgument(maximumAllocation > 0, "Invalid maximum
allocation size %s", maximumAllocation);
+ final long fileLength = file.getLength();
List<ParquetFileRange> ranges = new ArrayList<>(allParts.size());
+ List<Integer> partRangeCounts = new ArrayList<>(allParts.size());
long totalSize = 0;
for (ConsecutivePartList consecutiveChunks : allParts) {
final long len = consecutiveChunks.length;
- Preconditions.checkArgument(
- len < Integer.MAX_VALUE,
- "Invalid length %s for vectored read operation. It must be less than
max integer value.",
- len);
- ranges.add(new ParquetFileRange(consecutiveChunks.offset, (int) len));
+ final long start = consecutiveChunks.offset;
+ if (start < 0 || len < 0 || start > fileLength || len > fileLength -
start) {
+ throw new IOException(String.format(
+ "Invalid vectored read range (offset %d, length %d) for file
length %d",
+ start, len, fileLength));
+ }
+ final int firstRange = ranges.size();
+ long remaining = len;
+ long offset = start;
+ do {
+ int rangeLength = (int) Math.min(remaining, maximumAllocation);
+ ranges.add(new ParquetFileRange(offset, rangeLength));
+ offset += rangeLength;
+ remaining -= rangeLength;
+ } while (remaining > 0);
+ partRangeCounts.add(ranges.size() - firstRange);
totalSize += len;
}
LOG.debug("Reading {} bytes of data with vectored IO in {} ranges",
totalSize, ranges.size());
- // Request a vectored read;
- f.readVectored(ranges, options.getAllocator());
- int k = 0;
- for (ConsecutivePartList consecutivePart : allParts) {
- ParquetFileRange currRange = ranges.get(k++);
- consecutivePart.readFromVectoredRange(currRange, builder);
+ final long readStart = System.nanoTime();
+ try {
+ // Even a synchronous failure can occur after some reads have been
scheduled,
+ // so falling back to normal IO is unsafe once this call has been
entered.
+ f.readVectored(ranges, options.getAllocator());
+ int firstRange = 0;
+ for (int partIndex = 0; partIndex < allParts.size(); partIndex++) {
+ int endRange = firstRange + partRangeCounts.get(partIndex);
+
allParts.get(partIndex).readFromVectoredRanges(ranges.subList(firstRange,
endRange), builder);
+ firstRange = endRange;
+ }
+ } catch (IllegalArgumentException | UnsupportedOperationException e) {
Review Comment:
Kept the fail-fast behavior after submission is attempted. The updated PR
description now records this accepted trade-off and distinguishes it from safe
pre-submission metadata/preparation fallback.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
