You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/10/10 14:45:19 UTC
[PR] test: extend C2R and R2C microbenchmarks to cover nested types [datafusion-comet]
andygrove opened a new pull request, #6856:
URL: https://github.com/apache/datafusion-comet/pull/6856
## Which issue does this PR close?
Part of #6855.
## Rationale for this change
The C2R and R2C microbenchmarks covered nested types thinly: nested test data never contained nulls, collections held 2-5 simple elements, the isolated C2R bench and the native `row_columnar` bench had no strings or collections of structs, and nothing measured `CometSparkToColumnarExec` on nested input. Extending them found several cases where Comet is slower than Spark (filed under #6855).
## What changes are included in this PR?
Benchmarks only, no production code.
- `CometColumnarToRowBenchmark`: three new groups. Nullable nested types (null structs, arrays and maps, empty arrays, null elements), nested collections (array of struct, array of array, struct of array of struct, map of struct, decimals, timestamps and binary inside nested types), and large collections (100-element arrays, 20-entry maps).
- `CometC2RIsolatedBench`: a nested schema (struct, array, map, array of struct, with nulls) alongside the flat one, at batch sizes 8192, 512 and 32.
- `native/shuffle/benches/row_columnar.rs`: a small UnsafeRow encoder and 12 new cases covering `list<utf8>`, `list<struct>`, `list<list>`, lists with null elements, structs with strings, nested structs with strings, a struct holding a list, and `map<utf8,*>` variants.
- `CometArrowWriterBenchmark`: nine more nested types (nested structs, `array<struct>`, `array<array>`, `struct<array>`, maps with array values, and others).
- `CometSparkToColumnarBenchmark` (new): Spark Parquet scan converted to Arrow versus Spark and the native scan, for eight nested shapes, with plan checks on each arm.
- `CometShuffleBenchmark`: ten nested shapes across Spark shuffle, Comet Spark shuffle, and JVM and native shuffle, plus a `--nested-only` flag.
## How are these changes tested?
The new benchmarks were built in release mode and run on a Linux machine; results and findings are in #6855. The native bench encoder was checked to produce byte-identical rows to the existing hand-built `list` and `map` rows. `cargo fmt`, `cargo clippy --benches -D warnings`, `spotless` and `dev/ci/check-benchmark-runner.py` pass.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] test: extend C2R and R2C microbenchmarks to cover nested types [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #6856:
URL: https://github.com/apache/datafusion-comet/pull/6856#issuecomment-6110922552
@peterxcli you worked on the in-memory cache in #5869, so you know this area. If you have time, would you be able to review this one? Thanks.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] test: extend C2R and R2C microbenchmarks to cover nested types [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6856:
URL: https://github.com/apache/datafusion-comet/pull/6856#discussion_r4238019814
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometShuffleBenchmark.scala:
##########
@@ -720,6 +727,63 @@ object CometShuffleBenchmark extends CometBenchmarkBase {
}
}
+ /**
+ * Nested shapes the single-type array and struct groups above do not reach: collections of
+ * strings and structs, maps, nulls at every level, and structs with variable-width fields. Each
+ * shape is repartitioned by a round-robin so the shuffle moves the whole column.
+ */
+ private def runNestedShapeBenchmarks(): Unit = {
+ val numRows = 1024 * 1024
+ val shapes: Seq[(String, String)] = Seq(
+ "array<string>" ->
+ "array(concat('a_', cast(id as string)), concat('b_', cast(id as string)), 'c')",
+ "array<struct<int,string>>" ->
+ ("transform(sequence(0, cast(id % 4 as int)), i -> named_struct('i', cast(id as int) + i, " +
+ "'s', concat('n_', cast(id as string))))"),
+ "array<array<int>>" ->
+ ("transform(sequence(0, cast(id % 3 as int)), i -> transform(sequence(0, " +
+ "cast(id % 5 as int)), j -> cast(id as int) + i + j))"),
+ "array<int> (100 elements)" ->
+ "transform(sequence(1, 100), i -> cast(id % 1000 as int) + i)",
+ "map<string,int>" -> "map('k1', cast(id as int), 'k2', cast(id + 1 as int))",
+ "map<int,array<string>>" ->
+ "map(cast(id % 10 as int), array(concat('v_', cast(id as string)), 'x'))",
+ "struct<int,string,string,long>" ->
+ ("named_struct('i', cast(id as int), 's1', concat('s1_', cast(id as string)), " +
+ "'s2', concat('s2_', cast(id as string)), 'l', id)"),
+ "nullable struct, nullable fields" ->
+ ("if(id % 5 = 0, null, named_struct('i', if(id % 3 = 0, null, cast(id as int)), " +
+ "'s', if(id % 4 = 0, null, concat('s_', cast(id as string)))))"),
+ "nullable array<int> with null elements" ->
+ ("if(id % 6 = 0, null, array(cast(id as int), if(id % 3 = 0, null, cast(id + 1 as int)), " +
+ "cast(id + 2 as int)))"),
+ "struct<array<struct<long,string>>,string>" ->
+ ("named_struct('items', transform(sequence(0, cast(id % 3 as int)), i -> " +
+ "named_struct('k', id + i, 'v', concat('v_', cast(i as string)))), " +
+ "'label', concat('l_', cast(id as string)))"))
+
+ shapes.foreach { case (name, expr) =>
+ withTempPath { dir =>
+ val filename = dir.getCanonicalPath + "/nested.parquet"
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .range(numRows.toLong)
+ .selectExpr("id AS key", s"$expr AS c")
+ .write
+ .mode(SaveMode.Overwrite)
+ .parquet(filename)
+ }
+ Seq(5, manyPartitions).foreach { partitionNum =>
+ shuffleDeeplyNestedBenchmark(
Review Comment:
[P2] Make the added native-shuffle cases actually select native shuffle. Each new shape calls `shuffleDeeplyNestedBenchmark`, whose `Comet (native Shuffle)` arm uses `.repartition(partitionNum)` and sets `spark.comet.shuffle.mode=native`. At both 5 and 201 partitions this requests `RoundRobinPartitioning`, but `spark.comet.shuffle.native.partitioning.roundrobin.enabled` remains false. Consequently, the arm falls back to Spark shuffle instead of measuring native shuffle, duplicating the Spark-shuffle comparison and leaving all 20 added native measurements mislabeled. Please explicitly enable native round-robin, or repartition these fixtures by their scalar `key`, and verify a `CometNativeShuffle` exchange outside timing.
Evidence: Compiled the exact-head `CometConf.scala` and `CometShuffleExchangeExec.scala` with a disposable Spark 4.1.3 planner probe. For five representative added nested types at both partition counts, mode `native` returned `None` from `shuffleSupported`, with reasons `spark.comet.shuffle.native.partitioning.roundrobin.enabled is disabled` and `Comet columnar shuffle not enabled`. Mode `jvm` returned `Some(CometColumnarShuffle)`. Enabling the round-robin flag returned `Some(CometNativeShuffle)`. Spark sources for every supported profile map unkeyed repartitioning with more than one partition to `RoundRobinPartitioning`. The benchmark and standard launcher do not enable that flag.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] test: extend C2R and R2C microbenchmarks to cover nested types [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6856:
URL: https://github.com/apache/datafusion-comet/pull/6856#discussion_r4266110522
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometC2RIsolatedBench.scala:
##########
@@ -92,15 +148,58 @@ object CometC2RIsolatedBench {
batches.foreach(_.close())
}
+ /** Reads every field of the flat schema. */
+ private def consumeFlat(u: InternalRow): Long =
+ u.getLong(0) + u.getInt(1) + u.getDouble(2).toLong + u.getUTF8String(3).numBytes()
+
+ /**
+ * Reads each nested column through its accessor, descending into the struct, array and map
+ * elements, so that the cost of materialising a nested value is part of the measurement.
+ */
+ private def consumeNested(u: InternalRow): Long = {
+ var sink = u.getLong(0)
+ if (!u.isNullAt(1)) {
+ val s = u.getStruct(1, 3)
+ sink += s.getInt(0) + s.getUTF8String(1).numBytes()
+ if (!s.isNullAt(2)) sink += s.getStruct(2, 2).getLong(1)
+ }
+ if (!u.isNullAt(2)) {
+ val a = u.getArray(2)
+ var i = 0
+ while (i < a.numElements()) {
+ if (!a.isNullAt(i)) sink += a.getInt(i)
+ i += 1
+ }
+ }
+ if (!u.isNullAt(3)) {
+ val a = u.getArray(3)
+ var i = 0
+ while (i < a.numElements()) {
+ sink += a.getUTF8String(i).numBytes()
+ i += 1
+ }
+ }
+ if (!u.isNullAt(4)) {
+ val m = u.getMap(4)
+ sink += m.numElements() + m.valueArray().getInt(0)
+ }
+ val structs = u.getArray(5)
+ var i = 0
+ while (i < structs.numElements()) {
+ val e = structs.getStruct(i, 2)
+ sink += e.getInt(0) + e.getUTF8String(1).numBytes()
+ i += 1
+ }
+ sink
+ }
+
def main(args: Array[String]): Unit = {
- val schema = new StructType()
- .add("a", LongType)
- .add("b", IntegerType)
- .add("c", DoubleType)
- .add("d", StringType)
-
- runForBatchSize(schema, 8192)
- runForBatchSize(schema, 512)
- runForBatchSize(schema, 32)
+ runForBatchSize("flat", flatSchema, 8192, flatRow, consumeFlat)
+ runForBatchSize("flat", flatSchema, 512, flatRow, consumeFlat)
+ runForBatchSize("flat", flatSchema, 32, flatRow, consumeFlat)
+
+ runForBatchSize("nested", nestedSchema, 8192, nestedRow, consumeNested)
+ runForBatchSize("nested", nestedSchema, 512, nestedRow, consumeNested)
+ runForBatchSize("nested", nestedSchema, 32, nestedRow, consumeNested)
Review Comment:
[P2] Please bound the retained fixtures for this new 32-row nested case. `makeBatches` materializes all 1,048,576 rows into 32,768 independently allocated batches before timing. `ArrowWriter.create` sizes top-level fixed-width vectors, but nested vectors retain Arrow's default capacities. The resulting fixture allocation is approximately 12.14 GiB, versus 1.54 GiB for the existing flat 32-row case. This makes a small-batch conversion benchmark require over 12 GiB of native fixture memory before either arm runs, preventing it from fitting within an 8 GiB worker budget that could accommodate the previous fixture. Right-size the nested buffers during setup, or retain a bounded batch collection and repeat conversion to preserve the timed row count.
Evidence: A disposable Spark 4.1.3 / Arrow 18.3.0 probe compiled exact-head `ArrowWriters.scala` and `CometArrowConverters.scala` and used the benchmark's unchanged schema and row generators. Four 32-row nested batches retained 1,591,296 bytes at starting rows 0, 524288, and 1048448. That is 397,824 bytes per batch, or 13,035,896,832 bytes for 32,768 batches. The equivalent flat samples retained 201,728 bytes, totaling 1,652,555,776 bytes. A separate bounded 256-batch run retained 101,842,944 bytes for nested fixtures versus 12,910,592 bytes for flat fixtures. All buffers were closed after measurement.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] test: extend C2R and R2C microbenchmarks to cover nested types [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6856:
URL: https://github.com/apache/datafusion-comet/pull/6856#discussion_r4238504737
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometShuffleBenchmark.scala:
##########
@@ -720,6 +727,63 @@ object CometShuffleBenchmark extends CometBenchmarkBase {
}
}
+ /**
+ * Nested shapes the single-type array and struct groups above do not reach: collections of
+ * strings and structs, maps, nulls at every level, and structs with variable-width fields. Each
+ * shape is repartitioned by a round-robin so the shuffle moves the whole column.
+ */
+ private def runNestedShapeBenchmarks(): Unit = {
+ val numRows = 1024 * 1024
+ val shapes: Seq[(String, String)] = Seq(
+ "array<string>" ->
+ "array(concat('a_', cast(id as string)), concat('b_', cast(id as string)), 'c')",
+ "array<struct<int,string>>" ->
+ ("transform(sequence(0, cast(id % 4 as int)), i -> named_struct('i', cast(id as int) + i, " +
+ "'s', concat('n_', cast(id as string))))"),
+ "array<array<int>>" ->
+ ("transform(sequence(0, cast(id % 3 as int)), i -> transform(sequence(0, " +
+ "cast(id % 5 as int)), j -> cast(id as int) + i + j))"),
+ "array<int> (100 elements)" ->
+ "transform(sequence(1, 100), i -> cast(id % 1000 as int) + i)",
+ "map<string,int>" -> "map('k1', cast(id as int), 'k2', cast(id + 1 as int))",
+ "map<int,array<string>>" ->
+ "map(cast(id % 10 as int), array(concat('v_', cast(id as string)), 'x'))",
+ "struct<int,string,string,long>" ->
+ ("named_struct('i', cast(id as int), 's1', concat('s1_', cast(id as string)), " +
+ "'s2', concat('s2_', cast(id as string)), 'l', id)"),
+ "nullable struct, nullable fields" ->
+ ("if(id % 5 = 0, null, named_struct('i', if(id % 3 = 0, null, cast(id as int)), " +
+ "'s', if(id % 4 = 0, null, concat('s_', cast(id as string)))))"),
+ "nullable array<int> with null elements" ->
+ ("if(id % 6 = 0, null, array(cast(id as int), if(id % 3 = 0, null, cast(id + 1 as int)), " +
+ "cast(id + 2 as int)))"),
+ "struct<array<struct<long,string>>,string>" ->
+ ("named_struct('items', transform(sequence(0, cast(id % 3 as int)), i -> " +
+ "named_struct('k', id + i, 'v', concat('v_', cast(i as string)))), " +
+ "'label', concat('l_', cast(id as string)))"))
+
+ shapes.foreach { case (name, expr) =>
+ withTempPath { dir =>
+ val filename = dir.getCanonicalPath + "/nested.parquet"
+ withSQLConf(CometConf.COMET_ENABLED.key -> "false") {
+ spark
+ .range(numRows.toLong)
+ .selectExpr("id AS key", s"$expr AS c")
+ .write
+ .mode(SaveMode.Overwrite)
+ .parquet(filename)
+ }
+ Seq(5, manyPartitions).foreach { partitionNum =>
+ shuffleDeeplyNestedBenchmark(
Review Comment:
Fixed in 36acd72e8: the native arm turns on round-robin, and each Comet arm's planned shuffle is checked before timing, skipping an arm that falls back. All ten shapes now run native shuffle; the fuzz cases skip it because Spark's scan reads that schema.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org