andygrove commented on code in PR #5381: URL: https://github.com/apache/datafusion-comet/pull/5381#discussion_r3933916729
########## 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: Fixed, and you were right that nothing was reading a position. Both cases now end in `sum(pos)`, which comes out at 9437130 for `posexplode` and `posexplode_outer` alike — the outer variant's extra 52,430 rows carry a null position and `sum` skips them, so the two stay comparable. I took the "untimed result check" further than the position cases, because the failure mode you found is one the results table can never show: a sink that stops consuming what it names still reports a rate, just a better one. Every case now declares the aggregate row it must produce, and both engines run it once, untimed, against that row before the case is timed. The expected values are derived from the dataset parameters rather than pasted in, so they follow if the row count or fan-out changes. On the plan check: the harness already refuses to run a case whose Comet plan is not fully native, and all 15 are. I confirmed on the executed plan that `sum(pos)` survives to the operator rather than being folded away — `CometHashAggregate [pos, col], [Partial], [partial_sum(pos), partial_count(col)]` sitting directly on `CometExplode posexplode(arr), [pos, col], [pos, col]`. I kept the aggregate-cost caveat, and there is a bigger one to record. The numbers moved, and not because of this change. My earlier table was taken on a contended machine, and at `local[1]` contention lands almost entirely on the Spark arm. Re-running four times, the Comet arm reproduces to within a few ms everywhere while the Spark arm swings by up to 2x. To be sure the review fixes were not the cause I re-ran the *unchanged* previous revision in the same environment and got today's numbers, not the old ones. The corrected table is in the description; the short version is that Comet only wins at low fan-out over `array<bigint>`, and is roughly 2x slower on nested `array<struct>` and ~1.4x slower at fan-out 100. I would rather the benchmark say that than flatter us. -- 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]
