sunchao commented on code in PR #3726:
URL: https://github.com/apache/parquet-java/pull/3726#discussion_r3800579324
##########
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:
Yes, this is a deliberate and meaningful behavior change. Even a synchronous
`UnsupportedOperationException` can follow partial submission, and the Hadoop
bridge only copies the futures back to Parquet after submission returns. During
result consumption, the chunk builder may also already contain data. Falling
back in either situation can race outstanding reads or replay chunks and return
incorrect rows.
Ordinary I/O still handles disabled/unavailable vectored I/O, and
preparation-time `IllegalArgumentException`/`UnsupportedOperationException`
still permits fallback. I clarified that boundary in
[ad623d6](https://github.com/apache/parquet-java/pull/3726/commits/ad623d6f4a44934ee2ccb98ab28bfee7242b0885)
and corrected the availability documentation: the Hadoop bridge checks runtime
API/allocator support, not per-stream `hasCapability`. The tests cover
ordinary-read selection plus synchronous, asynchronous, and partial-submission
failures without fallback.
I do not have an observed production rejection from an otherwise supported
backend. A genuinely pre-submission rejection should be recoverable, but the
current interface does not reliably distinguish it from partial submission.
Restoring broader fallback would need an explicit no-work-submitted guarantee;
I have kept the conservative behavior for now.
--
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]
