Use configured copy buffer for Hadoop vectored reads

[binmahone]

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 optimize iceberg read #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:

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
• No user-facing change

Testing

• Added or modified tests to cover new code paths
• Covered by existing tests
• Not required

Performance

• Tests ran and results are added in the PR description
• Issue filed with a link in the PR description
• Not required

[res-life]

LGTM

[sameerz]

@binmahone please retarget to release/26.06 as this is a regression and we will need to release a hotfix.

In addition, please add performance results to the description of the PR.

[binmahone]

Done. This PR targets release/26.06, and the description now includes the dependent JNI build verification, the five-pair PerfIO-disabled results, the PerfIO-enabled control results, and the required checklist.

The current GitHub compile failures are the expected dependency-order failure: CI resolves the published 26.06 JNI artifact, which does not yet contain DEFAULT_READ_VECTORED_COPY_BUFFER_SIZE or readVectoredUsingCopyBuffer. Packaging succeeds when the JNI PR snapshot is installed and selected.

[binmahone]

@gerashegalov Following up here on your caller-supplied-buffer suggestion in NVIDIA/cudf-spark-jni#4765: this PR does use RapidsInputFile.readVectoredUsingCopyBuffer(..., byte[]), but it intentionally allocates the byte[] for each HadoopInputFile.readVectored invocation rather than retaining it across invocations. This does not realize the cross-call allocation amortization you originally highlighted.

We considered retaining the buffer in HadoopInputFile, but RapidsInputFile requires implementations to be thread-safe. A shared mutable buffer would be unsafe for concurrent calls; synchronizing its use would serialize reads; and a ThreadLocal could retain up to the configured allocation size per thread. Truly amortizing the allocation therefore needs buffer ownership at a reader scope with a well-defined lifecycle, which is a broader change than this hotfix. The current implementation keeps the per-call lifetime while using the caller-supplied helper to honor parquet.read.allocation.size.

We validated this exact implementation against 26.04.2 with five interleaved full-NDS runs per configuration, both with PerfIO disabled and enabled. The results did not reproduce a stable performance regression after the fix. The complete environment and per-run results are in #15163.

[greptile-apps]

Greptile Summary

This PR overrides readVectored in HadoopInputFile to use a caller-supplied copy buffer, fixing a performance regression on S3A and other Hadoop-backed filesystems when S3InputFile's optimized PerfIO path is not active.

• Reads parquet.read.allocation.size (defaulting to RapidsInputFile.DEFAULT_READ_VECTORED_COPY_BUFFER_SIZE ≈ 8 MiB) at construction time and stores it as copyBufferSize, restoring the pre-regression copy chunk size that the Hadoop loop previously used.
• readVectored allocates a fresh byte[] per call and delegates to the new JNI static helper RapidsInputFile.readVectoredUsingCopyBuffer, deliberately avoiding a retained shared buffer or ThreadLocal state.
• The empty-ranges early return prevents an unnecessary stream open when there is nothing to read.

Confidence Score: 5/5

The change is a focused, well-scoped override that restores correct copy-chunk sizing for Hadoop-backed vectored reads with no resource management or logic concerns.

The override is small and self-contained: it reads a well-known Parquet config key at construction, guards against non-positive values, allocates a fresh byte array per call (intentionally avoiding shared state), and delegates to the new JNI static helper. The empty-ranges early return is correct. No GPU resources, no AutoCloseable ownership concerns on this side of the call, and no logic paths that deviate from the stated design.

No files require special attention; the only outstanding gate is the dependent JNI PR landing so the build can resolve the new constant and helper.

Important Files Changed

Filename	Overview
sql-plugin/src/main/java/com/nvidia/spark/rapids/fileio/hadoop/HadoopInputFile.java	Adds readVectored override that allocates a per-call copy buffer sized from parquet.read.allocation.size and delegates to the new JNI helper; adds copyBufferSize field with positive-value guard; removes unused FileStatus import.

Sequence Diagram

%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
    participant GpuParquetScan
    participant HadoopInputFile
    participant RapidsInputFile
    participant HadoopInputStream

    GpuParquetScan->>HadoopInputFile: readVectored(output, copyRanges)
    alt copyRanges is empty
        HadoopInputFile-->>GpuParquetScan: return
    else
        HadoopInputFile->>HadoopInputFile: new byte[copyBufferSize]
        HadoopInputFile->>RapidsInputFile: readVectoredUsingCopyBuffer(this, output, copyRanges, copyBuffer)
        RapidsInputFile->>HadoopInputFile: open()
        HadoopInputFile->>HadoopInputStream: new HadoopInputStream(fs.open(filePath))
        HadoopInputStream-->>RapidsInputFile: SeekableInputStream
        loop for each CopyRange
            RapidsInputFile->>HadoopInputStream: seek + read into copyBuffer
            RapidsInputFile->>RapidsInputFile: "copy copyBuffer -> HostMemoryBuffer(output)"
        end
        RapidsInputFile->>HadoopInputStream: close()
        RapidsInputFile-->>GpuParquetScan: done
    end

Loading

%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
    participant GpuParquetScan
    participant HadoopInputFile
    participant RapidsInputFile
    participant HadoopInputStream

    GpuParquetScan->>HadoopInputFile: readVectored(output, copyRanges)
    alt copyRanges is empty
        HadoopInputFile-->>GpuParquetScan: return
    else
        HadoopInputFile->>HadoopInputFile: new byte[copyBufferSize]
        HadoopInputFile->>RapidsInputFile: readVectoredUsingCopyBuffer(this, output, copyRanges, copyBuffer)
        RapidsInputFile->>HadoopInputFile: open()
        HadoopInputFile->>HadoopInputStream: new HadoopInputStream(fs.open(filePath))
        HadoopInputStream-->>RapidsInputFile: SeekableInputStream
        loop for each CopyRange
            RapidsInputFile->>HadoopInputStream: seek + read into copyBuffer
            RapidsInputFile->>RapidsInputFile: "copy copyBuffer -> HostMemoryBuffer(output)"
        end
        RapidsInputFile->>HadoopInputStream: close()
        RapidsInputFile-->>GpuParquetScan: done
    end

Loading

_{Reviews (2): Last reviewed commit: "Merge branch 'release/26.06' into fix-re..." | Re-trigger Greptile}

[greptile-apps]

Dependency on unpublished JNI artifact

Both RapidsInputFile.DEFAULT_READ_VECTORED_COPY_BUFFER_SIZE (line 49) and RapidsInputFile.readVectoredUsingCopyBuffer (line 91) are added in NVIDIA/cudf-spark-jni#4765, which has not yet been published in the 26.06 artifact. The PR description notes the two cannot find symbol compile errors are expected until that JNI change ships. This PR cannot be merged independently and must be gated on the JNI PR landing first.

[gerashegalov]

LGTM

[gerashegalov]

Non-blocking for a follow-up: HadoopInputFile could allocate min(copyBufferSize, largest requested range) like the JNI default path does, instead of always allocating copyBufferSize. That keeps the current per-call lifetime and thread-safety properties, while avoiding an 8 MiB allocation for small footer/range reads.

[binmahone]

build

[binmahone]

build

[yinqingh]

build

[binmahone]

build

[yinqingh]

build

[greptile-apps]

Want your agent to iterate on Greptile's feedback? Try greploops.

[Copilot]

Pull request overview

This PR addresses a remote Parquet scan performance regression on Hadoop-backed filesystems (e.g., S3A fallback, GCS) by making the HadoopInputFile vectored-read fallback use a configurable copy buffer size (via parquet.read.allocation.size) and delegating the actual copy loop to the newer JNI helper introduced in the referenced JNI change.

Changes:

• Adds a parquet.read.allocation.size-driven copy buffer size to HadoopInputFile creation (defaulting to the JNI-provided fallback size).
• Overrides HadoopInputFile.readVectored to allocate a per-call byte buffer and use RapidsInputFile.readVectoredUsingCopyBuffer(...) for bounded buffered reads.

[yinqingh]

build