Use configured copy buffer for Hadoop vectored reads

Signed-off-by: Hongbin Ma (Mahone) <mahongbin@apache.org>

Author: binmahone

sql-plugin/src/main/java/com/nvidia/spark/rapids/fileio/hadoop/HadoopInputFile.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2025, NVIDIA CORPORATION.
+ * Copyright (c) 2025-2026, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
 
 package com.nvidia.spark.rapids.fileio.hadoop;
 
+import ai.rapids.cudf.HostMemoryBuffer;
 import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile;
 import com.nvidia.spark.rapids.jni.fileio.SeekableInputStream;
 import org.apache.hadoop.conf.Configuration;
@@ -24,6 +25,7 @@
 import org.apache.hadoop.fs.Path;
 
 import java.io.IOException;
+import java.util.List;
 import java.util.Objects;
 import java.util.OptionalLong;
 
@@ -34,21 +36,30 @@
  * for reading the file.
  */
 public class HadoopInputFile implements RapidsInputFile {
+    private static final String PARQUET_READ_ALLOCATION_SIZE = "parquet.read.allocation.size";
+
     private final Path filePath;
     private final FileSystem fs;
+    private final int copyBufferSize;
 
     public static HadoopInputFile create(Path filePath, Configuration conf) throws IOException {
         Objects.requireNonNull(filePath, "filePath can't be null!");
         Objects.requireNonNull(conf, "Hadoop conf can't be null");
         FileSystem fs = filePath.getFileSystem(conf);
-        return new HadoopInputFile(filePath, fs);
+        int copyBufferSize = conf.getInt(PARQUET_READ_ALLOCATION_SIZE,
+                RapidsInputFile.DEFAULT_READ_VECTORED_COPY_BUFFER_SIZE);
+        return new HadoopInputFile(filePath, fs, copyBufferSize);
     }
 
-    private HadoopInputFile(Path filePath, FileSystem fs) {
+    private HadoopInputFile(Path filePath, FileSystem fs, int copyBufferSize) {
         Objects.requireNonNull(filePath, "filePath can't be null!");
         Objects.requireNonNull(fs, "FileSystem can't be null");
+        if (copyBufferSize <= 0) {
+            throw new IllegalArgumentException(PARQUET_READ_ALLOCATION_SIZE + " must be positive");
+        }
         this.filePath = filePath;
         this.fs = fs;
+        this.copyBufferSize = copyBufferSize;
     }
 
     @Override
@@ -70,4 +81,10 @@ public OptionalLong getLastModificationTime() throws IOException {
     public SeekableInputStream open() throws IOException {
         return new HadoopInputStream(fs.open(filePath));
     }
+
+    @Override
+    public void readVectored(HostMemoryBuffer output, List<RapidsInputFile.CopyRange> copyRanges)
+            throws IOException {
+        RapidsInputFile.readVectoredUsingCopyBuffer(this, output, copyRanges, copyBufferSize);
+    }
 }