diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index 7c3c22ffa60..fdcd6c8ba5a 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -435,6 +435,13 @@ where type Item = DFResult; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + // Time the whole poll (driving the inner reader plus schema adaptation) as elapsed_compute. + // record_poll only records output rows; without this explicit timer elapsed_compute stays + // 0, so the Spark "scan time" metric never moves. Clone the Time metric (Arc-backed) so the + // guard does not hold a borrow of self across the inner poll below. + let elapsed_compute = self.baseline_metrics.elapsed_compute().clone(); + let _timer = elapsed_compute.timer(); + let poll_result = self.inner.poll_next_unpin(cx); let result = match poll_result { diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala index d02cc658ed4..65cbe300382 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala +++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometIcebergNativeScanExec.scala @@ -27,6 +27,7 @@ import org.apache.spark.sql.SparkSession import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression, SortOrder} import org.apache.spark.sql.catalyst.plans.QueryPlan import org.apache.spark.sql.catalyst.plans.physical.{Partitioning, UnknownPartitioning} +import org.apache.spark.sql.execution.SQLExecution import org.apache.spark.sql.execution.datasources.v2.BatchScanExec import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics} import org.apache.spark.sql.vectorized.ColumnarBatch @@ -193,7 +194,10 @@ case class CometIcebergNativeScanExec( } else { originalPlan.metrics .filterNot { case (name, _) => - // Filter out metrics that are now runtime metrics incremented on the native side + // Filter out metrics that are now runtime metrics incremented on the native side. + // numDeletes is a Java-reader runtime metric (deletes applied at read time); the native + // path never runs that reader and iceberg-rust exposes no deletes-applied count, so it + // would always read 0. Drop it rather than surface a misleading zero. name == "numOutputRows" || name == "numDeletes" || name == "numSplits" } .map { case (name, metric) => @@ -208,7 +212,10 @@ 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") @@ -216,8 +223,38 @@ case class CometIcebergNativeScanExec( 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 + val executionId = sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY) + SQLMetrics.postDriverMetricUpdates( + sparkContext, + executionId, + icebergPlanningMetrics.values.toSeq) + } + /** Executes using CometExecRDD - planning data is computed lazily on first access. */ override def doExecuteColumnar(): RDD[ColumnarBatch] = { + sendDriverMetrics() val nativeMetrics = CometMetricNode.fromCometPlan(this) val serializedPlan = CometExec.serializeNativePlan(nativeOp) // Key by the same (metadata_location, scan_hash_code) pair PlanDataInjector.injectPlanData diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala b/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala index d6bd2fe1bff..fe1a2e637ad 100644 --- a/spark/src/main/scala/org/apache/spark/sql/comet/operators.scala +++ b/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() if (iceberg.commonData.nonEmpty && iceberg.perPartitionData.nonEmpty) { // A self-join/self-merge can put two scans of the same table (same metadata_location) // in one native plan. Computing the key via IcebergPlanDataInjector.getKey, the same @@ -460,6 +465,7 @@ private[comet] object PlanDataInjector extends Logging { // directly -- there is no silent "not a leaf" skip. case s: CometLeafExec with CometScanWithPlanData => s.ensureSubqueriesResolved() + s.sendDriverMetrics() if (s.commonData.nonEmpty && s.perPartitionData.nonEmpty) { (Map(s.sourceKey -> s.commonData), Map(s.sourceKey -> s.perPartitionData)) } else { @@ -1194,6 +1200,15 @@ abstract class CometLeafExec extends CometNativeExec with LeafExecNode { prepare() waitForSubqueries() } + + /** + * Posts any driver-side SQL metrics this leaf scan computes (e.g. Iceberg planning metrics) to + * the SQL UI. Like [[ensureSubqueriesResolved]], it is called from + * [[PlanDataInjector.findAllPlanData]] at execution time, which is the only hook that reaches a + * leaf scan fused under a parent `CometNativeExec` (its own `doExecuteColumnar` never runs). + * Default is a no-op; leaf scans with driver-computed metrics override it. + */ + def sendDriverMetrics(): Unit = {} } /** diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 20cf045a2f5..9ceebfe6a8d 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/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") { + assume(icebergAvailable, "Iceberg not available in classpath") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + spark.sql(""" + CREATE TABLE test_cat.db.driver_metrics_test ( + id INT, + value DOUBLE + ) USING iceberg + """) + spark + .range(10000) + .selectExpr("CAST(id AS INT)", "CAST(id * 1.5 AS DOUBLE) as value") + .coalesce(1) + .write + .format("iceberg") + .mode("append") + .saveAsTable("test_cat.db.driver_metrics_test") + + val df = spark.sql("SELECT * FROM test_cat.db.driver_metrics_test WHERE id < 5000") + val scanNodes = df.queryExecution.executedPlan + .collectLeaves() + .collect { case s: CometIcebergNativeScanExec => s } + assert(scanNodes.nonEmpty, "Expected a CometIcebergNativeScanExec node") + + df.collect() + CometListenerBusUtils.waitUntilEmpty(spark.sparkContext) + + // Read the metrics the Spark SQL UI actually renders, from the status store. It resolves + // metric name -> accumulator id -> final aggregated value the way the UI does, so it does + // not depend on which scan node instance executed. Reading SQLMetric.value on the node (as + // other tests do) works for the driver-computed value but does not prove the value reaches + // the UI, which is exactly the gap this fix closes. + val store = spark.sharedState.statusStore + + // Of the CREATE / INSERT / SELECT executions, the SELECT is the one whose scan node + // declares the Iceberg planning metrics. + val candidateExecIds = store + .executionsList() + .map(_.executionId) + .filter(id => + store.execution(id).exists(_.metrics.exists(_.name == "totalDataManifest"))) + assert( + candidateExecIds.nonEmpty, + "no SQL execution exposed the Iceberg scan planning metrics") + val execId = candidateExecIds.max + + val nameToAccId = + store.execution(execId).get.metrics.map(m => m.name -> m.accumulatorId).toMap + val uiValues = store.executionMetrics(execId) + + // Every planning metric must have a value in the UI store. Without the driver post its + // accumulator id never receives an update, so the store holds no entry for it. + Seq( + "totalDataManifest", + "scannedDataManifests", + "resultDataFiles", + "totalDataFileSize", + "totalPlanningDuration").foreach { name => + val accId = + nameToAccId.getOrElse(name, fail(s"planning metric $name missing from scan node")) + assert( + uiValues.contains(accId), + s"planning metric $name (accId=$accId) has no value in the SQL UI store; " + + "the driver metric was not posted") + } + + // Counts known to be non-zero for this table should render as non-zero in the UI. + Seq("totalDataManifest", "resultDataFiles").foreach { name => + assert( + uiValues(nameToAccId(name)).trim != "0", + s"$name should be non-zero, got ${uiValues(nameToAccId(name))}") + } + + // Scan time (native elapsed_compute) is declared on the scan node and tracked in the UI. + // The status store keys base metrics by their display name, not the native metric key, so + // it appears as "scan time" here (unlike the Iceberg planning metrics, which are registered + // under their Iceberg names). + assert( + nameToAccId.contains("scan time"), + s"scan time should be declared on the scan node; metrics=${nameToAccId.keySet}") + assert( + uiValues.contains(nameToAccId("scan time")), + "scan time should have a value in the SQL UI store") + + spark.sql("DROP TABLE test_cat.db.driver_metrics_test") + } + } + } + test("verify manifest pruning metrics") { assume(icebergAvailable, "Iceberg not available in classpath")