andygrove commented on PR #5453: URL: https://github.com/apache/datafusion-comet/pull/5453#issuecomment-5427512626
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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
