andygrove commented on code in PR #5381: URL: https://github.com/apache/datafusion-comet/pull/5381#discussion_r3929184191
########## spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala: ########## @@ -0,0 +1,135 @@ +/* + * 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 + +/** + * 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 + * }}} + * + * `runExpressionBenchmark` reports whole-query totals, so the times below also include the + * Parquet scan, the result transfer, and the per-iteration query planning. That fixed cost 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. + */ +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. + */ + private def arrayColumn(elementExpr: String, len: Int): String = { + val full = s"transform(sequence(1, $len), x -> $elementExpr)" + s"""CASE + | WHEN id % 10 = 0 THEN NULL + | WHEN id % 10 = 1 THEN slice($full, 1, 0) + | ELSE $full + |END AS arr""".stripMargin + } + + /** + * 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("concat('str_', CAST(id + x AS STRING))", 10)), + "arr_struct10" -> Seq( + arrayColumn("struct(id + x AS a, concat('s', CAST(x AS STRING)) AS b)", 10)), + "arr_carry" -> Seq( + arrayColumn("id + x", 10), + "id AS k", + "CAST(id AS STRING) AS s", + "id * 2 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) + } + + override def runCometBenchmark(mainArgs: Array[String]): Unit = { + withTempPath { dir => + withTempTable(datasets.map(_._1): _*) { + datasets.foreach { case (name, selectExprs) => createView(dir, name, selectExprs) } + + // Cardinality is input rows for every case, so the numbers are per scanned row rather + // than per generated row. Fan-out is named in the case title: the 100-element case emits + // roughly 50 times as many rows as the 2-element case from the same 256K inputs. + runBenchmark("Explode - fan-out") { + Seq(2, 10, 100).foreach { len => + runExpressionBenchmark( + s"explode array<bigint>[$len]", + numRows, + s"SELECT explode(arr) FROM arr_len$len") + } + } + + runBenchmark("Explode - generator variants") { + Seq("explode", "posexplode", "explode_outer", "posexplode_outer").foreach { generator => Review Comment: Fixed by excluding the rule, symmetrically. `runExpressionBenchmark` appends ConstantFolding to whatever the caller already has in `spark.sql.optimizer.excludedRules` and applies the result to both arms, so setting it once around the four groups covers Spark and Comet and keeps the existing exclusion. Worth recording somewhere, since it cuts the other way: the filter is not a benchmark artifact. It fires in any real query with a non-outer generator over an attribute, so a plain `explode` in production usually reaches the operator with an array column that has no nulls and no empty rows. I've noted that in the class comment. It also means the null-free path is worth optimizing for specifically. -- 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]
