andygrove commented on code in PR #6027:
URL: https://github.com/apache/datafusion-comet/pull/6027#discussion_r4051234032
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala:
##########
@@ -208,16 +212,49 @@ case class CometIcebergNativeScanExec(
override lazy val metrics: Map[String, SQLMetric] = {
val baseMetrics = Map(
"output_rows" -> SQLMetrics.createMetric(sparkContext, "number of output
rows"),
- "bytes_scanned" -> SQLMetrics.createSizeMetric(sparkContext, "number of
bytes scanned"))
+ "bytes_scanned" -> SQLMetrics.createSizeMetric(sparkContext, "number of
bytes scanned"),
+ // Native read/decode time, aggregated across tasks. Fed by
iceberg-rust's BaselineMetrics
+ // (elapsed_compute, in nanoseconds) through the standard JNI metric
path.
+ "elapsed_compute" -> SQLMetrics.createNanoTimingMetric(sparkContext,
"scan time"))
// Add num_splits as a runtime metric (incremented on the native side
during execution)
val numSplitsMetric = SQLMetrics.createMetric(sparkContext, "number of
file splits processed")
baseMetrics ++ icebergPlanningMetrics + ("num_splits" -> numSplitsMetric)
}
+ /**
+ * Posts the Iceberg planning metrics (data/delete file and manifest counts,
file sizes, and
+ * total planning duration) to the SQL UI as driver metrics. Iceberg-Java
produces these during
+ * planFiles(); they live on the driver and are never updated by executor
tasks, so without an
+ * explicit post the SQL UI never receives their accumulator ids and the
scan node shows
+ * nothing. Mirrors [[CometScanExec.sendDriverMetrics]] for the Parquet
path. The native runtime
+ * metrics (output_rows, bytes_scanned, num_splits, elapsed_compute) travel
the separate JNI
+ * path and are not posted here.
+ *
+ * Called from two places because a native leaf scan does not always run its
own
+ * doExecuteColumnar: when this scan is fused under a parent native operator
(e.g. a CometFilter
+ * for a pushed predicate), the parent runs the whole subtree as one RDD and
this node's
+ * doExecuteColumnar is never invoked. CometNativeExec.findAllPlanData walks
the subtree at
+ * execution time and reaches every leaf scan (calling this leaf lifecycle
hook alongside
+ * ensureSubqueriesResolved), so it calls this too. Re-posting the same
values is harmless.
Review Comment:
I do not think re-posting is harmless.
`SQLAppStatusListener.onDriverAccumUpdates` appends with
`exec.driverAccumUpdates ++ accumUpdates`, and `aggregateMetrics` appends each
driver value into that accumulator's value array, after which
`SQLMetrics.stringValue` returns `values.sum` for `SUM_METRIC`. So a second
post of the same (id, value) pair would double `totalDataManifest`,
`resultDataFiles` and the rest, and shift the total and the min/med/max shown
for the size and timing ones. Same behaviour on 3.4, 3.5 and 4.0.
I walked the call graph and could not find a path that fires both sites
today. This class overrides `doExecuteColumnar` so it never reaches
`buildNativeContext`, and `buildNativeContext` no-ops `CometNativeExec`
children when it builds `inputs`, so the fused case and the standalone case
really are exclusive. Nothing is broken right now.
But that is a quiet invariant to rest on for a hook whose whole point is
being callable from two places, and the precedent we are citing goes the other
way. `CometScanExec.sendDriverMetrics` posts from inside the `filePartitions`
lazy val and so runs exactly once, and `FileSourceScanExec` does the same.
Could we guard on the execution id in a follow-up, something like a `@transient
private var postedExecutionId` that skips when it matches? Then the comment
would be promising something that actually holds.
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -435,6 +435,11 @@ private[comet] object PlanDataInjector extends Logging {
// executeQuery, leaving DPP unresolved and forcing a sync-on-this
await inside
// the serializedPartitionData lazy val initializer (a known deadlock
surface).
iceberg.ensureSubqueriesResolved()
+ // Post the Iceberg planning metrics to the SQL UI here rather than
only from the scan's
+ // doExecuteColumnar: when the scan is fused under a parent native
operator, its own
+ // doExecuteColumnar never runs, so this walk is the only
execution-time hook that reaches
+ // it. Safe to also call for a root scan (re-posting the same values
is a no-op).
+ iceberg.sendDriverMetrics()
Review Comment:
Same point as on `CometIcebergNativeScanExec.sendDriverMetrics`. A second
post is not a no-op, Spark appends driver accumulator updates and sums them for
sum-typed metrics.
##########
spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala:
##########
@@ -2048,6 +2048,104 @@ class CometIcebergNativeSuite
}
}
+ test("Iceberg planning metrics and scan time are posted to the SQL UI") {
Review Comment:
Building on Sun Chao's point about the raw duration. I think this test stays
green if the Rust timer is removed. `createNanoTimingMetric` starts at -1, and
`CometMetricNode.set` writes whatever native reports including 0, which is
enough to move it off -1 and get an entry into the store, so
`uiValues.contains` passes either way. Could we add
`assert(metrics("elapsed_compute").value > 0)` to "verify all Iceberg planning
metrics are populated" instead? That test already has the table and the node
handle sitting next to the `bytes_scanned` and `num_splits` assertions, so it
would also save a second 10k-row Iceberg write in a suite that is already slow.
Separately, `WHERE id < 5000` is on a non-partition column, so Iceberg
leaves it in `postScanFilters` and the scan ends up fused under a
`CometFilterExec`. If that is right then only the `findAllPlanData` hook runs
here and `doExecuteColumnar` never does, so we are covering one of the two call
sites and cannot tell which from the assertions. A bare `SELECT *`, or a
partition-column predicate like the one "verify manifest pruning metrics" uses,
would give the other shape. An AQE variant would be worth having too. This
query has no exchange and no subquery so `InsertAdaptiveSparkPlan` skips it,
and fused plus DPP is the case the hook exists for.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala:
##########
@@ -208,16 +212,49 @@ case class CometIcebergNativeScanExec(
override lazy val metrics: Map[String, SQLMetric] = {
val baseMetrics = Map(
"output_rows" -> SQLMetrics.createMetric(sparkContext, "number of output
rows"),
- "bytes_scanned" -> SQLMetrics.createSizeMetric(sparkContext, "number of
bytes scanned"))
+ "bytes_scanned" -> SQLMetrics.createSizeMetric(sparkContext, "number of
bytes scanned"),
+ // Native read/decode time, aggregated across tasks. Fed by
iceberg-rust's BaselineMetrics
+ // (elapsed_compute, in nanoseconds) through the standard JNI metric
path.
+ "elapsed_compute" -> SQLMetrics.createNanoTimingMetric(sparkContext,
"scan time"))
Review Comment:
Could we get a `CometIcebergNativeScan` section into
`docs/source/user-guide/latest/metrics.md`? The only `scan time` row there
today is under `CometScanExec` and reads "Total time to scan a Parquet file",
and this one means something different. The timer wraps
`IcebergStreamWrapper::poll_next`, so it covers driving the reader plus schema
adaptation but drops anything that comes back `Pending`, which makes it decode
time rather than scan latency. Someone comparing the two numbers would not
guess that.
This is also the first time the Iceberg planning metrics are visible in the
UI at all, so a table covering those plus `bytes_scanned` and `num_splits`
would earn its keep. The `numDeletes` explanation you added just above reads
like it belongs in that doc too, it is more useful to users than to readers of
this file.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala:
##########
@@ -208,16 +212,49 @@ case class CometIcebergNativeScanExec(
override lazy val metrics: Map[String, SQLMetric] = {
val baseMetrics = Map(
"output_rows" -> SQLMetrics.createMetric(sparkContext, "number of output
rows"),
- "bytes_scanned" -> SQLMetrics.createSizeMetric(sparkContext, "number of
bytes scanned"))
+ "bytes_scanned" -> SQLMetrics.createSizeMetric(sparkContext, "number of
bytes scanned"),
+ // Native read/decode time, aggregated across tasks. Fed by
iceberg-rust's BaselineMetrics
+ // (elapsed_compute, in nanoseconds) through the standard JNI metric
path.
+ "elapsed_compute" -> SQLMetrics.createNanoTimingMetric(sparkContext,
"scan time"))
// Add num_splits as a runtime metric (incremented on the native side
during execution)
val numSplitsMetric = SQLMetrics.createMetric(sparkContext, "number of
file splits processed")
baseMetrics ++ icebergPlanningMetrics + ("num_splits" -> numSplitsMetric)
}
+ /**
+ * Posts the Iceberg planning metrics (data/delete file and manifest counts,
file sizes, and
+ * total planning duration) to the SQL UI as driver metrics. Iceberg-Java
produces these during
+ * planFiles(); they live on the driver and are never updated by executor
tasks, so without an
+ * explicit post the SQL UI never receives their accumulator ids and the
scan node shows
+ * nothing. Mirrors [[CometScanExec.sendDriverMetrics]] for the Parquet
path. The native runtime
+ * metrics (output_rows, bytes_scanned, num_splits, elapsed_compute) travel
the separate JNI
+ * path and are not posted here.
+ *
+ * Called from two places because a native leaf scan does not always run its
own
+ * doExecuteColumnar: when this scan is fused under a parent native operator
(e.g. a CometFilter
+ * for a pushed predicate), the parent runs the whole subtree as one RDD and
this node's
+ * doExecuteColumnar is never invoked. CometNativeExec.findAllPlanData walks
the subtree at
+ * execution time and reaches every leaf scan (calling this leaf lifecycle
hook alongside
+ * ensureSubqueriesResolved), so it calls this too. Re-posting the same
values is harmless.
+ */
+ override def sendDriverMetrics(): Unit = {
+ if (icebergPlanningMetrics.isEmpty) {
+ return
+ }
+ // Force planning so originalPlan.metrics are populated;
LazyIcebergMetric.value reads them.
+ val _ = serializedPartitionData
Review Comment:
Is this force needed? `postDriverMetricUpdates` reads `m.value`, and
`LazyIcebergMetric.value` already does `ensureSubqueriesResolved()` and then
forces `serializedPartitionData` itself. This line is the one variant that
forces planning without resolving DPP first, so dropping it would keep that
ordering in a single place.
--
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]
