This is an automated email from the ASF dual-hosted git repository.
agrove pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new c0ec49b2 perf: Add metric for time spent casting in native scan (#919)
c0ec49b2 is described below
commit c0ec49b23c9d214a499a6e5c0678f0d45dde3e58
Author: Andy Grove <[email protected]>
AuthorDate: Tue Sep 10 17:22:49 2024 -0600
perf: Add metric for time spent casting in native scan (#919)
* add native metric for time spent casting in native scan
* save progress
* save progress
* fix test
* address feedback
---
native/core/src/execution/operators/scan.rs | 13 +++++++++++--
.../apache/spark/sql/comet/CometMetricNode.scala | 9 +++++++++
.../org/apache/spark/sql/comet/CometScanExec.scala | 7 +++++--
.../org/apache/comet/exec/CometExecSuite.scala | 22 ++++++++++++++++++++++
4 files changed, 47 insertions(+), 4 deletions(-)
diff --git a/native/core/src/execution/operators/scan.rs
b/native/core/src/execution/operators/scan.rs
index 0816a5c1..7d75f7f1 100644
--- a/native/core/src/execution/operators/scan.rs
+++ b/native/core/src/execution/operators/scan.rs
@@ -39,7 +39,9 @@ use arrow_data::ffi::FFI_ArrowArray;
use arrow_data::ArrayData;
use arrow_schema::ffi::FFI_ArrowSchema;
use arrow_schema::{DataType, Field, Schema, SchemaRef};
-use datafusion::physical_plan::metrics::{BaselineMetrics,
ExecutionPlanMetricsSet, MetricsSet};
+use datafusion::physical_plan::metrics::{
+ BaselineMetrics, ExecutionPlanMetricsSet, MetricBuilder, MetricsSet, Time,
+};
use datafusion::{
execution::TaskContext,
physical_expr::*,
@@ -345,16 +347,20 @@ struct ScanStream<'a> {
baseline_metrics: BaselineMetrics,
/// Cast options
cast_options: CastOptions<'a>,
+ /// elapsed time for casting columns to different data types during scan
+ cast_time: Time,
}
impl<'a> ScanStream<'a> {
pub fn new(scan: ScanExec, schema: SchemaRef, partition: usize) -> Self {
let baseline_metrics = BaselineMetrics::new(&scan.metrics, partition);
+ let cast_time =
MetricBuilder::new(&scan.metrics).subset_time("cast_time", partition);
Self {
scan,
schema,
baseline_metrics,
cast_options: CastOptions::default(),
+ cast_time,
}
}
@@ -375,7 +381,10 @@ impl<'a> ScanStream<'a> {
.zip(schema_fields.iter())
.map(|(column, f)| {
if column.data_type() != f.data_type() {
- cast_with_options(column, f.data_type(),
&self.cast_options)
+ let mut timer = self.cast_time.timer();
+ let cast_array = cast_with_options(column, f.data_type(),
&self.cast_options);
+ timer.stop();
+ cast_array
} else {
Ok(Arc::clone(column))
}
diff --git
a/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala
b/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala
index 195a0fde..07173c30 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometMetricNode.scala
@@ -79,6 +79,15 @@ object CometMetricNode {
"total time (in ms) spent in this operator"))
}
+ /**
+ * SQL Metrics for Comet native ScanExec
+ */
+ def scanMetrics(sc: SparkContext): Map[String, SQLMetric] = {
+ Map(
+ "cast_time" ->
+ SQLMetrics.createNanoTimingMetric(sc, "Total time for casting
columns"))
+ }
+
/**
* SQL Metrics for DataFusion HashJoin
*/
diff --git
a/spark/src/main/scala/org/apache/spark/sql/comet/CometScanExec.scala
b/spark/src/main/scala/org/apache/spark/sql/comet/CometScanExec.scala
index db36ce4b..2cbe8961 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometScanExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometScanExec.scala
@@ -198,9 +198,12 @@ case class CometScanExec(
// Tracking scan time has overhead, we can't afford to do it for each row,
and can only do
// it for each batch.
if (supportsColumnar) {
- Some("scanTime" -> SQLMetrics.createNanoTimingMetric(sparkContext, "scan
time"))
+ Map(
+ "scanTime" -> SQLMetrics.createNanoTimingMetric(
+ sparkContext,
+ "scan time")) ++ CometMetricNode.scanMetrics(sparkContext)
} else {
- None
+ Map.empty
}
} ++ {
relation.fileFormat match {
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
index c26af28c..7e6334e0 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala
@@ -517,6 +517,28 @@ class CometExecSuite extends CometTestBase {
}
}
+ test("Comet native metrics: scan") {
+ withSQLConf(CometConf.COMET_EXEC_ENABLED.key -> "true") {
+ withTempDir { dir =>
+ val path = new Path(dir.toURI.toString, "native-scan.parquet")
+ makeParquetFileAllTypes(path, dictionaryEnabled = true, 10000)
+ withParquetTable(path.toString, "tbl") {
+ val df = sql("SELECT * FROM tbl WHERE _2 > _3")
+ df.collect()
+
+ val metrics =
find(df.queryExecution.executedPlan)(_.isInstanceOf[CometScanExec])
+ .map(_.metrics)
+ .get
+
+ assert(metrics.contains("scanTime"))
+ assert(metrics.contains("cast_time"))
+ assert(metrics("scanTime").value > 0)
+ assert(metrics("cast_time").value > 0)
+ }
+ }
+ }
+ }
+
test("Comet native metrics: project and filter") {
withSQLConf(CometConf.COMET_EXEC_ENABLED.key -> "true") {
withParquetTable((0 until 5).map(i => (i, i + 1)), "tbl") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
