Skip to content

Commit cbae849

Browse files
authored
Use configured copy buffer for Hadoop vectored reads (#15164)
Fixes #15163. ### Description When an optimized input-file implementation is not selected, remote Parquet reads use `HadoopInputFile` and the generic `RapidsInputFile.readVectored` fallback. Before NVIDIA/cudf-spark-jni#4765, that fallback used `HostMemoryBuffer.copyFromStream` with a 128 KiB internal copy chunk, while the pre-NVIDIA/cudf-spark#14674 Hadoop copy loop used `parquet.read.allocation.size` with an 8 MiB default. This PR overrides `HadoopInputFile.readVectored` to use the caller-supplied-buffer helper from NVIDIA/cudf-spark-jni#4765: - read the copy size from `parquet.read.allocation.size` - default to the JNI fallback size of 8 MiB - allocate a temporary buffer for each `readVectored` call - avoid a retained shared buffer or `ThreadLocal` state - leave `GpuParquetScan` on the `inputFile.readVectored` abstraction introduced by #14674 The measured regression was on the S3A fallback with PerfIO disabled. The same `HadoopInputFile` fallback is also used by GCS and other Hadoop-backed filesystems that do not provide their own optimized `readVectored` implementation. When S3 PerfIO is enabled, `S3InputFile.readVectored` continues to use the optimized PerfIO path and bypasses this fallback. Requires the JNI change from NVIDIA/cudf-spark-jni#4765, merged as `f5c95d88e16846a66a61870441e73c84863e706c`. The `release/26.06` base branch selects JNI `26.06.1-SNAPSHOT` through #15228. ### Testing The JNI change (`901ca4f570cb843c814121a7698ba8b3d5896fbf`) passed all eight `RapidsInputFileTest` tests. After installing that JNI artifact locally, the cudf-spark change commit (`ba4487c750ba73cf2f2281288ca319ad7f3752be`) was packaged on an x86 host with: ```text mvn -B -pl sql-plugin -am -DskipTests \ -Dspark-rapids-jni.version=26.06.1-SNAPSHOT package ``` Result: `BUILD SUCCESS`. ### Performance Validation Five interleaved full-NDS runs per configuration did not reproduce a stable performance regression after the fix, with either PerfIO disabled or enabled. See #15163 for the complete environment, per-run results, and analysis. ### Checklists Documentation - [ ] Updated for new or modified user-facing features or behaviors - [x] No user-facing change Testing - [ ] Added or modified tests to cover new code paths - [ ] Covered by existing tests - [x] Not required Performance - [ ] Tests ran and results are added in the PR description - [x] Issue filed with a link in the PR description - [ ] Not required Signed-off-by: Hongbin Ma (Mahone) <mahongbin@apache.org> Signed-off-by: Hongbin Ma <mahongbin@apache.org>
1 parent 00e8528 commit cbae849

1 file changed

Lines changed: 24 additions & 4 deletions

File tree

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

Lines changed: 24 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright (c) 2025, NVIDIA CORPORATION.
2+
* Copyright (c) 2025-2026, NVIDIA CORPORATION.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -16,14 +16,15 @@
1616

1717
package com.nvidia.spark.rapids.fileio.hadoop;
1818

19+
import ai.rapids.cudf.HostMemoryBuffer;
1920
import com.nvidia.spark.rapids.jni.fileio.RapidsInputFile;
2021
import com.nvidia.spark.rapids.jni.fileio.SeekableInputStream;
2122
import org.apache.hadoop.conf.Configuration;
22-
import org.apache.hadoop.fs.FileStatus;
2323
import org.apache.hadoop.fs.FileSystem;
2424
import org.apache.hadoop.fs.Path;
2525

2626
import java.io.IOException;
27+
import java.util.List;
2728
import java.util.Objects;
2829
import java.util.OptionalLong;
2930

@@ -34,21 +35,30 @@
3435
* for reading the file.
3536
*/
3637
public class HadoopInputFile implements RapidsInputFile {
38+
private static final String PARQUET_READ_ALLOCATION_SIZE = "parquet.read.allocation.size";
39+
3740
private final Path filePath;
3841
private final FileSystem fs;
42+
private final int copyBufferSize;
3943

4044
public static HadoopInputFile create(Path filePath, Configuration conf) throws IOException {
4145
Objects.requireNonNull(filePath, "filePath can't be null!");
4246
Objects.requireNonNull(conf, "Hadoop conf can't be null");
4347
FileSystem fs = filePath.getFileSystem(conf);
44-
return new HadoopInputFile(filePath, fs);
48+
int copyBufferSize = conf.getInt(PARQUET_READ_ALLOCATION_SIZE,
49+
RapidsInputFile.DEFAULT_READ_VECTORED_COPY_BUFFER_SIZE);
50+
return new HadoopInputFile(filePath, fs, copyBufferSize);
4551
}
4652

47-
private HadoopInputFile(Path filePath, FileSystem fs) {
53+
private HadoopInputFile(Path filePath, FileSystem fs, int copyBufferSize) {
4854
Objects.requireNonNull(filePath, "filePath can't be null!");
4955
Objects.requireNonNull(fs, "FileSystem can't be null");
56+
if (copyBufferSize <= 0) {
57+
throw new IllegalArgumentException(PARQUET_READ_ALLOCATION_SIZE + " must be positive");
58+
}
5059
this.filePath = filePath;
5160
this.fs = fs;
61+
this.copyBufferSize = copyBufferSize;
5262
}
5363

5464
@Override
@@ -70,4 +80,14 @@ public OptionalLong getLastModificationTime() throws IOException {
7080
public SeekableInputStream open() throws IOException {
7181
return new HadoopInputStream(fs.open(filePath));
7282
}
83+
84+
@Override
85+
public void readVectored(HostMemoryBuffer output, List<RapidsInputFile.CopyRange> copyRanges)
86+
throws IOException {
87+
if (copyRanges.isEmpty()) {
88+
return;
89+
}
90+
byte[] copyBuffer = new byte[copyBufferSize];
91+
RapidsInputFile.readVectoredUsingCopyBuffer(this, output, copyRanges, copyBuffer);
92+
}
7393
}

0 commit comments

Comments
 (0)