feat: expose native Parquet scan I/O and read-amplification metrics - #5453
Open
sunchao wants to merge 12 commits into
Open
feat: expose native Parquet scan I/O and read-amplification metrics#5453sunchao wants to merge 12 commits into
sunchao wants to merge 12 commits into
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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:
Without an object-store-boundary measurement, both a genuinely efficient 128-byte read and this 524,416-byte read can present the same
bytes_scannedvalue. 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_scannedor adding per-row instrumentation.At the Parquet reader boundary,
scan_io_data_bytesmeasures returned projected data-page bytes, whilescan_io_metadata_bytesmeasures 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_readsandscan_io_footer_bytesfurther 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_callscounts GET operations after range coalescing,scan_io_object_store_get_requested_bytesrecords the coalesced ranges requested, andscan_io_object_store_response_bytes_readrecords 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
ObjectStoreAPI, 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_hitsandscan_io_metadata_cache_missesclassify successful metadata loads according to whether storage was actually read. For example: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::testsAll 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;
CometTaskMetricsSuitewas run with Spark 3.5, 4.0, and 4.2. ScalaStyle and Spotless checks were also run.