parthchandra opened a new pull request, #6085:
URL: https://github.com/apache/datafusion-comet/pull/6085
## Which issue does this PR close?
Follow-up to #6027, addressing review comments from @andygrove and @sunchao.
#### Rationale for this change
#6027 added Iceberg planning metrics and a native scan-time metric to the
Spark UI. Review raised three things worth fixing:
- The claim that re-posting the driver metrics is harmless is not correct.
Spark's SQLAppStatusListener.onDriverAccumUpdates appends driver updates and
aggregateMetrics sums them for SUM-typed metrics, so posting the same values
twice would double totalDataManifest, resultDataFiles, and the rest, and skew
the size/timing min/med/max. No path posts twice today (the fused and
standalone execution paths are mutually exclusive), but that was an unenforced
invariant for a hook whose whole point is being callable from two places.
- The new metrics were not documented, and the native scan-time metric means
something different from the existing scan time under CometScanExec.
- The scan-time test passed even if the native timer was removed, and it
only exercised one of the two call sites.
#### What changes are included in this PR?
1. Guard against double-posting. sendDriverMetrics now records the execution
id it posted under (@transient private var postedExecutionId) and returns early
if asked to post again for the same execution. A second call is now a real
no-op instead of relying on the two call sites staying exclusive. Corrected the
"harmless" comments in CometIcebergNativeScanExec and
PlanDataInjector.findAllPlanData.
2. Removed a redundant planning force. Dropped val _ =
serializedPartitionData from sendDriverMetrics. postDriverMetricUpdates reads
each metric's value, and LazyIcebergMetric.value already resolves DPP
subqueries and then forces serializedPartitionData. Removing the extra force
keeps the "resolve DPP before planning" ordering in one place.
3. Documented the metrics. Added a CometIcebergNativeScan section to
docs/source/user-guide/latest/metrics.md: a table for the runtime metrics
(output rows, bytes scanned, file splits, scan time) and one for the Iceberg
planning metrics. The scan-time row notes it is decode/compute time from the
native poll (reader plus schema adaptation), not end-to-end latency, and that
it differs from CometScanExec's scan time. Moved the numDeletes explanation
into that doc.
4. Strengthened the tests.
- Added assert(metrics("elapsed_compute").value > 0) to "verify all
Iceberg planning metrics are populated" so the test fails if the native timer
is removed. (The presence-only check passed at 0 because createNanoTimingMetric
starts at -1 and any set, even 0, moves it off -1.)
- Refactored the status-store test to run against both the fused query
(predicate on a non-partition column, so only the findAllPlanData hook posts)
and a bare SELECT * (standalone, so the scan's own doExecuteColumnar posts),
covering both call sites.
#### How are these changes tested?
- CometIcebergNativeSuite on Spark 4.0 and 4.1 (the modified suite).
- Verified in spark-shell on a 20M-row Iceberg table that the planning
metrics match Iceberg-Java and are not doubled (resultDataFiles=16, not 32),
and that scan time is non-zero (~5.3 s across tasks).
--
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]
