GH-3719: Fix vectored read allocation limits and fallback safety

parquet-hadoop/README.md

@@ -403,6 +403,17 @@ ParquetInputFormat to materialize records. It should be a the descendant class o
 **Description:** The number of threads will be spawned to compress in parallel. More workers improve speed, but also increase memory usage. When it is 0, it works as single-threaded mode.  
 **Default value:** `0`
 
+## Class: ParquetReadOptions
+
+**Property:** `parquet.read.allocation.size`
+
+**Description:** Size in bytes used to split planned column-chunk ranges into read buffers.
+For vectored IO, this limits the length of each range requested from the filesystem.
+Filesystems may merge or align ranges for checksums and allocate larger buffers.
+This setting does not limit other allocations, such as footer reads or decoding, or total memory use.
+
+**Default value:** `8388608` (8 MiB)
+
 ## Class: HadoopReadOptions
 
 **Property:** `parquet.crypto.factory.class`  
@@ -547,6 +558,54 @@ if not found then the library will use the classic non-vectored reads: it is saf
 
 ---
 
+**JVM system property:** `parquet.hadoop.vectored.io.threads`
+
+**Description:** Maximum number of workers in the default shared vectored-read pool. Set a positive integer
+with `-Dparquet.hadoop.vectored.io.threads=N` before the first vectored read; this is
+read once per JVM, not from the Hadoop configuration. Invalid values fail initialization.
+Readers without a caller-supplied executor share this limit. Waiting for a worker, submitting the read, and waiting
+for its results share the same 300-second deadline. If admission times out, the read
+fails before submitting filesystem work and the reader retains its stream.
+
+A worker remains occupied until submission returns, the caller transfers or abandons
+the buffers, and any failure cleanup closes the stream. A backend that ignores
+interruption therefore retains capacity; repeated timeouts cannot create workers
+beyond the configured limit. If all workers are blocked, further vectored reads time
+out waiting for capacity. Idle workers expire after 60 seconds. This limit does not
+bound threads created by the filesystem or total buffer memory.
+
+**Default value:** `64`
+
+Applications can supply their own executor through `ParquetReadOptions` (including
+`HadoopReadOptions`) or `ParquetReader.Builder.withVectoredReadExecutor` instead of
+using the default pool:
+
+```java
+ExecutorService ioPool = Executors.newFixedThreadPool(8);
+VectoredReadExecutor vectoredReads = new VectoredReadExecutor(ioPool, 8);
+ParquetReadOptions options = ParquetReadOptions.builder()
+    .withVectoredReadExecutor(vectoredReads)
+    .build();
+```
+
+Reuse the same `VectoredReadExecutor` across readers to share its admission limit.
+Copies of the read options retain that same instance. The limit bounds accepted
+vectored operations, including queued work and failure cleanup, independently of
+other tasks submitted to the underlying executor. A supplied executor bypasses the
+JVM default and its system property; passing `null` restores the default.
+
+The application owns the supplied executor: Parquet never shuts it down, including
+when a reader closes or fails. Its `execute` method must return promptly, run tasks asynchronously, and either accept
+them for execution or reject them with an exception, without silently dropping
+tasks. Inline execution (including `CallerRunsPolicy`) is rejected before filesystem
+submission. Reading threads must not occupy all of the supplied pool's workers
+while waiting for vectored reads on that same pool. After readers finish, use an
+orderly shutdown that drains accepted tasks; `shutdownNow()` or cancelling queued
+tasks can prevent deferred cleanup. Closing a failed reader does not imply that
+an interruption-ignoring backend or its cleanup has finished.
+
+---
+
 **Property:** `parquet.column.statistics.enabled`  
 **Description:** Whether to enable column statistics collection.
 If `true`, statistics will be collected for all columns unless explicitly disabled for specific columns.

parquet-hadoop/src/main/java/org/apache/parquet/HadoopReadOptions.java

@@ -30,6 +30,7 @@
 import org.apache.parquet.filter2.compat.FilterCompat;
 import org.apache.parquet.format.converter.ParquetMetadataConverter.MetadataFilter;
 import org.apache.parquet.hadoop.ParquetMetricsCallback;
+import org.apache.parquet.hadoop.VectoredReadExecutor;
 
 public class HadoopReadOptions extends ParquetReadOptions {
   private final Configuration conf;
@@ -52,7 +53,8 @@ private HadoopReadOptions(
       Map<String, String> properties,
       Configuration conf,
       FileDecryptionProperties fileDecryptionProperties,
-      ParquetMetricsCallback metricsCallback) {
+      ParquetMetricsCallback metricsCallback,
+      VectoredReadExecutor vectoredReadExecutor) {
     super(
         useSignedStringMinMax,
         useStatsFilter,
@@ -71,7 +73,8 @@ private HadoopReadOptions(
         properties,
         fileDecryptionProperties,
         metricsCallback,
-        new HadoopParquetConfiguration(conf));
+        new HadoopParquetConfiguration(conf),
+        vectoredReadExecutor);
     this.conf = conf;
   }
 
@@ -134,7 +137,8 @@ public ParquetReadOptions build() {
           properties,
           conf,
           fileDecryptionProperties,
-          metricsCallback);
+          metricsCallback,
+          vectoredReadExecutor);
     }
   }
 

parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java

@@ -46,6 +46,7 @@
 import org.apache.parquet.filter2.compat.FilterCompat;
 import org.apache.parquet.format.converter.ParquetMetadataConverter;
 import org.apache.parquet.hadoop.ParquetMetricsCallback;
+import org.apache.parquet.hadoop.VectoredReadExecutor;
 import org.apache.parquet.hadoop.util.HadoopCodecs;
 
 // Internal use only
@@ -80,6 +81,7 @@ public class ParquetReadOptions {
   private final FileDecryptionProperties fileDecryptionProperties;
   private final ParquetConfiguration conf;
   private final ParquetMetricsCallback metricsCallback;
+  private final VectoredReadExecutor vectoredReadExecutor;
 
   ParquetReadOptions(
       boolean useSignedStringMinMax,
@@ -139,6 +141,48 @@ public class ParquetReadOptions {
       FileDecryptionProperties fileDecryptionProperties,
       ParquetMetricsCallback metricsCallback,
       ParquetConfiguration conf) {
+    this(
+        useSignedStringMinMax,
+        useStatsFilter,
+        useDictionaryFilter,
+        useRecordFilter,
+        useColumnIndexFilter,
+        usePageChecksumVerification,
+        useBloomFilter,
+        useOffHeapDecryptBuffer,
+        useHadoopVectoredIo,
+        recordFilter,
+        metadataFilter,
+        codecFactory,
+        allocator,
+        maxAllocationSize,
+        properties,
+        fileDecryptionProperties,
+        metricsCallback,
+        conf,
+        null);
+  }
+
+  ParquetReadOptions(
+      boolean useSignedStringMinMax,
+      boolean useStatsFilter,
+      boolean useDictionaryFilter,
+      boolean useRecordFilter,
+      boolean useColumnIndexFilter,
+      boolean usePageChecksumVerification,
+      boolean useBloomFilter,
+      boolean useOffHeapDecryptBuffer,
+      boolean useHadoopVectoredIo,
+      FilterCompat.Filter recordFilter,
+      ParquetMetadataConverter.MetadataFilter metadataFilter,
+      CompressionCodecFactory codecFactory,
+      ByteBufferAllocator allocator,
+      int maxAllocationSize,
+      Map<String, String> properties,
+      FileDecryptionProperties fileDecryptionProperties,
+      ParquetMetricsCallback metricsCallback,
+      ParquetConfiguration conf,
+      VectoredReadExecutor vectoredReadExecutor) {
     this.useSignedStringMinMax = useSignedStringMinMax;
     this.useStatsFilter = useStatsFilter;
     this.useDictionaryFilter = useDictionaryFilter;
@@ -157,6 +201,7 @@ public class ParquetReadOptions {
     this.fileDecryptionProperties = fileDecryptionProperties;
     this.metricsCallback = metricsCallback;
     this.conf = conf;
+    this.vectoredReadExecutor = vectoredReadExecutor;
   }
 
   public boolean useSignedStringMinMax() {
@@ -195,6 +240,15 @@ public boolean useHadoopVectoredIo() {
     return useHadoopVectoredIo;
   }
 
+  /**
+   * Returns the caller-supplied vectored-read executor, or {@code null} to use the shared default.
+   *
+   * @return the caller-supplied executor, or {@code null}
+   */
+  public VectoredReadExecutor getVectoredReadExecutor() {
+    return vectoredReadExecutor;
+  }
+
   public FilterCompat.Filter getRecordFilter() {
     return recordFilter;
   }
@@ -211,6 +265,14 @@ public ByteBufferAllocator getAllocator() {
     return allocator;
   }
 
+  /**
+   * Returns the size in bytes used to split planned column-chunk ranges into read buffers.
+   * For vectored IO, this bounds each requested range length; filesystems may merge or
+   * checksum-align ranges and allocate larger buffers. This does not limit other allocations,
+   * such as footer reads or decoding, or total memory use.
+   *
+   * @return the configured read allocation size in bytes
+   */
   public int getMaxAllocationSize() {
     return maxAllocationSize;
   }
@@ -268,6 +330,7 @@ public static class Builder {
     protected FileDecryptionProperties fileDecryptionProperties = null;
     protected ParquetConfiguration conf;
     protected ParquetMetricsCallback metricsCallback;
+    protected VectoredReadExecutor vectoredReadExecutor;
 
     public Builder() {
       this(new HadoopParquetConfiguration());
@@ -338,6 +401,19 @@ public Builder withUseHadoopVectoredIo(boolean useHadoopVectoredIo) {
       return this;
     }
 
+    /**
+     * Sets the executor used for vectored reads. Share the same {@link VectoredReadExecutor}
+     * across readers which should share an admission limit. The caller owns the underlying
+     * executor; closing or aborting a reader never shuts it down.
+     *
+     * @param executor the caller-supplied executor, or {@code null} to use the shared default
+     * @return this builder
+     */
+    public Builder withVectoredReadExecutor(VectoredReadExecutor executor) {
+      this.vectoredReadExecutor = executor;
+      return this;
+    }
+
     public Builder useColumnIndexFilter(boolean useColumnIndexFilter) {
       this.useColumnIndexFilter = useColumnIndexFilter;
       return this;
@@ -405,6 +481,12 @@ public Builder withAllocator(ByteBufferAllocator allocator) {
       return this;
     }
 
+    /**
+     * Sets the read allocation size described by {@link ParquetReadOptions#getMaxAllocationSize()}.
+     *
+     * @param allocationSizeInBytes the maximum column-chunk read buffer or requested vectored range size in bytes
+     * @return this builder
+     */
     public Builder withMaxAllocationInBytes(int allocationSizeInBytes) {
       this.maxAllocationSize = allocationSizeInBytes;
       return this;
@@ -437,6 +519,7 @@ public Builder copy(ParquetReadOptions options) {
       useRecordFilter(options.useRecordFilter);
       withRecordFilter(options.recordFilter);
       withUseHadoopVectoredIo(options.useHadoopVectoredIo);
+      withVectoredReadExecutor(options.vectoredReadExecutor);
       withMetadataFilter(options.metadataFilter);
       withCodecFactory(options.codecFactory);
       withAllocator(options.allocator);
@@ -477,7 +560,8 @@ public ParquetReadOptions build() {
           properties,
           fileDecryptionProperties,
           metricsCallback,
-          conf);
+          conf,
+          vectoredReadExecutor);
     }
   }
 }