andygrove commented on code in PR #5381: URL: https://github.com/apache/datafusion-comet/pull/5381#discussion_r3929181843
########## 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)), Review Comment: Fixed. The struct's string field is now the same expression as the standalone string column, so both hold the same 260,000-odd distinct values and neither stays under the dictionary page threshold. The two datasets now differ only in that the struct wraps the string alongside a bigint, which is what I wanted the case to isolate. ########## 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( Review Comment: You're right, and this was the one that mattered most. Every case now terminates in a count aggregate instead of `.noop()`, so the only row boundary is above the final exchange rather than one per generated row. A matched baseline wouldn't have worked here for the reason you give: the conversion scales 50x across the fan-out group, so there is no fixed quantity to subtract. The old numbers sat flat at roughly 2x across every dimension, which in hindsight is exactly what a shared cost proportional to generated rows looks like. They now vary from 1.2x to 1.8x, and the variation is the useful part: Comet's margin narrows as fan-out grows, posexplode costs it 6ms over explode where Spark gets slightly cheaper, and posexplode_outer and array<struct> are its two weakest shapes. -- 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]
