sunchao commented on code in PR #5381: URL: https://github.com/apache/datafusion-comet/pull/5381#discussion_r3929304039
########## spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala: ########## @@ -0,0 +1,219 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.spark.sql.benchmark + +import java.io.File + +import org.apache.spark.sql.catalyst.optimizer.InferFiltersFromGenerate +import org.apache.spark.sql.internal.SQLConf + +/** + * Benchmark to measure performance of Comet's explode operator (`CometExplodeExec`) against + * Spark's `GenerateExec`, across the dimensions that drive generator cost: fan-out, generator + * variant, element type, and the number of columns replicated alongside the generated one. To + * run: + * {{{ + * SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometExplodeBenchmark + * }}} + * + * Every case counts the generated columns rather than writing them, so the only row boundary is + * one row per partition. Writing them with `.noop()` would put a columnar-to-row conversion of + * every generated row inside the Comet arm and none inside the Spark arm, whose `GenerateExec` + * already emits rows: 419K conversions at fan-out 2 and 21M at fan-out 100, scaling with the very + * dimension the case is meant to isolate. `CometColumnarToRowBenchmark` measures that conversion + * on its own. + * + * Times are still whole-query totals and include the Parquet scan, the counting aggregate and its + * exchange, and per-iteration planning. The scan is a large share of the total at fan-out 2, + * where it compresses the ratio between the two engines, and a small one at fan-out 100. Issue + * #5363 tracks reporting operator cost against a scan baseline instead; when that lands, this + * paragraph should go. + * + * `InferFiltersFromGenerate` is excluded, for both engines. It infers `size(arr) > 0 AND arr IS + * NOT NULL` below a generator but matches only on `outer = false`, so leaving it enabled hands + * `explode` and `posexplode` 209,714 rows and an extra filter while their outer variants get all + * 262,144 — a comparison of two different plans, reported against a row count neither arm + * processes. Worth knowing when reading these numbers: real queries do get that filter, so a + * plain `explode` in production usually sees an array column with no nulls and no empty rows. + */ +object CometExplodeBenchmark extends CometBenchmarkBase { + + private val numRows = 256 * 1024 + + /** + * A SQL expression for an array column of `len` elements of `elementExpr`, where `elementExpr` + * may reference the row's `id` and the element's one-based position `x`. + * + * One in ten rows holds a null array and another one in ten holds an empty array, so that + * `explode` and `explode_outer` are a real comparison rather than the same query twice: the + * outer variants emit a null row for those 20% of rows where the plain variants emit nothing. + * + * The empty array is built with `slice`, not `array()`, because `array()` types as + * `array<null>` and would give that row's column a different element type. + * + * Elements are wrapped in a never-taken null branch so the array types as `containsNull`; see + * [[nullableExpr]]. + */ + private def arrayColumn(elementExpr: String, len: Int): String = { + val full = s"transform(sequence(1, $len), x -> ${nullableExpr(elementExpr, "x = 0")})" + s"""CASE + | WHEN id % 10 = 0 THEN NULL + | WHEN id % 10 = 1 THEN slice($full, 1, 0) + | ELSE $full + |END AS arr""".stripMargin + } + + /** + * Types `expr` as nullable without ever evaluating to null, `guard` being a predicate that is + * never true. + * + * Every column these queries count has to be nullable: `NullPropagation` rewrites `count(c)` to + * `count(1)` when `c` is not, and the counted column then has no reader at all. For the + * carried-column case that is fatal, because column pruning goes on to drop `k`, `s` and `v` + * from the generator's input, which is the entire dimension being measured. It would also split + * the variant group, since `outer` forces the generated column nullable and the plain variants + * do not. Parquet columns are usually optional in practice anyway. + */ + private def nullableExpr(expr: String, guard: String): String = s"IF($guard, NULL, $expr)" + + /** + * The string element, shared by the string and struct datasets so that the element-type cases + * differ in element type alone. A struct field of `s1` through `s10` would hold 10 distinct + * values against this column's 260,000-odd, which stays under the writer's 1 MiB dictionary + * page threshold where this one does not, so the comparison would also be measuring the + * difference between a dictionary-encoded column and a plain one. + */ + private val stringElement = "concat('str_', CAST(id + x AS STRING))" + + /** + * The temp views the benchmark reads, each with the expressions that build it. + * + * Each array column gets its own view rather than sharing one wide table, so that a case is + * never charged for scanning an array column it does not read. + */ + private val datasets: Seq[(String, Seq[String])] = Seq( + "arr_len2" -> Seq(arrayColumn("id + x", 2)), + "arr_len10" -> Seq(arrayColumn("id + x", 10)), + "arr_len100" -> Seq(arrayColumn("id + x", 100)), + "arr_str10" -> Seq(arrayColumn(stringElement, 10)), + "arr_struct10" -> Seq(arrayColumn(s"struct(id + x AS a, $stringElement AS b)", 10)), + "arr_carry" -> Seq( + arrayColumn("id + x", 10), + s"${nullableExpr("id", "id < 0")} AS k", + s"${nullableExpr("CAST(id AS STRING)", "id < 0")} AS s", + s"${nullableExpr("id * 2", "id < 0")} AS v")) + + /** Writes `selectExprs` over `numRows` rows to Parquet and registers it as a temp view. */ + private def createView(dir: File, name: String, selectExprs: Seq[String]): Unit = { + val path = s"${dir.getAbsolutePath}/$name" + spark.range(numRows).selectExpr(selectExprs: _*).write.parquet(path) + spark.read.parquet(path).createOrReplaceTempView(name) + } + + /** + * A query that applies `generator` to `view`'s `arr` column, carries `carried` through the + * generator alongside it, and counts every column that comes out. + * + * The position column of the `posexplode` variants is counted too, because a generator whose + * second output nothing reads is not the generator being named. + */ + private def countGenerated( + generator: String, + view: String, + carried: Seq[String] = Nil, + where: Option[String] = None): String = { + val generated = if (generator.startsWith("posexplode")) Seq("pos", "col") else Seq("col") + val alias = + if (generated.length == 1) s"AS ${generated.head}" + else generated.mkString("AS (", ", ", ")") + val projectList = (carried :+ s"$generator(arr) $alias").mkString(", ") + val filter = where.map(w => s" WHERE $w").getOrElse("") + val counts = (carried ++ generated).map(c => s"count($c)").mkString(", ") Review Comment: [P2] Make the position aggregate consume position values Could both position cases use a value-dependent aggregate such as `sum(pos)`, with an untimed result and plan check? The nullable array-element wrapper changes `col`, but `posexplode` still declares `pos` non-null. In the maintained Spark 3.5/4.0 sources, `NullPropagation` therefore rewrites `count(pos)` to `count(1)`. Even without that rewrite, `Count.updateExpressions` does not read non-null argument values. The outer variant checks only the position's null flag. This leaves the position values unconsumed despite the helper's stated intent. Spark supplies its existing loop index as the position, while Comet's planner constructs and unnests a separate positions list. A value-dependent sink makes the comparison exercise the output it names. On this dataset, `sum(pos)` should be `9437130` for both variants. Please retain the whole-query/aggregate-cost caveat when rerunning the measurements. -- 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]
