wgtmac commented on code in PR #3726:
URL: https://github.com/apache/parquet-java/pull/3726#discussion_r3878936749
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -2343,32 +2396,39 @@ private void setReadMetrics(long startNs, long len) {
}
/**
- * Populate data in a parquet file range from a vectored range; will block
for up
- * to {@link #HADOOP_VECTORED_READ_TIMEOUT_SECONDS} seconds.
- * @param currRange range to populated.
+ * Populate data in a parquet file range from one or more bounded vectored
ranges; together
+ * they may block for up to {@link #HADOOP_VECTORED_READ_TIMEOUT_SECONDS}
seconds.
+ * @param ranges bounded ranges containing this part.
* @param builder used to build chunk list to read the pages for the
different columns.
* @throws IOException if there is an error while reading from the stream,
including a timeout.
*/
- public void readFromVectoredRange(ParquetFileRange currRange,
ChunkListBuilder builder) throws IOException {
- ByteBuffer buffer;
+ public void readFromVectoredRanges(List<ParquetFileRange> ranges,
ChunkListBuilder builder) throws IOException {
+ List<ByteBuffer> buffers = new ArrayList<>(ranges.size());
+ ParquetFileRange currentRange = null;
final long timeoutSeconds = HADOOP_VECTORED_READ_TIMEOUT_SECONDS;
+ final long timeoutNanos = TimeUnit.SECONDS.toNanos(timeoutSeconds);
long readStart = System.nanoTime();
try {
- LOG.debug(
- "Waiting for vectored read to finish for range {} with timeout {}
seconds",
- currRange,
- timeoutSeconds);
- buffer = FutureIO.awaitFuture(currRange.getDataReadFuture(),
timeoutSeconds, TimeUnit.SECONDS);
- setReadMetrics(readStart, currRange.getLength());
+ for (ParquetFileRange range : ranges) {
+ currentRange = range;
+ LOG.debug(
+ "Waiting for vectored read to finish for range {} with timeout
{} seconds",
+ range,
+ timeoutSeconds);
+ long remainingNanos = Math.max(timeoutNanos - (System.nanoTime() -
readStart), 0L);
+ buffers.add(FutureIO.awaitFuture(range.getDataReadFuture(),
remainingNanos, TimeUnit.NANOSECONDS));
Review Comment:
`readFromVectoredRanges()` never registers returned buffers with
`builder.addBuffersToRelease(...)`; tracking/pooling allocators therefore leak
them.
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -1357,32 +1334,105 @@ 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);
+ if (vectoredReadFileLength < 0) {
+ vectoredReadFileLength = file.getLength();
+ }
+ 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);
+ final long readStart = System.nanoTime();
+ try {
+ // Even a synchronous rejection can follow partial submission. The
Hadoop bridge
+ // publishes futures only after submission returns, so missing futures
do not prove
+ // that no reads started. Once this call is entered, normal-read
fallback is unsafe.
+ 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) {
+ // 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);
+ awaitRemainingVectoredReads(ranges, readStart, failure);
Review Comment:
After partial submission, futures are copied back to `ParquetFileRange` only
when `readWrappedRanges()` returns. A backend failure can leave started reads
untracked, so cleanup cannot wait for them.
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -1357,32 +1334,105 @@ 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);
+ if (vectoredReadFileLength < 0) {
+ vectoredReadFileLength = file.getLength();
+ }
+ 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);
+ final long readStart = System.nanoTime();
Review Comment:
The timeout starts only after `f.readVectored(...)` returns, so a blocking
submission is outside the 300-second limit.
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -1357,32 +1334,105 @@ 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);
+ if (vectoredReadFileLength < 0) {
+ vectoredReadFileLength = file.getLength();
Review Comment:
`file.getLength()` can fail before submission, but that `IOException`
bypasses the ordinary-read fallback.
##########
parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java:
##########
@@ -1357,32 +1334,105 @@ 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);
+ if (vectoredReadFileLength < 0) {
+ vectoredReadFileLength = file.getLength();
+ }
+ 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);
+ final long readStart = System.nanoTime();
+ try {
+ // Even a synchronous rejection can follow partial submission. The
Hadoop bridge
+ // publishes futures only after submission returns, so missing futures
do not prove
+ // that no reads started. Once this call is entered, normal-read
fallback is unsafe.
+ 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) {
+ // 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);
+ awaitRemainingVectoredReads(ranges, readStart, failure);
+ throw failure;
+ } catch (IOException | RuntimeException e) {
+ awaitRemainingVectoredReads(ranges, readStart, e);
+ throw e;
+ }
+ }
+
+ /**
+ * Wait for submitted reads with published futures to finish before their
stream can be
+ * closed. Cancelling result futures does not stop all Hadoop backends from
continuing IO.
+ */
+ private void awaitRemainingVectoredReads(List<ParquetFileRange> ranges, long
readStart, Throwable failure) {
+ if (Thread.currentThread().isInterrupted()
+ || failure instanceof InterruptedIOException && failure.getCause()
instanceof InterruptedException) {
Review Comment:
An interrupted wait returns before sibling futures are drained; `close()`
may then race with backend reads.
--
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]
