sunchao commented on code in PR #3726:
URL: https://github.com/apache/parquet-java/pull/3726#discussion_r4178557437
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -1358,32 +1355,150 @@ 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 bounds requested range lengths. Filesystems may
merge or
+ * checksum-align ranges and allocate larger buffers; decoders may also
require
+ * larger contiguous buffers. This is not a total memory 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);
+ 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);
+ if (vectoredReadExecutor == null) {
+ vectoredReadExecutor = VectoredReadOperation.newExecutor();
+ }
+ VectoredReadOperation operation = new VectoredReadOperation(
+ f,
+ ranges,
+ options.getAllocator(),
+ vectoredReadExecutor,
+ HADOOP_VECTORED_READ_TIMEOUT_SECONDS,
+ TimeUnit.SECONDS);
+ try {
+ // Even a synchronous rejection can follow partial submission. The
Hadoop bridge
+ // may expose futures for reads which were never scheduled. Once
submission is
+ // attempted, neither missing futures nor a synchronous error permit
replay.
+ operation.awaitSubmission();
+ 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, operation);
+ firstRange = endRange;
+ }
+ operation.transferTo(builder.releaser);
+ } catch (TimeoutException e) {
+ IOException failure = new IOException("Timed out submitting vectored
reads", e);
+ abortVectoredRead(operation, failure);
+ throw failure;
+ } catch (IllegalArgumentException | UnsupportedOperationException e) {
+ // Consumption may also have populated the builder. Do not replay those
chunks.
+ IOException failure =
+ new IOException("Vectored read failed after asynchronous reads may
have been submitted", e);
+ if (operation.submissionSucceeded()) {
+ awaitRemainingVectoredReads(ranges, operation, failure);
+ }
+ abortVectoredRead(operation, failure);
Review Comment:
Fixed in
[8ea4302](https://github.com/apache/parquet-java/pull/3726/commits/8ea4302a61254de697f66ef20e5edc75729bbf69).
Both catch branches now run `abortVectoredRead` in `finally`, so an Error from
draining a sibling cannot bypass reader invalidation and deferred cleanup. The
Error still escapes, with the original read failure attached as suppressed.
Added four full-reader cases covering initial IOException,
IllegalArgumentException, UnsupportedOperationException, and Error. Each forces
a pending sibling to raise an Error and leaves a third read pending through
stream closure; the tests verify no premature buffer release, rejected reader
reuse, and complete reclamation after that last read finishes. Removing only
the finally guards makes all four cases fail.
Current validation: full parquet-hadoop module 869 tests, zero
failures/errors, 26 Hadoop 3.3.0 capability skips; focused Hadoop 3.4.1 run 110
tests, zero failures/errors/skips. Spotless and RAT pass.
--
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]
