andygrove commented on PR #5265:
URL:
https://github.com/apache/datafusion-comet/pull/5265#issuecomment-5257129758
Thanks for tracking this one down. The diagnosis looks right to me. I traced
the shuffle path as well, and `CometShuffleExchangeExec.nativeChildContext`
matches on `child: CometNativeExec`, which `CometIcebergNativeScanExec`
satisfies directly, so `spark.table(t).repartition(...)` really was producing
`hasScanInput = false` before this change. The two reporting sites you
identified are the right ones.
A few things I would like your thoughts on.
### The comment overstates what CSV contributes
`CometCsvNativeScanExec` has no `metrics` override, so it inherits
`CometNativeExec.metrics`, which is `CometMetricNode.baselineMetrics` and only
carries `output_rows` and `elapsed_compute`. There is no `bytes_scanned`. The
native side matches, `native/core/src/execution/operators/csv_scan.rs`
registers no metrics at all.
So CSV scans still report nothing after this change. The
`scanLeaves.filter(_.metrics.contains("bytes_scanned"))` guard in
`reportScanInputMetrics` makes that a safe no-op, which the second half of your
comment does say. But naming `CometCsvNativeScanExec` in the list of scans that
"can contribute `bytes_scanned` / `output_rows`" reads like it is fixed here.
Could we drop CSV from that list, or note that it has no `bytes_scanned` metric
yet? The PR title and description carry the same implication.
### The gate now depends on a filter in another class
The comment justifies the broad match by saying `reportScanInputMetrics`
self-filters on `bytes_scanned`. That is accurate today, but it makes the
correctness of `hasScanInput` a function of an implementation detail in
`CometMetricNode`, with nothing linking the two. What do you think about
expressing the condition directly instead?
```scala
hasScanInput = sparkPlans.exists(p =>
p.isInstanceOf[CometLeafExec] && p.metrics.contains("bytes_scanned"))
```
That says what it means without the cross-class reasoning. It still picks up
contrib scans without a compile-time reference, which was the original
motivation for matching on the base class. It also avoids registering a
task-completion listener for plans where there is nothing to report, and it
would let the comment shrink to a line or two.
### Test duplication
These are good tests. Asserting the plan shape before the metrics is the
right call, and the note about pinning `COMET_SHUFFLE_MODE` rather than
trusting `auto` is appreciated.
Between these two and the existing `SELECT *` test at line 3580 there are
now three copies of the catalog config, the table DDL, the
`range(10000).repartition(5)` load, the listener, and the drain-clear-run-drain
sequence. Would it be worth pulling that into a helper that takes a table name
and a body? It would shrink the diff here and make the next metrics test much
cheaper to add.
### Listener asymmetry between the two new tests
The listener in the fused-block test filters on `if (im.bytesRead > 0)`
before recording, while the shuffle test records unconditionally. The
unconditional version is stronger, because a filtered listener cannot tell
"every task reported zero" apart from "no task ran". Both leave the buffer
empty. I realise the first one mirrors the existing test above it, but could
both new ones drop the filter?
### A follow-up, not something for this PR
`reportScanInputMetrics` uses `setBytesRead` / `setRecordsRead` rather than
the incrementing variants. So if a fallback Spark scan reaches a native block
through `CometSparkToColumnarExec` alongside a native scan, Comet's completion
listener overwrites whatever Spark's `FileScanRDD` accumulated for that task.
That is pre-existing behaviour for `CometNativeScanExec` and not introduced
here, but this change widens the set of plans that can hit it. Could you file a
tracking issue so it does not get lost? Happy to link it from here.
### CI
The head commit does not have any check runs yet. The previous commit was
green, 28 success and 2 skipped, and that run did include the `[scans]` bucket
that carries `CometIcebergNativeSuite` on Spark 4.1 and 4.2. It did not show
`[scans]` for Spark 3.4 or 3.5 though. Worth letting a full run finish on the
current head before merging.
---
I used an LLM to help work through this review. Please treat the points
above as a starting point rather than a checklist. You are much closer to this
code than I am, so use your judgement on which ones are worth acting on and
push back or just reply in the thread on any you disagree with.
--
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]
