From 040ec20c001942a20fc3b48931298300d9109c93 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sun, 16 Aug 2026 09:03:24 -0600 Subject: [PATCH 1/5] test: add explode operator microbenchmark Measures CometExplodeExec against Spark's GenerateExec across the four dimensions that drive generator cost: fan-out (array length 2, 10, 100), generator variant (explode, posexplode, and their outer forms), element type (bigint, string, struct), and the number of columns replicated alongside the generated one. One in ten rows holds a null array and another one in ten holds an empty array, so the outer variants do different work from the plain ones rather than measuring the same query twice. Each array column gets its own temp view so that no case is charged for scanning a column it does not read. --- .../sql/benchmark/CometExplodeBenchmark.scala | 169 ++++++++++++++++++ 1 file changed, 169 insertions(+) create mode 100644 spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala new file mode 100644 index 00000000000..1576c9d0890 --- /dev/null +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala @@ -0,0 +1,169 @@ +/* + * 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.SparkConf +import org.apache.spark.sql.SparkSession + +import org.apache.comet.CometSparkSessionExtensions + +/** + * 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 + * }}} + * + * Reported times are whole-query totals, so they include the Parquet scan and the transfer of + * results out of the engine. At fan-out 2 the scan is a large share of the total and the ratio + * understates the difference between the two explode implementations; at fan-out 100 the + * generator dominates and the ratio is close to the operator ratio. + */ +object CometExplodeBenchmark extends CometBenchmarkBase { + + override def getSparkSession: SparkSession = { + val conf = new SparkConf() + .setAppName("CometExplodeBenchmark") + .set("spark.master", "local[5]") + .setIfMissing("spark.driver.memory", "3g") + .setIfMissing("spark.executor.memory", "3g") + .set( + "spark.shuffle.manager", + "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager") + + val sparkSession = SparkSession + .builder() + .config(conf) + .withExtensions(new CometSparkSessionExtensions) + .getOrCreate() + sparkSession.conf.set("spark.sql.shuffle.partitions", "2") + sparkSession + } + + 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` 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 + } + + /** + * Writes `selectExprs` over `numRows` rows to Parquet and registers it as a temp view. + * + * 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 def createView(dir: File, name: String, selectExprs: 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 = { + val views = + Seq("arr_len2", "arr_len10", "arr_len100", "arr_str10", "arr_struct10", "arr_carry") + + withTempPath { dir => + withTempTable(views: _*) { + createView(dir, "arr_len2", arrayColumn("id + x", 2)) + createView(dir, "arr_len10", arrayColumn("id + x", 10)) + createView(dir, "arr_len100", arrayColumn("id + x", 100)) + createView(dir, "arr_str10", arrayColumn("concat('str_', CAST(id + x AS STRING))", 10)) + createView( + dir, + "arr_struct10", + arrayColumn("struct(id + x AS a, concat('s', CAST(x AS STRING)) AS b)", 10)) + createView( + dir, + "arr_carry", + arrayColumn("id + x", 10), + "id AS k", + "CAST(id AS STRING) AS s", + "id * 2 AS v") + + // 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[$len]", + numRows, + s"SELECT explode(arr) AS e FROM arr_len$len") + } + } + + runBenchmark("Explode - generator variants") { + // The pos- variants produce two output columns, so they need two aliases. + Seq( + "explode" -> "AS e", + "posexplode" -> "AS (p, e)", + "explode_outer" -> "AS e", + "posexplode_outer" -> "AS (p, e)").foreach { case (generator, alias) => + runExpressionBenchmark( + s"$generator array[10]", + numRows, + s"SELECT $generator(arr) $alias FROM arr_len10") + } + } + + runBenchmark("Explode - element type") { + Seq("bigint" -> "arr_len10", "string" -> "arr_str10", "struct" -> "arr_struct10") + .foreach { case (elementType, view) => + runExpressionBenchmark( + s"explode array<$elementType>[10]", + numRows, + s"SELECT explode(arr) AS e FROM $view") + } + } + + runBenchmark("Explode - carried columns") { + runExpressionBenchmark( + "explode alone", + numRows, + "SELECT explode(arr) AS e FROM arr_carry") + runExpressionBenchmark( + "explode plus 3 carried columns", + numRows, + "SELECT k, s, v, explode(arr) AS e FROM arr_carry") + } + } + } + } +} From 8df011f87abc66f1022344a0a6b9d8ab7e5b2c25 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Sun, 16 Aug 2026 09:24:56 -0600 Subject: [PATCH 2/5] refactor: drop duplicated session setup from explode benchmark Inherit the session from CometBenchmarkBase instead of copying the override a fifth time. The copy differed from the base only in using local[5], while silently dropping the base defaults for the vectorized reader, whole-stage codegen, the Comet toggles, and ANSI mode, and carrying a shuffle-partitions setting that no query here shuffles. Also drive view creation and cleanup from one dataset table rather than maintaining the view names in two places, drop generator aliases that no measurement depends on, and attribute the whole-query-total caveat to the harness with a pointer to #5363. --- .../sql/benchmark/CometExplodeBenchmark.scala | 92 ++++++------------- 1 file changed, 29 insertions(+), 63 deletions(-) diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala index 1576c9d0890..10aea5a17eb 100644 --- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala @@ -21,11 +21,6 @@ package org.apache.spark.sql.benchmark import java.io.File -import org.apache.spark.SparkConf -import org.apache.spark.sql.SparkSession - -import org.apache.comet.CometSparkSessionExtensions - /** * Benchmark to measure performance of Comet's explode operator (`CometExplodeExec`) against * Spark's `GenerateExec`, across the dimensions that drive generator cost: fan-out, generator @@ -35,32 +30,14 @@ import org.apache.comet.CometSparkSessionExtensions * SPARK_GENERATE_BENCHMARK_FILES=1 make benchmark-org.apache.spark.sql.benchmark.CometExplodeBenchmark * }}} * - * Reported times are whole-query totals, so they include the Parquet scan and the transfer of - * results out of the engine. At fan-out 2 the scan is a large share of the total and the ratio - * understates the difference between the two explode implementations; at fan-out 100 the - * generator dominates and the ratio is close to the operator ratio. + * `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 { - override def getSparkSession: SparkSession = { - val conf = new SparkConf() - .setAppName("CometExplodeBenchmark") - .set("spark.master", "local[5]") - .setIfMissing("spark.driver.memory", "3g") - .setIfMissing("spark.executor.memory", "3g") - .set( - "spark.shuffle.manager", - "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager") - - val sparkSession = SparkSession - .builder() - .config(conf) - .withExtensions(new CometSparkSessionExtensions) - .getOrCreate() - sparkSession.conf.set("spark.sql.shuffle.partitions", "2") - sparkSession - } - private val numRows = 256 * 1024 /** @@ -84,38 +61,35 @@ object CometExplodeBenchmark extends CometBenchmarkBase { } /** - * Writes `selectExprs` over `numRows` rows to Parquet and registers it as a temp view. + * 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 def createView(dir: File, name: String, selectExprs: String*): Unit = { + 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 = { - val views = - Seq("arr_len2", "arr_len10", "arr_len100", "arr_str10", "arr_struct10", "arr_carry") - withTempPath { dir => - withTempTable(views: _*) { - createView(dir, "arr_len2", arrayColumn("id + x", 2)) - createView(dir, "arr_len10", arrayColumn("id + x", 10)) - createView(dir, "arr_len100", arrayColumn("id + x", 100)) - createView(dir, "arr_str10", arrayColumn("concat('str_', CAST(id + x AS STRING))", 10)) - createView( - dir, - "arr_struct10", - arrayColumn("struct(id + x AS a, concat('s', CAST(x AS STRING)) AS b)", 10)) - createView( - dir, - "arr_carry", - arrayColumn("id + x", 10), - "id AS k", - "CAST(id AS STRING) AS s", - "id * 2 AS v") + 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 @@ -125,21 +99,16 @@ object CometExplodeBenchmark extends CometBenchmarkBase { runExpressionBenchmark( s"explode array[$len]", numRows, - s"SELECT explode(arr) AS e FROM arr_len$len") + s"SELECT explode(arr) FROM arr_len$len") } } runBenchmark("Explode - generator variants") { - // The pos- variants produce two output columns, so they need two aliases. - Seq( - "explode" -> "AS e", - "posexplode" -> "AS (p, e)", - "explode_outer" -> "AS e", - "posexplode_outer" -> "AS (p, e)").foreach { case (generator, alias) => + Seq("explode", "posexplode", "explode_outer", "posexplode_outer").foreach { generator => runExpressionBenchmark( s"$generator array[10]", numRows, - s"SELECT $generator(arr) $alias FROM arr_len10") + s"SELECT $generator(arr) FROM arr_len10") } } @@ -149,19 +118,16 @@ object CometExplodeBenchmark extends CometBenchmarkBase { runExpressionBenchmark( s"explode array<$elementType>[10]", numRows, - s"SELECT explode(arr) AS e FROM $view") + s"SELECT explode(arr) FROM $view") } } runBenchmark("Explode - carried columns") { - runExpressionBenchmark( - "explode alone", - numRows, - "SELECT explode(arr) AS e FROM arr_carry") + runExpressionBenchmark("explode alone", numRows, "SELECT explode(arr) FROM arr_carry") runExpressionBenchmark( "explode plus 3 carried columns", numRows, - "SELECT k, s, v, explode(arr) AS e FROM arr_carry") + "SELECT k, s, v, explode(arr) FROM arr_carry") } } } From 3f29d2b47bd1770f88b9cd729781373b2f74c775 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Thu, 3 Sep 2026 16:13:19 -0600 Subject: [PATCH 3/5] test: address review of the explode benchmark Four fixes, each to a measurement that did not isolate what its case claimed to. Count the generated columns instead of writing them. `.noop()` writes `InternalRow`, so the Comet arm was converting every generated row and the Spark arm, whose `GenerateExec` already emits rows, was not: 419K conversions at fan-out 2 against 21M at fan-out 100, scaling with the dimension the group exists to measure. Terminating in an aggregate puts the only row boundary above the final exchange. Exclude `InferFiltersFromGenerate` for both engines. It matches on `outer = false`, so it gave `explode` and `posexplode` 209,714 rows and an extra filter while their outer variants got all 262,144, and the rate was normalized on rows the non-outer arms never saw. Give the struct dataset the same string field as the string dataset. `s1` through `s10` stayed under the writer's dictionary page threshold where 260,000-odd distinct values do not, so the element-type group was also comparing a dictionary-encoded column against a plain one. Equalize the scan between the carried-column cases with an always-true filter over k, s and v. Column pruning drops them from the generator's input, so `explode alone` still does not replicate them, but the scan reads and decodes all four columns in both cases rather than three fewer in one of them. Every counted column is now nullable. `NullPropagation` rewrites `count(c)` to `count(1)` when `c` is not, which would leave the carried columns unreferenced and let pruning drop them before the generator -- the whole dimension. --- .../sql/benchmark/CometExplodeBenchmark.scala | 172 +++++++++++++----- 1 file changed, 128 insertions(+), 44 deletions(-) diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala index 10aea5a17eb..3ff25612f20 100644 --- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala @@ -21,6 +21,9 @@ 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 @@ -30,11 +33,25 @@ import java.io.File * 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. + * 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 { @@ -50,9 +67,12 @@ object CometExplodeBenchmark extends CometBenchmarkBase { * * The empty array is built with `slice`, not `array()`, because `array()` types as * `array` 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 -> $elementExpr)" + 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) @@ -60,6 +80,28 @@ object CometExplodeBenchmark extends CometBenchmarkBase { |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. * @@ -70,14 +112,13 @@ object CometExplodeBenchmark extends CometBenchmarkBase { "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_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), - "id AS k", - "CAST(id AS STRING) AS s", - "id * 2 AS v")) + 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 = { @@ -86,48 +127,91 @@ object CometExplodeBenchmark extends CometBenchmarkBase { 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(", ") + s"SELECT $counts FROM (SELECT $projectList FROM $view$filter)" + } + 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[$len]", - numRows, - s"SELECT explode(arr) FROM arr_len$len") - } - } + // `runExpressionBenchmark` appends ConstantFolding to whatever the caller has already + // excluded, and applies the result to both arms, so setting this here excludes both. + withSQLConf(SQLConf.OPTIMIZER_EXCLUDED_RULES.key -> InferFiltersFromGenerate.ruleName) { - runBenchmark("Explode - generator variants") { - Seq("explode", "posexplode", "explode_outer", "posexplode_outer").foreach { generator => - runExpressionBenchmark( - s"$generator array[10]", - numRows, - s"SELECT $generator(arr) FROM arr_len10") - } - } - - runBenchmark("Explode - element type") { - Seq("bigint" -> "arr_len10", "string" -> "arr_str10", "struct" -> "arr_struct10") - .foreach { case (elementType, view) => + // 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<$elementType>[10]", + s"explode array[$len]", numRows, - s"SELECT explode(arr) FROM $view") + countGenerated("explode", s"arr_len$len")) } - } + } + + runBenchmark("Explode - generator variants") { + Seq("explode", "posexplode", "explode_outer", "posexplode_outer").foreach { + generator => + runExpressionBenchmark( + s"$generator array[10]", + numRows, + countGenerated(generator, "arr_len10")) + } + } - runBenchmark("Explode - carried columns") { - runExpressionBenchmark("explode alone", numRows, "SELECT explode(arr) FROM arr_carry") - runExpressionBenchmark( - "explode plus 3 carried columns", - numRows, - "SELECT k, s, v, explode(arr) FROM arr_carry") + runBenchmark("Explode - element type") { + Seq("bigint" -> "arr_len10", "string" -> "arr_str10", "struct" -> "arr_struct10") + .foreach { case (elementType, view) => + runExpressionBenchmark( + s"explode array<$elementType>[10]", + numRows, + countGenerated("explode", view)) + } + } + + // Both cases read all four columns. The filter is always true and references k, s and + // v below the generator, where column pruning then drops them from the generator's + // input rather than replicating them; without it, `explode alone` would prune three + // columns from the Parquet scan and the difference between the two cases would be + // scan and string-decode work as much as replication. What the filter does not + // equalize is the three extra counts the carried case runs over the generated rows. + // Those are cheaper than the three gathers they exist to measure, but not free. + runBenchmark("Explode - carried columns") { + val readAllColumns = Some("k >= 0 AND v >= 0 AND length(s) > 0") + runExpressionBenchmark( + "explode alone", + numRows, + countGenerated("explode", "arr_carry", where = readAllColumns)) + runExpressionBenchmark( + "explode plus 3 carried columns", + numRows, + countGenerated( + "explode", + "arr_carry", + carried = Seq("k", "s", "v"), + where = readAllColumns)) + } } } } From 0305fb37d7a6a2eff92fa3f3ce27913dc2c72dcf Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 4 Sep 2026 06:21:17 -0600 Subject: [PATCH 4/5] test: sum the position column, verify every sink, and cover nested input Addresses the second round of review. The position cases now end in `sum(pos)` rather than `count(pos)`. `pos` is declared non-null, so `NullPropagation` rewrote the count to `count(1)` and no position value was ever read -- which matters here because Spark hands the generator its existing loop index while Comet materializes a parallel List through `ListPositionsExpr` and unnests it alongside the values. Every case now declares the aggregate row it must produce, and both engines are run against it untimed before the case is timed. A sink that stops reading what it names still reports a rate; this makes the next one fail loudly instead. Adds an `Explode - nested input` group for the wide, deeply nested shape asked for in review: a customer profile whose event list sits eight struct accessors down, with a second array of four-field structs inside each element, over 100K rows. The requested outer container was a map; that is an array of structs here because Comet has no native generator over maps yet (#2837), so the Comet arm of a map case would silently be Spark. --- .../sql/benchmark/CometExplodeBenchmark.scala | 335 ++++++++++++++---- 1 file changed, 272 insertions(+), 63 deletions(-) diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala index 3ff25612f20..865389bb388 100644 --- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala @@ -24,23 +24,29 @@ import java.io.File import org.apache.spark.sql.catalyst.optimizer.InferFiltersFromGenerate import org.apache.spark.sql.internal.SQLConf +import org.apache.comet.CometConf + /** * 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: + * variant, element type, how deeply the exploded array is nested in its input row, 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 case aggregates 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 + * Every case's aggregate is checked against the row it must produce, under both engines, before + * it is timed; see [[verifySink]]. A sink that quietly stops reading what it names still returns + * a number, and the results table cannot tell the difference. + * + * Times are still whole-query totals and include the Parquet scan, the 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 @@ -52,32 +58,55 @@ import org.apache.spark.sql.internal.SQLConf * 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. + * + * Only array inputs are covered. Comet declines to convert a generator over a map + * (https://github.com/apache/datafusion-comet/issues/2837), so the Comet arm of such a case would + * be Spark's `GenerateExec` behind a columnar-to-row transition and its timing would say nothing + * about `CometExplodeExec`. The nesting group below therefore reaches its event list through an + * `array>` where a map would be the more natural modeling choice. */ object CometExplodeBenchmark extends CometBenchmarkBase { private val numRows = 256 * 1024 + /** One benchmark case: the query to time, and the single row it must produce. */ + private case class Case(query: String, expected: Seq[Long]) + + /** A temp view the benchmark reads: its row count, and the expressions that build it. */ + private case class TempView(name: String, rows: Int, columns: Seq[String]) + + /** + * Rows of an `rows`-row dataset whose [[arrayColumn]] is neither NULL nor empty, which is how + * many rows reach a non-outer generator with something to emit. + */ + private def nonEmptyRows(rows: Int): Long = rows - (rows + 9) / 10 - (rows + 8) / 10 + /** - * 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`. + * A SQL expression for an array of `len` elements of `elementExpr`, where `elementExpr` may + * reference the row's `id` and the element's one-based position `v`. * - * 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. + * Elements are wrapped in a never-taken null branch so the array types as `containsNull`; see + * [[nullableExpr]]. + */ + private def fullArray(elementExpr: String, len: Int, v: String = "x"): String = + s"transform(sequence(1, $len), $v -> ${nullableExpr(elementExpr, s"$v = 0")})" + + /** + * [[fullArray]], but one row in ten 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` 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")})" + private def arrayColumn(elementExpr: String, len: Int, v: String = "x"): String = { + val full = fullArray(elementExpr, len, v) s"""CASE | WHEN id % 10 = 0 THEN NULL | WHEN id % 10 = 1 THEN slice($full, 1, 0) | ELSE $full - |END AS arr""".stripMargin + |END""".stripMargin } /** @@ -102,57 +131,188 @@ object CometExplodeBenchmark extends CometBenchmarkBase { */ private val stringElement = "concat('str_', CAST(id + x AS STRING))" + /** Rows in the nesting group's dataset. Its rows are far larger than the other datasets'. */ + private val nestedRows = 100 * 1024 + + /** Event lists per row, and events per list, in the nesting group's dataset. */ + private val eventsLen = 4 + private val entriesLen = 5 + + /** The struct fields between `profile.account` and `sessions`, outermost first. */ + private val accountPath = Seq("settings", "preferences", "notifications", "activity") + + /** `profile.account.settings.preferences.notifications.activity.sessions.events`. */ + private val deepEvents = + (Seq("profile", "account") ++ accountPath ++ Seq("sessions", "events")).mkString(".") + + /** One event: four scalar fields, so that carrying it through a generator is not free. */ + private val entryStruct = + """named_struct( + | 'type', concat('t_', CAST(y AS STRING)), + | 'ts', id * 1000 + y, + | 'page', concat('page_', CAST(id + y AS STRING)), + | 'source', concat('src_', CAST(id % 4 AS STRING)))""".stripMargin + + /** One platform's event list, the element of the array the nesting group explodes. */ + private val eventStruct = + s"""named_struct( + | 'platform', concat('p_', CAST(x AS STRING)), + | 'entries', ${fullArray(entryStruct, entriesLen, "y")})""".stripMargin + + private val eventsArray = arrayColumn(eventStruct, eventsLen) + + /** [[eventsArray]] wrapped in the struct chain that puts it eight accessors down. */ + private val profileColumn = { + val sessions = + s"""named_struct( + | 'device', named_struct('type', 'mobile', 'os', 'iOS'), + | 'events', $eventsArray)""".stripMargin + val account = accountPath.foldRight(s"named_struct('sessions', $sessions)") { + (field, inner) => s"named_struct('$field', $inner)" + } + s"""named_struct( + | 'account', $account, + | 'billing', named_struct('currency', 'USD', 'country', 'US'), + | 'addresses', array(named_struct('city', 'San Jose', 'country', 'US')))""".stripMargin + } + /** - * The temp views the benchmark reads, each with the expressions that build it. + * The temp views the benchmark reads. * * 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. + * never charged for scanning an array column it does not read. The nesting group is the + * exception: `events` and `profile` hold the same array at two different depths, and keeping + * them in one view is what makes them the same array. Nested schema pruning stops either case + * from reading the other's copy. */ - 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) + private val views: Seq[TempView] = Seq( + TempView("arr_len2", numRows, Seq(s"${arrayColumn("id + x", 2)} AS arr")), + TempView("arr_len10", numRows, Seq(s"${arrayColumn("id + x", 10)} AS arr")), + TempView("arr_len100", numRows, Seq(s"${arrayColumn("id + x", 100)} AS arr")), + TempView("arr_str10", numRows, Seq(s"${arrayColumn(stringElement, 10)} AS arr")), + TempView( + "arr_struct10", + numRows, + Seq(s"${arrayColumn(s"struct(id + x AS a, $stringElement AS b)", 10)} AS arr")), + TempView( + "arr_carry", + numRows, + Seq( + s"${arrayColumn("id + x", 10)} AS arr", + 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")), + TempView( + "nested", + nestedRows, + Seq( + s"${nullableExpr("id", "id < 0")} AS k", + s"${nullableExpr("concat('r_', CAST(id % 8 AS STRING))", "id < 0")} AS region", + s"$eventsArray AS events", + s"$profileColumn AS profile"))) + + /** Writes a view's rows to Parquet and registers it. */ + private def createView(dir: File, view: TempView): Unit = { + val path = s"${dir.getAbsolutePath}/${view.name}" + spark.range(view.rows).selectExpr(view.columns: _*).write.parquet(path) + spark.read.parquet(path).createOrReplaceTempView(view.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. + * A case that applies `generator` to `arrayExpr` over `view`, carries `carried` through the + * generator alongside it, and aggregates every column that comes out. + * + * The position column of the `posexplode` variants is summed rather than counted, because + * counting it does not read a position. `pos` is declared non-null, so `NullPropagation` + * rewrites `count(pos)` to `count(1)`; even without that rewrite `Count` never reads a non-null + * argument's value. The two engines produce those values very differently — Spark hands over + * the loop index it already has, while Comet materializes a parallel `List` through + * `ListPositionsExpr` and unnests it alongside the values — so a sink that ignores them is not + * measuring the generator it names. `sum` over the same column is value-dependent. * - * The position column of the `posexplode` variants is counted too, because a generator whose - * second output nothing reads is not the generator being named. + * Everything else is counted rather than summed: the element type varies across these cases and + * most of the element types cannot be summed, and unlike `pos` the generated column is nullable + * everywhere, so the count is not rewritten away. */ - private def countGenerated( + private def generatorCase( generator: String, view: String, + len: Int, + rows: Int = numRows, + arrayExpr: String = "arr", carried: Seq[String] = Nil, - where: Option[String] = None): String = { - val generated = if (generator.startsWith("posexplode")) Seq("pos", "col") else Seq("col") + where: Option[String] = None): Case = { + val position = generator.startsWith("posexplode") + val generated = if (position) 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 projectList = (carried :+ s"$generator($arrayExpr) $alias").mkString(", ") val filter = where.map(w => s" WHERE $w").getOrElse("") - val counts = (carried ++ generated).map(c => s"count($c)").mkString(", ") - s"SELECT $counts FROM (SELECT $projectList FROM $view$filter)" + val aggregates = (carried ++ generated).map(aggregate) + val query = + s"SELECT ${aggregates.mkString(", ")} FROM (SELECT $projectList FROM $view$filter)" + + // The outer variants add one all-null row for each row the plain variants drop, which the + // carried columns count and the generated columns do not. + val elements = nonEmptyRows(rows) * len + val outputRows = + if (generator.endsWith("_outer")) elements + (rows - nonEmptyRows(rows)) else elements + val expected = carried.map(_ => outputRows) ++ generated.map { + // Positions run 0 until len on every row that emits, and are null on the rows only an + // outer variant emits, which `sum` skips. + case "pos" => elements * (len - 1) / 2 + case _ => elements + } + Case(query, expected) + } + + /** See [[generatorCase]] for why `pos` alone is summed. */ + private def aggregate(column: String): String = + if (column == "pos") s"sum($column)" else s"count($column)" + + /** + * Runs `query` under both engines, untimed, and fails unless each produces exactly `expected`. + * + * This guards the sinks. Every timing here is only worth reading if the case's aggregate + * actually consumes the generated columns, and an aggregate that has stopped consuming them is + * invisible in the results table: it still reports a rate, just a better one. `count(pos)` was + * such a sink until this benchmark switched to `sum(pos)`. Pinning the expected values makes + * the next one loud. + * + * Constant folding is left enabled, unlike in the timed runs. Excluding it there keeps per-row + * work per-row; it cannot change a result, and these queries hold no constant subexpression for + * it to fold in any case. + */ + private def verifySink(name: String, query: String, expected: Seq[Long]): Unit = + Seq(false, true).foreach { cometEnabled => + val engine = if (cometEnabled) "Comet" else "Spark" + withSQLConf( + CometConf.COMET_ENABLED.key -> cometEnabled.toString, + CometConf.COMET_EXEC_ENABLED.key -> cometEnabled.toString) { + val actual = spark.sql(query).collect().head.toSeq.map { + case null => null + case value: Number => value.longValue() + case other => other + } + if (actual != expected) { + throw new AssertionError( + s"$name: $engine produced $actual, expected $expected. The case is not aggregating " + + s"what it names.\n$query") + } + } + } + + /** Verifies a case's aggregate under both engines, then times it. */ + private def runCase(name: String, benchmarkCase: Case, rows: Int = numRows): Unit = { + verifySink(name, benchmarkCase.query, benchmarkCase.expected) + runExpressionBenchmark(name, rows, benchmarkCase.query) } override def runCometBenchmark(mainArgs: Array[String]): Unit = { withTempPath { dir => - withTempTable(datasets.map(_._1): _*) { - datasets.foreach { case (name, selectExprs) => createView(dir, name, selectExprs) } + withTempTable(views.map(_.name): _*) { + views.foreach(view => createView(dir, view)) // `runExpressionBenchmark` appends ConstantFolding to whatever the caller has already // excluded, and applies the result to both arms, so setting this here excludes both. @@ -163,33 +323,34 @@ object CometExplodeBenchmark extends CometBenchmarkBase { // 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( + runCase( s"explode array[$len]", - numRows, - countGenerated("explode", s"arr_len$len")) + generatorCase("explode", s"arr_len$len", len)) } } runBenchmark("Explode - generator variants") { Seq("explode", "posexplode", "explode_outer", "posexplode_outer").foreach { generator => - runExpressionBenchmark( + runCase( s"$generator array[10]", - numRows, - countGenerated(generator, "arr_len10")) + generatorCase(generator, "arr_len10", 10)) } } runBenchmark("Explode - element type") { Seq("bigint" -> "arr_len10", "string" -> "arr_str10", "struct" -> "arr_struct10") .foreach { case (elementType, view) => - runExpressionBenchmark( - s"explode array<$elementType>[10]", - numRows, - countGenerated("explode", view)) + runCase(s"explode array<$elementType>[10]", generatorCase("explode", view, 10)) } } + runBenchmark("Explode - nested input") { + nestedInputCases.foreach { case (name, benchmarkCase) => + runCase(name, benchmarkCase, nestedRows) + } + } + // Both cases read all four columns. The filter is always true and references k, s and // v below the generator, where column pruning then drops them from the generator's // input rather than replicating them; without it, `explode alone` would prune three @@ -199,16 +360,15 @@ object CometExplodeBenchmark extends CometBenchmarkBase { // Those are cheaper than the three gathers they exist to measure, but not free. runBenchmark("Explode - carried columns") { val readAllColumns = Some("k >= 0 AND v >= 0 AND length(s) > 0") - runExpressionBenchmark( + runCase( "explode alone", - numRows, - countGenerated("explode", "arr_carry", where = readAllColumns)) - runExpressionBenchmark( + generatorCase("explode", "arr_carry", 10, where = readAllColumns)) + runCase( "explode plus 3 carried columns", - numRows, - countGenerated( + generatorCase( "explode", "arr_carry", + 10, carried = Seq("k", "s", "v"), where = readAllColumns)) } @@ -216,4 +376,53 @@ object CometExplodeBenchmark extends CometBenchmarkBase { } } } + + /** + * The nesting group, over 100K rows rather than 256K because each row carries 20 events. + * + * The shape is the one asked for in review: a customer profile whose event list sits eight + * struct accessors down, holding a second array of four-field structs inside each element. The + * requested outer container was a map keyed by platform, which is where a real schema would put + * it; that is an `array>` here because Comet has no native generator + * over maps yet (#2837) and the Comet arm would silently be Spark. + * + * The whole event struct is counted rather than one of its fields, so nested schema pruning + * cannot narrow the exploded element and leave the case measuring a two-column gather. It does + * prune the siblings no case reads: on the executed plan the `depth 8` scan's `profile` column + * is `struct>>>` with `billing`, + * `addresses` and `device` gone, and the `depth 1` scan does not project `profile` at all. + * + * `depth 1` and `depth 8` explode the same array, written twice into the same file, so the pair + * measures what the struct chain costs: extra definition levels in the Parquet column and a + * chain of `GetStructField` above the scan. `then its inner array` chains a second generator + * onto the first, which is the shape the review asked about; its fan-out is 20 against the + * other two's 4, so read it on its own rather than against them. + */ + private def nestedInputCases: Seq[(String, Case)] = { + val carried = Seq("k", "region") + val single = Seq("depth 1" -> "events", "depth 8" -> deepEvents).map { case (label, path) => + s"explode array[$eventsLen] at $label" -> + generatorCase( + "explode", + "nested", + eventsLen, + rows = nestedRows, + arrayExpr = path, + carried = carried) + } + + // The inner arrays are always full, so the second generator drops nothing and the case's + // fan-out is the product of the two lengths. + val events = nonEmptyRows(nestedRows) * eventsLen + val entries = events * entriesLen + val outer = + s"SELECT k, region, ev.platform AS platform, ev.entries AS entries " + + s"FROM (SELECT k, region, explode($deepEvents) AS ev FROM nested)" + val chained = + s"SELECT ${Seq("k", "region", "platform", "entry").map(aggregate).mkString(", ")} " + + s"FROM (SELECT k, region, platform, explode(entries) AS entry FROM ($outer))" + + single :+ (s"explode array[$eventsLen] at depth 8, then its inner array" -> + Case(chained, Seq.fill(4)(entries))) + } } From f79d0b8e95a127dfe88b7b0090e69c988594f380 Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Fri, 4 Sep 2026 06:39:56 -0600 Subject: [PATCH 5/5] fix: drop redundant string interpolator flagged by scalafix RedundantSyntax --- .../org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala index 865389bb388..aa16417ca2c 100644 --- a/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala +++ b/spark/src/test/scala/org/apache/spark/sql/benchmark/CometExplodeBenchmark.scala @@ -416,7 +416,7 @@ object CometExplodeBenchmark extends CometBenchmarkBase { val events = nonEmptyRows(nestedRows) * eventsLen val entries = events * entriesLen val outer = - s"SELECT k, region, ev.platform AS platform, ev.entries AS entries " + + "SELECT k, region, ev.platform AS platform, ev.entries AS entries " + s"FROM (SELECT k, region, explode($deepEvents) AS ev FROM nested)" val chained = s"SELECT ${Seq("k", "region", "platform", "entry").map(aggregate).mkString(", ")} " +