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

docs/source/user-guide/latest/metrics.md

@@ -79,11 +79,50 @@ Here is a guide to some of the native metrics.
 | `spilled_bytes`        | Actual bytes written to native shuffle spill files on disk.           |
 | `memory_spilled_bytes` | Uncompressed Arrow backing-buffer and partition-index memory spilled. |
 
+### Native Parquet scans
+
+Native Parquet scans expose these counters in the Spark SQL metric map as well as native
+execution metrics. Counters accumulate per scan operator; they do not instrument individual rows.
+
+| Metric                                     | Description                                                                                                                                                                                                                  |
+| ------------------------------------------ | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `scan_io_data_bytes`                       | Bytes returned to the Parquet reader for projected data-page ranges.                                                                                                                                                         |
+| `scan_io_metadata_bytes`                   | Bytes returned for footer prefetches, page indexes, and Bloom filters. A footer prefetch can also contain unused data bytes.                                                                                                 |
+| `scan_io_footer_reads`                     | Storage reads of complete serialized footer payloads, counted once per metadata open. Plaintext payloads must decode successfully; encrypted payloads are counted before key retrieval/decryption, even if those later fail. |
+| `scan_io_footer_bytes`                     | Serialized footer payload bytes, excluding the final eight-byte trailer. These bytes are already included in `scan_io_metadata_bytes`.                                                                                       |
+| `scan_io_object_store_get_calls`           | Nonempty GET operations at the native remote `ObjectStore` API, after range coalescing. Does not count transport retries or HEAD requests.                                                                                   |
+| `scan_io_object_store_get_requested_bytes` | Requested coalesced range bytes at that API. A failed or partly consumed request can contribute requested bytes without the same number of response bytes.                                                                   |
+| `scan_io_object_store_response_bytes_read` | Response bytes actually consumed at that API, including bytes fetched between coalesced ranges. Not HTTP wire bytes.                                                                                                         |
+| `scan_io_metadata_cache_hits`              | Successful, cache-eligible metadata opens requiring no storage reads.                                                                                                                                                        |
+| `scan_io_metadata_cache_misses`            | Successful, cache-eligible metadata opens requiring storage reads. Failed opens and encrypted opens, which bypass this shared cache, increment neither cache counter.                                                        |
+
+Reader-level and object-store bytes are two views of the same reads; do not add them together.
+Likewise, footer bytes are a subset of metadata bytes, not a third reader-level category. A warm
+metadata-cache hit contributes no new metadata or footer I/O. A valid plaintext footer followed by
+a page-index failure still contributes footer bytes; an invalid plaintext footer does not.
+
+Remote counters follow the backend selected during object-store construction, including native
+S3 (`s3`/`s3a`), GCS (`gs`), Azure (`az`, `adl`, `azure`, `abfs`, `abfss`), and HTTP(S) stores.
+Local files, in-memory stores, and HDFS/custom backends (including cloud-looking schemes selected
+through `fs.comet.libhdfs.schemes`) retain reader-level counters but have zero remote counters.
+The native cloud wrapper observes default range coalescing; custom `get_ranges` implementations
+require an explicit accounting contract before being composed with that wrapper.
+
+For data-bearing scans, comparing remote response bytes with reader data bytes can reveal
+coalescing and metadata overhead. For a metadata-only scan, data bytes are zero: report the
+metadata and remote totals instead of dividing by zero. Cancellation can leave late asynchronous
+work outside the final metric snapshot; these counters are not a guarantee of complete network
+traffic accounting after cancellation.
+
 ## Task-Level Input Metrics on Spark 4.1+
 
-Comet's native scans set `inputMetrics.bytesRead` to the actual file IO performed by the
-DataFusion parquet reader (`bytes_scanned`). This is the truthful number you would see at the
-filesystem layer.
+Comet's native scans populate `inputMetrics.bytesRead` from the existing `bytes_scanned`
+counter. It counts requested data/Bloom-filter ranges through the Parquet reader's byte-read
+methods, not all filesystem I/O. Footer and page-index reads through metadata loading bypass
+this counter, and range coalescing can fetch more bytes than the logical ranges request. The
+additional scan I/O metrics above expose those differences without changing `bytes_scanned`.
+The native `scan_efficiency_ratio` still uses `bytes_scanned` as its numerator and has the same
+blind spots; it is not the remote read-amplification ratio described above.
 
 Spark 4.1 changed its own parquet reader to pre-open the `SeekableInputStream` and read the file
 footer outside the `FileScanRDD.compute()` thread. Spark's `inputMetrics.bytesRead` is updated
@@ -100,4 +139,5 @@ unaffected and remains exactly equal between Comet and Spark.
 
 If you compare Comet's `bytesRead` against vanilla Spark's on Spark 4.1+ (via the Spark UI or
 the REST API), expect Comet's number to be substantially larger for small files, and closer to
-Spark's for large files. Comet's value reflects what the storage layer actually delivered.
+Spark's for large files in that workload. Neither metric should be interpreted as complete
+filesystem or network traffic accounting.

native/core/src/execution/operators/parquet_writer.rs

@@ -311,7 +311,7 @@ impl ParquetWriterExec {
             #[cfg(feature = "hdfs-opendal")]
             {
                 // Use prepare_object_store_with_configs to create and register the object store
-                let (_object_store_url, object_store_path) = prepare_object_store_with_configs(
+                let (_object_store_url, object_store_path, _) = prepare_object_store_with_configs(
                     _runtime_env,
                     output_file_path.to_string(),
                     object_store_options,

native/core/src/execution/planner.rs

@@ -1631,11 +1631,12 @@ impl PhysicalPlanner {
                     .iter()
                     .map(|(k, v)| (k.clone(), v.clone()))
                     .collect();
-                let (object_store_url, _) = prepare_object_store_with_configs(
-                    self.session_ctx.runtime_env(),
-                    one_file,
-                    &object_store_options,
-                )?;
+                let (object_store_url, _, object_store_backend) =
+                    prepare_object_store_with_configs(
+                        self.session_ctx.runtime_env(),
+                        one_file,
+                        &object_store_options,
+                    )?;
 
                 // Get files for this partition
                 let files = self.get_partitioned_files(partition_files)?;
@@ -1646,6 +1647,7 @@ impl PhysicalPlanner {
                     Some(data_schema),
                     Some(partition_schema),
                     object_store_url,
+                    object_store_backend,
                     file_groups,
                     Some(projection_vector),
                     Some(data_filters?),
@@ -1683,7 +1685,7 @@ impl PhysicalPlanner {
                     .and_then(|f| f.partitioned_file.first())
                     .map(|f| f.file_path.clone())
                     .ok_or(GeneralError("Failed to locate file".to_string()))?;
-                let (object_store_url, _) = prepare_object_store_with_configs(
+                let (object_store_url, _, _) = prepare_object_store_with_configs(
                     self.session_ctx.runtime_env(),
                     one_file,
                     &object_store_options,