feat: expose native Parquet scan I/O and read-amplification metrics

[sunchao]

Which issue does this PR close?

Closes #5508.

The original-scheme routing fix (#5825) and deterministic backend/cache isolation and encryption URI fix (#5503, tracking #5502) have merged into main and are included in this rebase. Producer cancellation and final-snapshot waiting remain separate in #5505 (tracking #5504). This PR contains the scan metrics and their documentation/tests. The shutdown PR is independent of the metrics definitions.

Why are the changes needed?

Column projection and predicate pruning can make a Parquet scan appear inexpensive while the underlying storage still performs substantially more I/O. The scan needs more than the selected data pages: it may also fetch a footer, page indexes, and Bloom filters. Separately, the object-store reader can merge several small logical ranges into a much larger physical GET. Existing scan metrics do not distinguish these layers, so they cannot explain whether a slow or expensive scan is caused by actual data, metadata, range coalescing, or ineffective metadata caching.

Consider the deterministic range-coalescing case covered by this PR. The Parquet reader requests two 64-byte ranges, but the object-store layer combines them into one much larger GET:

Projected ranges requested by the Parquet reader:
  [0, 64) + [524352, 524416) = 128 bytes

Existing bytes_scanned:
  128 bytes

Actual coalesced ObjectStore GET:
  [0, 524416) = 524,416 bytes

Object-store response consumed:
  524,416 bytes

Observed read amplification:
  524,416 / 128 = 4,097x

Without an object-store-boundary measurement, both a genuinely efficient 128-byte read and this 524,416-byte read can present the same bytes_scanned value. Projection and predicate pushdown may therefore look effective while the expensive part of the read remains invisible.

Metadata creates a different blind spot. A metadata-only scan can read a footer, page indexes, or Bloom filters without returning a single projected data page. On the next scan, the same metadata may be served entirely from cache. Previously there was no reliable way to distinguish "no data pages were needed," "metadata still required storage I/O," and "metadata was already cached."

What changes were proposed in this PR?

The change introduces an end-to-end I/O accounting model with two deliberately different observation points: what the Parquet reader actually receives, and what a recognized remote object store actually services. These measurements are exposed through existing native execution metrics and propagated to Spark SQL metrics without changing the meaning of bytes_scanned or adding per-row instrumentation.

At the Parquet reader boundary, scan_io_data_bytes measures returned projected data-page bytes, while scan_io_metadata_bytes measures returned footer-prefetch, page-index, and Bloom-filter bytes. This separates useful projected data from the metadata needed to open and prune a file. scan_io_footer_reads and scan_io_footer_bytes further identify how often a serialized footer payload was actually read from storage and how large that payload was. Footer bytes are already included in metadata bytes; they are a more specific breakdown, not another category to add to the total.

At the remote object-store boundary, scan_io_object_store_get_calls counts GET operations after range coalescing, scan_io_object_store_get_requested_bytes records the coalesced ranges requested, and scan_io_object_store_response_bytes_read records response bytes as they are actually consumed. The coalescing example above therefore becomes directly observable: 128 reader-visible data bytes, one object-store GET, and 524,416 requested and consumed response bytes. Comparing object-store response bytes with projected data bytes reveals read amplification; comparing requested bytes with consumed bytes also distinguishes a fully consumed request from an early-terminated response.

This boundary is intentionally precise: the object-store metrics describe the ObjectStore API, not HTTP wire bytes, lower-level retries, compression, or transport implementation details. Their classification comes from the backend selected by object-store construction, using the pinned parser rather than a separate scheme allowlist. This includes native Azure azure/adl aliases. Local filesystem reads, in-memory stores, and HDFS/custom backends retain reader-level counters and have zero remote counters. The pinned parser does not accept the proposed gcs, wasb, wasbs, or s3n native aliases. The merged #5503 includes backend identity in the cache key, so a colliding cached store cannot contradict the selected classification.

At the metadata cache boundary, scan_io_metadata_cache_hits and scan_io_metadata_cache_misses classify successful metadata loads according to whether storage was actually read. For example:

First metadata-only scan, cold cache:
  data bytes = 0
  metadata bytes > 0
  footer reads = 1
  metadata cache misses = 1

Second metadata-only scan, warm cache:
  data bytes = 0
  metadata bytes = 0
  footer reads = 0
  metadata cache hits = 1

Together, these layers answer separate questions without double counting them: what reached the Parquet reader, what crossed the remote object-store API, and whether metadata access required storage at all. In particular, reader-level bytes and object-store bytes are alternative views of the same read path, not values that should be summed together. Metadata-only scans have no projected-data denominator, so their useful diagnostic is metadata and object-store traffic rather than an amplification ratio.

Footer accounting follows the metadata decoder protocol and is documented at the recording and read-method boundaries. A plaintext payload must decode; a subsequent page-index failure does not undo that footer read. A complete encrypted payload is counted before key retrieval, authentication, or metadata validation, so corrupt encrypted payloads can contribute footer bytes and a footer read. Incomplete payloads do not count. The new corrupt-authentication-tag regression makes this distinction explicit.

The native remote wrapper deliberately repeats the current cloud stores' default range coalescing so it can observe physical API requests. It therefore bypasses an inner get_ranges override; a future cache or custom implementation needs an explicit accounting contract. New recording-store tests check both that behavior and actual local/custom range delegation. This is documented as a composition constraint, not silently treated as universal support for arbitrary wrappers.

bytes_scanned, task inputMetrics.bytesRead, and scan_efficiency_ratio retain their existing meanings and blind spots. The new counters expose the missing I/O without redefining those existing metrics. The user guide now defines all nine counters, their overlap, supported backends, cache/encryption behavior, and the zero-data denominator. Cancellation can still leave late asynchronous updates outside the final snapshot; this PR does not promise complete traffic accounting after cancellation.

How was this PR tested?

September 10 rebase after #5503

Rebased both commits onto bdd4b90b49751b2ffedbddf87805b5f6311e9dfa, which includes #5825 and #5503. Resolved the overlaps in parquet_support.rs and parquet_exec.rs while preserving original-scheme routing, backend/configuration cache identity, deterministic registration, and encryption URI normalization. The Bloom-filter API correction is unchanged.

Andy’s routing concern is covered at both normalization and preparation boundaries. The normalization regression checks s3a and configured blob aliases with fs.comet.libhdfs.schemes=s3. The merged isolation tests now also assert the returned I/O classification: native S3 is Remote, Hadoop routing is Other, and native files are Local. The S3/Hadoop tests verify distinct cached stores and their contents in both registration orders. No isolation prerequisite remains outstanding.

Validation at 1cce3b2b4f9df7b1124aa8cc9f1297eb7c9e243f:

• 149 native Parquet tests passed, 1 ignored, using a freshly compiled test binary: cargo test --locked --no-default-features -p datafusion-comet --lib parquet::. This includes the original-routing and cache-classification regressions, coalescing/delegation, footer accounting, and encryption URI tests. The previously missing registry dependency is now available; no dependency manifests or lockfile changes were needed.
• The full Spark 4.1 JVM reactor passed production and test compilation, ScalaStyle, and Spotless with JDK 21: ./mvnw -B -ntp -Pspark-4.1 -DskipTests test-compile. This command did not execute Spark tests.
• Rust formatting and diff whitespace checks passed. Independent source review checked the range-diff, conflict resolutions, original-scheme routing, registration/encryption behavior, and all changed call signatures.
• Default-feature native tests and Spark runtime tests for this head require fresh hosted CI results.

The previous head’s macOS Spark 4.0 scans failure was a native crash after the fake-filesystem test passed. Symbolizing the exact consumed artifact resolves the saved return address to hdfsThreadDestructor + 0x50, matching the existing teardown issue #5023 with proposed fix #5036. This identifies the failure; it does not count as a passing macOS run.

September 10 rebase before #5503 (historical)

Rebased both commits onto 2d3eca2100d8d8684d31c496b376388a5fd79f18, including the original-scheme routing fix from #5825. The metrics classification now consumes the libhdfs decision carried by URL normalization. Added regression coverage for s3a and configured blob aliases with fs.comet.libhdfs.schemes=s3, and updated the preparation test to assert the returned Remote classification. The existing Bloom-filter API correction was preserved. #5503 had not yet merged at this validation point.

• The full Spark 4.1 JVM reactor passed production and test compilation, ScalaStyle, and Spotless with JDK 21: ./mvnw -B -ntp -Pspark-4.1 -DskipTests test-compile. Tests were not executed by this command.
• Rust formatting and diff whitespace checks passed. An independent review checked the range-diff, original-scheme routing, and all changed call signatures.
• Both routing/classification tests passed in an isolated Rust harness using the unchanged production helpers and tests with pinned object_store 0.13.2 and url 2.5.8. This is supplemental validation of those helpers, not a full native build.
• Native test compilation with cargo test --locked --no-default-features -p datafusion-comet --lib parquet::parquet_support::tests:: --no-run remains blocked before source compilation: the configured registry does not provide locked aws-smithy-runtime-api 1.16.0. Native manifests and the lockfile are unchanged from the new base. Full native tests and Spark tests using a newly built library require fresh CI validation.

September 9 rebase validation (historical)

Rebased the final reviewed patch onto 424c31aa79d13fddf743ffa29bae3c6f146e6c5e, consolidating the review iterations into one commit. The resolutions preserve DataFusion 55's concrete metadata cache and direct object-store reader APIs, the removed legacy JNI reader, Unicode name folding, and S3-compatible alias normalization. Both newer dynamic-filter test callers now pass the selected local backend. #5503 had not yet merged at this validation point.

• The full Spark 4.1.3 reactor passed production and test compilation, ScalaStyle, and Spotless with JDK 21: ./mvnw -Pspark-4.1 test-compile -DskipTests. This compiled the scan-metrics tests and accumulator benchmark; it did not execute tests.
• Rust formatting, metrics-document formatting, and diff whitespace checks passed. An independent source review checked the conflict resolutions, changed call signatures, and pinned Parquet 59.2/DataFusion 55 metadata-read contracts.
• Native validation was attempted with cargo test --locked -p datafusion-comet --lib parquet:: --no-run, but dependency resolution failed before source compilation: the local registry mirror did not provide aws-smithy-runtime-api 1.16.0, required by the existing lockfile. The native dependency manifests and lockfile are unchanged from the new base. Native compilation, native tests/Clippy, and Spark tests using a newly built native library remain unverified for this rebase and require fresh CI results.

August 27 validation (historical; predates the rebase)

For the August 27 review follow-up, the production native library was built with default features and a full JDK 21. All 100 native Parquet tests passed, including corrupt encrypted payloads, default remote coalescing, and actual custom/local range delegation. Native Clippy for the library and tests passed with warnings denied. Rust formatting, the changed metrics page's Prettier check, and the PR diff check passed.

The complete Spark 4.1.3 JVM reactor compiled and passed Spotless/ScalaStyle using the newly built metrics native library, whose copied resource was verified by SHA-256. Twelve Spark tests passed: native scan metric propagation, collect-limit behavior, and all ten CometTaskMetricsSuite cases. These are fresh tests of this revision's production code, not a reused unrelated native library.

The accumulator benchmark is part of CometReadBenchmark. It runs real Spark jobs with 10,000 tasks, comparing zero versus nine extra SQL accumulators (five size counters and four ordinary counters), and verifies all nine final values after each job. Each task adds 64 to each extra accumulator. Two fresh JVM runs reverse the case order; each case has warmup and three measured job iterations. Spark 4.1.3/JDK 17 ran on AMD EPYC-Milan with local[1], no competing workspace builds/tests, and no storage work.

Case order	Zero extra metrics: mean ± stddev	Nine extra metrics: mean ± stddev	Observed mean increase
Zero, then nine	7.410 ± 0.180 s	7.930 ± 0.036 s	0.520 s / 7.0%
Nine, then zero	7.357 ± 0.106 s	7.771 ± 0.113 s	0.414 s / 5.6%

That is approximately 41–52 microseconds per task on this deliberately tiny-task fixture. This measures additional accumulator serialization, scheduler reporting, and merging in a local job; it is not an end-to-end Parquet scan benchmark or a distributed-network result. Native counters, JNI traversal, SQL UI rendering, and storage I/O are excluded. The driver creates these accumulators per scan operator, not one new driver metric identity per task. A separate 1,024-operator creation loop observed best times of 4.19–4.43 microseconds per group of nine, with considerable GC variation in the averages. The in-process copy/merge loop is only a JVM microbenchmark and can benefit from escape analysis; its cost is not substituted for the real-task result.

The metrics remain enabled with the other scan metrics. This measurement shows a nonzero cost and does not claim that cost is free or that the same percentage applies to real scan tasks.

To reproduce the accumulator measurement from the repository root (after the normal release build prerequisites are available):

make benchmark-org.apache.spark.sql.benchmark.CometReadBenchmark -- --scan-metric-overhead
make benchmark-org.apache.spark.sql.benchmark.CometReadBenchmark -- --scan-metric-overhead --reverse-cases

The broader validation reported before this follow-up included native suites with and without default features and Spark 3.4/3.5/4.0/4.1/4.2 coverage. Those earlier cross-version results are historical; this follow-up reran the default-feature native tests and Spark 4.1.3 coverage described above. The extracted PRs describe their own current validation and cancellation measurements separately.

[andygrove]

This is a first pass review using an LLM. I will also review manually.

Thanks for this. The layering is well thought out and the split between what the reader receives and what a remote store actually services is genuinely useful. The description is the clearest explanation of Parquet read amplification I have seen in this repo.

My main request is that we split this into three PRs. The object store registry isolation in prepare_object_store_with_configs looks like an independent correctness fix rather than part of the metrics work. If s3 is in fs.comet.libhdfs.schemes and s3a is not, both s3a://bucket and s3://bucket collapse to the registry key s3://bucket, so a scan can end up reading through the wrong backend's store. That is a data path bug, and I would rather review it and think about backporting it on its own terms. The stop_batch_producer change in jni_api.rs is separable too. Splitting would also let us fill in the Closes # line for each, which is empty here.

On bytes_scanned, the TODO you removed in parquet_exec.rs said metadata I/O bypasses it, and this PR builds exactly the byte counting wrapper that TODO asks for, but the bytes land in scan_io_metadata_bytes and bytes_scanned is unchanged. That leaves three places still under-reporting. CometMetricNode.scala:71-73 feeds bytes_scanned into inputMetrics.bytesRead, scan_efficiency_ratio uses it as its numerator, and metrics.md describes it as "the truthful number you would see at the filesystem layer". Is leaving it alone a deliberate call to avoid changing an existing metric's meaning? If so that seems right to me, but could we say that in a comment where the TODO was, and fix the claim in metrics.md? As it stands, removing the TODO reads as if the gap is closed when bytesRead still misses footer and page index I/O.

Could you add the nine metrics to docs/source/user-guide/latest/metrics.md? It is hand maintained and already has a scan section. I would especially like the point from your description, that reader level bytes and object store bytes are alternative views of the same read path and must not be summed, written down there. The Spark UI puts them side by side with no hint of that.

Two things about the isolated registration URL. It is built from original_url.scheme(), so an s3a:// input becomes s3a+comet-<hash>-native://bucket and get_options then derives uri_base = "s3a://bucket/". In the non-isolated case the same table gives s3://bucket/, which is also what it gave before this PR, so the same encrypted table can get a different uri_base depending on whether isolation kicked in. Any KeyRetriever keyed on uri_base would resolve differently. Could we use the normalized scheme variable here instead? For HDFS backends the two are already the same, so nothing is lost. Separately, mangling only when a different Arc is already registered makes the resulting URL depend on which file happened to be planned first. Could we make it unconditional and derive it purely from (config_hash, backend) so the same inputs always give the same ObjectStoreUrl? That would also make the uri_base question go away on its own.

What does stop_batch_producer buy us? releasePlan is a JNI entry point, so this blocks the Spark task thread for up to 100ms on every plan release, and the full 100ms looks reachable since abort() only lands at the next await point and a Parquet decode can run a long way without one. Before this change the cleanup already happened on its own: Box::from_raw dropped the receiver, that closed the channel, and the producer's tx.send returned Err. The description says it stops background work from distorting published metrics, but the cost of not doing it is a slightly stale final counter push, and aborting mid-batch loses the in-flight metrics anyway. Do you have a measurement showing the accuracy gain is worth the teardown latency? Also, stops_finished_batch_producer_with_exhausted_runtime_budget suggests you were worried about this running on a tokio worker. Is that reachable? If it is, the thread::sleep there is parking a worker.

The scheme allowlist in scan_io_source is missing schemes that object_store::parse_url accepts, including azure, wasb, wasbs and adl. Those fall through to OtherObjectStore and silently lose the object store metrics. Since is_hdfs_object_store already tells us about the HDFS backend, could we invert the test and treat anything that is not file and not HDFS as a remote object store? Then the list cannot drift as object_store grows.

Could record_returned get a comment block explaining the footer protocol? Deferring record_footer until a later get_ranges asks only for ranges below footer_start is a nice way to say "the footer decoded, so it was real", but it is tightly coupled to the shape of the ParquetMetaDataPushDecoder loop in datafusion-datasource-parquet. If that fetch pattern changes upstream I would like the next person to have a chance of working out what broke, and the header comment on that file sets a high bar for this kind of explanation. One related detail: the description says malformed footers are not reported as successful footer reads, which holds on the plaintext path, but on the encrypted path record_footer_immediately records before validation is possible, so a corrupt encrypted footer does get counted.

The data versus metadata split relies on parquet-rs using get_byte_ranges for column chunks and get_bytes for Bloom filters and page indexes. That holds today but nothing pins it, and if a future version fetches a column chunk through get_bytes the amplification ratio goes quietly wrong with no test failing. Could we note the assumption at least? Classifying by comparing the requested range against the column chunk offsets from the metadata would be robust to whichever method upstream picks.

planner.rs:1631 and parquet/mod.rs:161 both re-parse the URL and call is_hdfs_scheme again when prepare_object_store_with_configs computes the same thing two lines later, so could it just return the flag? does_not_wait_indefinitely_for_blocked_batch_producer blocks a worker on the shared process-wide get_runtime() for 500ms via std::sync::mpsc::recv(), which will slow down anything else running in that test binary. And preserves_custom_hdfs_backend_range_reads_for_cloud_schemes only asserts scan_io_source classification, so the name promises range read behavior it does not check.

[sunchao]

Thanks, @andygrove ! splitting this makes sense. I’ll move the object-store isolation fix and producer-shutdown change into separate PRs with their own tracking issues, and keep this PR focused on scan I/O metrics.

Leaving bytes_scanned unchanged was deliberate, to preserve its existing semantics. I’ll explain that where the TODO was removed, correct the filesystem-level claim in metrics.md, and document all nine metrics, including which measurements overlap and must not be summed.

You’re right about the encrypted-footer wording. That path records a complete footer payload before decryption and validation, so the description overstates the guarantee. I’ll clarify the semantics and add coverage for corrupt encrypted footers.

I’ll also document the footer protocol and read-method assumptions, consolidate backend classification, and strengthen the HDFS test so it checks actual range-read delegation.

A few details from checking the implementation:

• The s3/s3a URI inconsistency is real, although CometFileKeyUnwrapper already normalizes both before key lookup. I still agree that normalized, deterministic registration would be cleaner.
• The pinned object_store recognizes azure and adl, but not wasb/wasbs. Comet’s native Azure integration supports abfs/abfss. I’ll keep classification aligned with backend construction; simply treating every non-file, non-HDFS store as remote would also include in-memory stores.
• The existing projection/pruning tests would catch a broad change that classified data reads as metadata, but I agree the dependency on upstream call patterns should be explicit.

For producer shutdown, the wait can improve the final metrics snapshot, but it does not guarantee complete accounting of in-flight work. I agree the latency tradeoff needs separate evidence. I’ll address that, along with the shared-runtime test concern, in the separate PR.

[andygrove]

Note on this review: this was generated by an LLM (Claude Code) at my request while I worked through a review backlog. I have not verified the individual findings myself. Please treat everything below as suggestions to evaluate rather than as authoritative review feedback, and push back on anything that is wrong or already handled.

This is a genuinely useful capability. Being unable to tell "128 bytes of data pages" apart from "a 512 KB coalesced GET" is a real gap when you are debugging a slow scan against object storage, and the layering into reader boundary, object-store boundary, and cache boundary is the right decomposition. The native test coverage is unusually thorough for a metrics change.

That said, I have several concerns that I think need resolving before this merges.

ScanIoObjectStore::get_ranges replaces the inner store's implementation

For the ObjectStore role, get_ranges calls coalesce_ranges(ranges, |range| self.get_range(location, range), OBJECT_STORE_COALESCE_DEFAULT) rather than delegating to self.inner.get_ranges. That is the same body as object_store's default trait method, so for stores that do not override it nothing changes. But any store that does override get_ranges loses its own implementation as soon as the wrapper is installed. Comet already has an object-store data cache in flight (#4828), and a caching store is exactly the kind of thing that wants to own get_ranges. This turns a metrics change into a change in read behavior, which I do not think we want.

I understand why it is written this way, since delegating to the inner store would hide the coalescing that the metrics exist to expose. Is there a way to instrument the coalescing without owning it? If not, this tradeoff should at least be spelled out in a comment at the get_ranges site, so that whoever adds a store with a custom get_ranges finds out before their implementation is silently bypassed.

get_opts collapses GetResultPayload::File into an in-memory stream

The File branch does result.bytes().await? and rewraps the whole thing as a single-chunk stream. That discards the file-handle payload, which is the mechanism that lets callers avoid buffering the whole object. The metadata-role wrapper is installed regardless of ScanIoSource, so this applies to local filesystem reads as well. Metadata is bounded so it is probably fine in practice, but "probably fine" plus "silently changes how local reads are serviced" is worth a comment explaining the bound, or a check that the range is small before buffering.

stop_batch_producer is unrelated to metrics and looks risky

Java_org_apache_comet_Native_releasePlan now aborts the producer task and busy-waits up to 100ms in 1ms sleeps on the calling JNI thread, then gives up and proceeds regardless. Two things I would like to understand.

What happens on the give-up path? The context is dropped shortly after, while a task that has been asked to abort but has not yielded may still hold references into it. If that is safe, why is it safe? If it is only safe because 100ms is empirically enough, that is a race, not a bound.

Second, why is this in a metrics PR at all? The description says cancellation should not leave background work distorting published metrics, but this is a change to plan teardown semantics and it deserves its own PR and its own review. Blocking a Spark task thread for up to 100ms per plan release is not free either. Could this be split out?

Nine always-on SQL metrics per scan

Every entry in the scan metric map becomes a driver-side accumulator per task and a line in the SQL UI node. Going from one bytes_scanned to ten scan-I/O metrics is a real cost on a job with tens of thousands of tasks, and it makes the scan node in the SQL UI considerably busier for every user, including the ones who will never look at read amplification. Was gating these behind a config considered? If they stay always-on, it would help to see a note in the description about the accumulator overhead measured on a job with a large task count.

docs/source/user-guide/latest/metrics.md is not updated

That page is hand-maintained and has a ScanExec section. Nine new user-facing metrics with names like scan_io_object_store_response_bytes_read need an explanation of what they mean, which ones are subsets of which (footer bytes inside metadata bytes), and the warning from the description that reader-level and object-store bytes are alternative views rather than values to add together. That last point in particular is going to be got wrong by users if it lives only in the PR description.

The remote-scheme allowlist will silently under-report

scan_io_source matches s3, s3a, gs, az, abfs, abfss, http, https. That leaves out gcs, azure, wasb, wasbs, s3n, and anything a user registers through a custom scheme, all of which fall to OtherObjectStore and report zeros for the object-store metrics. A user on wasbs:// sees three zero counters and has no way to tell that from "no coalescing happened". Would it be safer to invert this, so that everything except file and HDFS counts as remote? If the allowlist stays, the metrics doc should list which schemes are covered.

Encryption uri_base change

get_options now builds uri_base from physical_object_store_scheme plus the host and port slice, instead of object_store_url.to_string(). That changes the key the encryption factory is keyed on. It looks deliberate and connected to the +comet- scheme suffix, but it is a functional change to encryption in a PR about metrics, and I did not see a test that covers the before and after of the uri_base value specifically. Could you either call it out explicitly in the description with a test, or split it?

[sunchao]

Updated in 130ee02bb.

The two reviews are addressed in this revision as follows:

1. Split and issue references. This PR now contains the scan metrics, documentation, and their tests, closing Expose native Parquet scan I/O and read-amplification metrics #5508. Deterministic object-store registration and encryption URI normalization are in fix: isolate object-store registration by backend and configuration #5503, closing Avoid object-store cache and registry collisions across backends and configurations #5502. Producer cancellation and waiting are in fix: cancel background batch producers before final metrics #5505, closing Cancel background batch producers before collecting final plan metrics #5504. The metrics PR depends on fix: isolate object-store registration by backend and configuration #5503: the backend identity returned during construction must also be part of the cache identity, or an old cache collision could return the wrong kind of store. The shutdown change is independent.

2. bytes_scanned, task bytes, and efficiency. Keeping the existing meaning is deliberate. I restored an explanation beside the reader-factory setup, added the same qualification at the byte-read method, and corrected the filesystem-level claim in the guide. Footer/page-index loads bypass bytes_scanned, and coalescing can fetch more than its requested logical ranges. Task bytesRead and scan_efficiency_ratio still inherit those limitations; the new metrics do not silently redefine them.

3. The nine metrics and overlap. The metrics guide now defines every counter, explains footer bytes as a subset of metadata bytes, and says explicitly not to sum reader and object-store byte totals. It also covers cache eligibility, encrypted opens, metadata-only scans with no data denominator, backend scope, and incomplete cancellation snapshots.

4. Registration order and encryption URIs. fix: isolate object-store registration by backend and configuration #5503 derives isolated registration URLs deterministically from canonical backend/configuration identity from the first registration. Its tests use real distinct store contents and check both registration orders, configuration separation, cloud/HDFS alias routing, and ordinary/isolated encryption URI equivalence, including ports and custom schemes. Native s3a is canonicalized to s3. The previous direct URI test was preserved and strengthened there. CometFileKeyUnwrapper already normalizes the S3 aliases, so I am not describing the original spelling difference as a demonstrated failure of that implementation.

5. Shutdown latency, ownership, and test runtime. Those concerns are covered in fix: cancel background batch producers before final metrics #5505. The helper returns an explicit finished/timeout result, logs an incomplete snapshot on timeout, and tests use dedicated Tokio runtimes. A producer owns its stream and sender, not the raw execution-context pointer; that does not guarantee that all external task state remains usable after timeout. The 100 ms budget bounds the requested wait, not memory safety or arbitrary scheduler delays. Production release comes from the Spark executor thread; the exhausted-budget case is a synthetic helper test, not evidence that the JNI entry point normally runs on a Tokio worker. Local helper measurements show finished producers returning in microseconds, cooperative pending producers around one millisecond, and blocked work reaching the budget with incomplete counters. The separate PR reports the numbers and their limits; this metrics PR no longer claims complete accounting after cancellation.

6. Backend classification and duplicate parsing. prepare_object_store_with_configs now returns the selected backend classification with its URL/path. Both callers use it. The classification uses the pinned object-store parser, covering azure and adl, while explicitly retaining local, memory, and HDFS/custom categories. The pinned parser does not support native gcs, wasb, wasbs, or s3n; blanket inversion would also incorrectly label memory stores as remote. Tests check the accepted aliases and HDFS routing, and the guide lists the scope.

7. Footer protocol and corrupt encryption. Comments now explain the tail/payload protocol, when the plaintext decoder can advance to indexes, and why a later index failure does not erase a footer read. They also explain that a complete encrypted payload is counted before key retrieval/authentication. A new regression corrupts the encrypted footer's authentication tag and verifies that its completed I/O is still counted when authentication fails. The description no longer claims every malformed encrypted footer is excluded.

8. Data versus metadata methods. The code documents the pinned Parquet contract: get_bytes for Bloom filters, get_byte_ranges for data pages, and separate metadata loading for footers/indexes. Existing projection/pruning and Bloom-filter assertions already catch a broad reclassification; the dependency is now explicit rather than relying only on those tests. I did not replace it with an unvalidated offset classifier.

9. Custom get_ranges behavior. New tests perform actual reads through a recording store, checking returned bytes and which methods were called. Local/custom paths retain delegation. The native remote case intentionally exercises default coalescing and confirms the inner override is bypassed. A comment at that branch and the guide call out the composition constraint for a future data cache. Simply forwarding the call would hide the physical API requests this metric is intended to observe.

10. The File payload branch. That branch can buffer GetResult.range, not necessarily a whole file. In the pinned local metadata path, get_ranges delegates to LocalFileSystem.get_ranges and does not reach the wrapper's get_opts branch; local data reads also bypass the remote-role wrapper. The new delegation tests protect that path. I did not add an arbitrary range limit without a reachable local-scan regression.

11. Accumulator overhead and a config. Driver metric identities are created per scan operator; task copies and reporting still scale with task count. The new benchmark runs actual 10,000-task Spark jobs with zero/nine extra SQL accumulators, reverses the case order in a second JVM, and verifies all final values. Mean time increased from 7.410 to 7.930 seconds in one pass and 7.357 to 7.771 seconds in the other: about 41–52 microseconds per task, or 5.6–7.0% on this deliberately tiny-task local[1] fixture. This includes task serialization and scheduler updates, but excludes native counters, JNI, distributed networking, storage, and UI rendering. The description includes standard deviations, the separate driver-creation measurement, and reproduction commands. The metrics remain enabled alongside the other scan metrics; I am not claiming they are free or that this percentage predicts scan-query overhead.

Validation for the follow-up: 100 native Parquet tests and 12 Spark 4.1.3 tests passed, including all ten task-metrics tests, with the newly built metrics native library. The full JVM reactor, ScalaStyle/Spotless, native library/test Clippy with warnings denied, Rust formatting, and metrics-page formatting passed. Broader cross-version results are explicitly labeled as earlier validation rather than reruns of this revision.

[sunchao]

Two current CI failures at 130ee02bb have different causes.

The Linux Spark 3.5 scans job stopped during Maven dependency resolution with HTTP 429 for org.apache:apache:pom:23, before project compilation or tests.

The macOS Spark 4.0 scans job reached the tests, then suffered a native SIGSEGV just after native scan on fake fs passed. I verified the exact downloaded native artifact's digest and symbolicated the saved return address to hdfsThreadDestructor + 0x50. This matches the existing libhdfs teardown bug in #5023, with a fix proposed in PR #5036. The HDFS/JNI teardown path, fake-filesystem suite and dependency versions are unchanged from the pinned base ec0f7975.

The macOS crash is not a passing test result, and I have not claimed a clean macOS run. The parent workflow is still active. The fresh local native and Spark metrics results in the description remain separate from these hosted checks.

[andygrove]

The footer accounting, the encrypted-footer disclosure, and the scheme classification all address what I raised earlier. The comment block above record_returned in eager_page_index_reader_factory.rs documents the tail/payload protocol well enough that a future reader doesn't have to re-derive it from the ParquetMetaDataPushDecoder loop. counts_complete_encrypted_footer_even_when_authentication_fails covers the case where a corrupted encrypted footer's I/O still counts before authentication runs, which matches the corrected wording in metrics.md.

Moving the backend classification into object_store_backend in parquet_support.rs, keyed off ObjectStoreScheme::parse instead of a hand-rolled scheme list, resolves the drift concern too. I checked that the schemes it can't classify, gcs, wasb, wasbs, and s3n, are exactly the ones that already fail a few lines later when prepare_object_store_with_configs tries to build the real store through the same parser. So there's no case where a scan works but silently loses its object-store metrics.

metrics.md now covers all nine counters, says plainly that reader-level and object-store bytes are alternative views that should not be summed, and gives the zero-denominator guidance for metadata-only scans. I traced through the case where a remote scan's metadata fetch wraps an already ObjectStore-role store, since that is where I would expect double counting to show up, and it composes correctly. The outer wrapper records logical bytes once per range, the inner one records the coalesced physical GET once, and neither adds the other's bytes into its own total.

Splitting the object-store isolation fix and stop_batch_producer into separate PRs was the right call, and neither shows up in this diff anymore.

[andygrove]

@sunchao this is conflicting with main now. Could you rebase? Everything was green beforehand.

[sunchao]

Just rebased @andygrove

[andygrove]

The rebase brought in more than a replay, and one piece of it interacts with the feature this PR adds.

Against 130ee02bb, the head I approved, the metrics work itself is unchanged. What moved is inherited: Arc<FileMetadataCache> instead of Arc<dyn FileMetadataCache>, and prepare_object_store_with_configs now opens with main's

let url = normalize_object_store_url(url.as_str(), object_store_configs)?;
let is_hdfs_scheme = is_hdfs_scheme(&url, object_store_configs);

That ordering, from #5314, classifies libhdfs routing against the rewritten scheme, so an s3a:// URL with fs.comet.libhdfs.schemes=s3 and no s3a entry normalizes to s3:// and then matches the list. I confirmed it on 424c31aa7 rather than reading it off the diff:

PROBE typed scheme      = s3a
PROBE typed  is_hdfs    = false
PROBE normalized scheme = s3
PROBE final  is_hdfs    = true

I filed that as #5816, and it turned up while chasing the failure on #5503, whose isolates_backends_even_when_s3_alias_and_configs_match covers exactly that pair.

The reason it matters here rather than only there is that object_store_backend takes is_hdfs as its first, precedence-taking input:

if is_hdfs {
    return Ok(ObjectStoreBackend::Other);
}

So under that configuration an s3a:// scan is classified Other while it is in fact served by the native S3 store, and every counter gated on Remote reads zero. That is the read-amplification view this PR exists to provide, going silently absent for a real S3 configuration, and it is exactly the failure mode your own comment warns against: "Callers must use the returned backend classification rather than infer it from an original scheme alias." The classification is right to be centralised; its is_hdfs input is what is wrong.

This does not change my view of the metrics work, and I am not asking you to widen this PR's scope. But since #5453 depends on #5503 by your own note, and #5503's test is the natural regression guard for #5816, could the ordering fix land in #5503 and this rebase onto it? Then ObjectStoreBackend here is fed a correct decision on arrival rather than inheriting a wrong one. If you would rather keep them independent, a Remote versus Other assertion for s3a under fs.comet.libhdfs.schemes=s3 in this PR's tests would at least fail loudly instead of reporting zeros.

My approval stands for everything else. metrics.md, the nine counters and the double-counting note are all as I reviewed them.