Add named accumulators to track PerfIO S3 backend usage per executor

[zpuller]

Description

This change adds three named LongAccumulator instances to GpuTaskMetrics:

• perfio.s3.netty.executors — executors using the PerfIO Netty backend
• perfio.s3.crt.executors — executors using the PerfIO CRT backend
• perfio.s3.s3a.executors — executors falling back to S3A

Each accumulator is incremented at most once per executor per stage. The accumulator ID is used as a key in a per-JVM ConcurrentHashMap-backed set (PerfIO.reportedBackendAccIds) so that even if an executor reads many S3 parquet files in a stage, it contributes exactly 1 to the appropriate counter.

The accumulators appear in the stage Accumulables section of the Spark event log and History Server UI. On an N-executor cluster where all nodes use PerfIO Netty, perfio.s3.netty.executors will equal N after any stage that performs S3 parquet reads.

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

[greptile-apps]

Greptile Summary

Adds three named LongAccumulator instances (perfio.s3.netty.executors, perfio.s3.crt.executors, perfio.s3.s3a.executors) to GpuTaskMetrics to track which PerfIO S3 backend is active on each executor per stage. The accumulators are deduplicated via PerfIO.reportedBackendAccIds (a per-JVM ConcurrentHashMap-backed set keyed on accumulator ID) and are incremented at two call sites in GpuParquetScan: once during footer reads and once during remote block data copies — covering both the cache-miss footer path and the data-read path that was flagged in the prior review thread.

Confidence Score: 5/5

Safe to merge — no P0/P1 issues remain; prior review concerns are addressed.

All three previously-flagged concerns are resolved: the footer-cache-warm undercount is addressed by the second call site in copyRemoteBlocksData, the unregistered-accumulator throw is guarded by a try/catch, and the reportedBackendAccIds growth was acknowledged as acceptable. Remaining observations are purely P2 style (case-sensitivity of backend name matching, degenerate fully-cached stage where nothing fires). These do not block merge.

No files require special attention.

Important Files Changed

Filename	Overview
sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuTaskMetrics.scala	Adds three LongAccumulator fields and recordPerfioS3BackendOnce() with deduplication via PerfIO.reportedBackendAccIds; guards acc.id call against unregistered accumulators via try/catch.
sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/GpuParquetScan.scala	Adds two call sites to recordPerfioS3BackendOnce(): one in readFooterBuffer (footer cache miss path) and one in copyRemoteBlocksData (remote data copy path).

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A[S3 Parquet read task] --> B{fileIO is HadoopFileIO?}
    B -- No --> Z[Skip metric]
    B -- Yes --> C{filePath scheme starts with 's3'?}
    C -- No --> Z
    C -- Yes --> D[GpuTaskMetrics.get.recordPerfioS3BackendOnce]
    D --> E{PerfIO.s3BackendName}
    E -- netty --> F[perfioS3NettyExecutors]
    E -- crt --> G[perfioS3CrtExecutors]
    E -- other/s3a --> H[perfioS3S3aExecutors]
    F --> I{acc.id already in reportedBackendAccIds?}
    G --> I
    H --> I
    I -- Yes --> J[no-op: already counted this stage]
    I -- No --> K[acc.add 1]
    I -- IllegalArgumentException --> L[catch silently: unregistered accumulator backstop]
    subgraph Call sites
        CS1[readFooterBuffer footer cache-miss path]
        CS2[copyRemoteBlocksData remote data-copy path]
    end
    CS1 --> D
    CS2 --> D

Loading

_{Reviews (3): Last reviewed commit: "catch accumulator not yet registered for..." | Re-trigger Greptile}

[greptile-apps]

reportedBackendAccIds grows without bound

PerfIO.reportedBackendAccIds is a per-JVM static set that is only ever written to (add), never pruned. Each time a new GpuTaskMetrics instance is created (once per stage per query plan — see GpuTransitionOverrides), up to three new accumulator IDs can be added to this set. Over many queries in a long-running executor JVM the set grows monotonically. Each entry is a boxed Long in a ConcurrentHashMap, so the practical impact is low, but a cleanup hook keyed on stage completion (or resetting at plan-creation time) would prevent unbounded growth.

[zpuller]

At stage-level granularity this shouldn't be an issue

[zpuller]

build

[amahussein]

LGTM