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

[sunchao]

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. They are enabled only for recognized remote object-store schemes. Local filesystem reads, HDFS/custom backends, and ambiguous stores may still contribute reader-level data and metadata bytes, but they are not mislabeled as remote object-store traffic.

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.

The accounting also remains meaningful around less obvious lifecycle boundaries. Footer payloads are recorded once, including encrypted reads and valid footers followed by page-index failures; malformed or incompletely read footers are not reported as successful footer reads. Native producer shutdown is bounded so cancellation does not leave background work distorting published metrics, and object-store registrations remain isolated so different storage backends cannot be confused with each other.

How was this PR tested?

The native tests exercise the full accounting path rather than only checking that counters exist. They cover the exact 128-byte/524,416-byte coalescing example above, cold and warm metadata-only reads, projection and predicate pruning, page-index and Bloom-filter classification, local versus remote storage, encrypted and malformed footers, page-index failures, early producer termination, and object-store registration isolation.

On the published head:

cargo fmt --all -- --check
cargo test -p datafusion-comet --lib parquet::parquet_exec::tests

All 16 focused native scan tests passed. Native Rust library suites were also exercised with and without default features, together with both Clippy configurations and warnings denied.

Spark integration coverage verifies that all nine metrics reach the Spark SQL metric map, that reader-level counters are populated, and that remote object-store counters remain zero for local scans. Focused native-scan and collect-limit coverage was run with Spark 3.4, 3.5, 4.0, 4.1, and 4.2; CometTaskMetricsSuite was run with Spark 3.5, 4.0, and 4.2. ScalaStyle and Spotless checks were also run.

[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.