fix: report Input column when native Iceberg scan is enabled or native shuffle is enabled

[hsiang-c]

Which issue does this PR close?

Closes #5264

Rationale for this change

• Spark's UI shows 0.0B for the Input column of the executor tab.

What changes are included in this PR?

CometIcebergNativeScanExec is a sibling of CometNativeScanExec under CometLeafExec. So the old check only recognizes the Parquet leaf, and NativeExecContext.hasScanInput was false for Iceberg (and CSV, and contributed Delta) scans.

How are these changes tested?

Unit tests

• SELECT id+1 — scan fused under a native parent, parent's doExecuteColumnar → executeColumnarWithContext
• repartition(4, id) — scan inlined under the writer

Running the same job again with the patch, Input column show values now:

[andygrove]

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?

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.