Skip to content

feat: admit every array and map type in Spark-to-Comet conversion #6726

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
andygrove wants to merge 2 commits into
base: main
Choose a base branch
from
Open

feat: admit every array and map type in Spark-to-Comet conversion #6726

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
106ac88
feat: admit every array and map type in Spark-to-Comet conversion
andygrove Oct 6, 2026
48aeec0
fix: leave calendar intervals out of Spark-to-Comet conversion
andygrove Oct 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 11 additions & 6 deletions docs/source/user-guide/latest/datasources.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,12 +73,17 @@ operator in `spark.comet.sparkToColumnar.supportedOperatorList` by its Spark cla

### Spark-to-Comet conversion types

Spark-to-Comet conversion supports `ARRAY<STRING>` and `MAP<STRING,STRING>` with binary
string semantics, both as top-level fields and inside supported structs. Arrays and maps may
be null; array elements and map values may also be null. Map keys must be non-null.
This applies to Spark row and columnar inputs when conversion is enabled for the source.
Other array element types, other map key/value types, nested collections, and non-binary
string collations remain unsupported at this conversion boundary. Source defaults are unchanged.
Spark-to-Comet conversion supports booleans, integers, floats, decimals, strings, binary, dates
and timestamps, and arrays, maps and structs of them, nested to any depth. Arrays and maps may be
null; array elements and map values may also be null. Map keys must be non-null. This applies to
Spark row and columnar inputs when conversion is enabled for the source. A column with any other
type is not converted, and neither is one with a string that has a non-binary collation, or a
struct that has duplicate field names or no fields, anywhere inside it. The operators above such a
column stay on Spark. Source defaults are unchanged.

Calendar intervals are not converted, at the top level or inside an array, map or struct. Arrow
holds the time part of an interval in nanoseconds, so it cannot hold one of more than about 292
years, which Spark can ([#5279](https://github.com/apache/datafusion-comet/issues/5279)).

This includes row-backed `ExistingRDD` inputs when `spark.comet.convert.rdd.enabled=true`. Spark
still produces the RDD rows; conversion lets eligible downstream operators execute in Comet.
Expand Down
7 changes: 2 additions & 5 deletions docs/source/user-guide/latest/in-memory-cache.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -169,11 +169,8 @@ several times faster, at six times the memory.
The two relations are not comparable to each other — different row counts, and a struct column
carries several values per row.

Array and map columns are deliberately absent from the benchmark, not from the format — the cache
stores and projects them, and `CometInMemoryCacheSuite` covers them. They cannot be measured _here_
because the left column would not exist: it needs Spark's cache scan to bridge into Comet operators,
and `CometSparkToColumnarExec` declines `ArrayType` and `MapType`, so a query projecting one falls
back to Spark row execution above the scan and the two columns stop measuring the same boundary.
Array and map columns are not in the benchmark, though the cache stores and projects them, and
`CometInMemoryCacheSuite` covers them.

Read what this compares carefully. Comet execution is on in both columns, so the aggregation runs
on Comet either way and only the cache-scan boundary moves: on the left, Spark's
Expand Down
7 changes: 3 additions & 4 deletions docs/source/user-guide/latest/tuning/shuffle.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,10 +70,9 @@ Setting `spark.comet.convert.shuffleInput.enabled=true` makes such a shuffle use
converts the child's rows to Arrow, as the `spark.comet.convert.*` settings do for Spark scans, and shuffles the
batches natively. With hash partitioning that is usually faster than columnar shuffle, while with range partitioning
the two come out about even. This applies only where native shuffle supports the partitioning and Comet can convert
all of the child's columns, which rules out calendar intervals, and arrays and maps other than `array<string>` and
`map<string,string>`. A shuffle that hashes a decimal wider than 18 digits, or a string or a value computed from a
string, stays on columnar shuffle, so that it partitions rows as Spark does. The converted batches are allocated on
the JVM outside Comet's memory pools. The setting is disabled by default.
all of the child's columns, which rules out calendar intervals. A shuffle that hashes a decimal wider than 18 digits, or
a string or a value computed from a string, stays on columnar shuffle, so that it partitions rows as Spark does. The
converted batches are allocated on the JVM outside Comet's memory pools. The setting is disabled by default.

### Automatic Revert to Spark Shuffle

Expand Down
15 changes: 11 additions & 4 deletions spark/src/main/scala/org/apache/spark/sql/comet/CometSparkToColumnarExec.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ import org.apache.spark.sql.comet.execution.arrow.{CometArrowStream, CometNative
import org.apache.spark.sql.comet.util.Utils
import org.apache.spark.sql.execution.{RowToColumnarTransition, SparkPlan}
import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
import org.apache.spark.sql.types._
import org.apache.spark.sql.types.{CalendarIntervalType, DataType}
import org.apache.spark.sql.vectorized.ColumnarBatch

import org.apache.comet.{CometConf, DataTypeSupport}
Expand Down Expand Up @@ -138,13 +138,20 @@ object CometSparkToColumnarExec extends CometSink[SparkPlan] with DataTypeSuppor
CometScanWrapper(nativeOp, CometSparkToColumnarExec(op))
}

/**
* Declines calendar intervals at any depth. Arrow holds the time part of an interval in
* nanoseconds, so `IntervalMonthDayNanoWriter` overflows on one of more than about 292 years,
* which Spark accepts (#5279).
*/
override def isTypeSupported(
dt: DataType,
name: String,
fallbackReasons: ListBuffer[String]): Boolean = dt match {
case ArrayType(StringType, _) => true
case MapType(StringType, StringType, _) => true
case _: ArrayType | _: MapType => false

@viirya viirya Oct 10, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

With this case gone, the shared rule admits CalendarIntervalType inside arrays and maps too. IntervalMonthDayNanoWriter multiplies microseconds by 1000 with multiplyExact. So an interval whose time part is longer than about 292 years fails the task with ArithmeticException: long overflow, where Spark returns it. An RDD with an ARRAY<INTERVAL> column holding new CalendarInterval(1, 2, Long.MaxValue / 10) reproduces it under spark.comet.convert.rdd.enabled=true, and so does a map value. Before this PR those columns fell back.

Top-level intervals already had this (#5279), and #6607 rules intervals out of the shuffle-input conversion for the same reason. Could this sink decline CalendarIntervalType with a reason pointing at #5279, at least inside arrays and maps?

@andygrove andygrove Oct 11, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The sink now declines CalendarIntervalType with a reason that links #5279. I made it decline at every level rather than only inside arrays and maps, because the shared rule's recursion calls this same isTypeSupported and the writer overflows on a top-level interval too. On main, the top-level case of the new test fails with the same long overflow. It also matches what #6607 does for the shuffle-input conversion. The cost is that a top-level interval column from an RDD, a cache or a typed Dataset now keeps the operators above it on Spark, but each of those conversions is opt-in.

SparkToColumnar leaves calendar intervals to Spark runs your repro as a top-level value, an array element and a map value, and checks Spark's answer with no conversion in the plan.

case CalendarIntervalType =>

@sunchao sunchao Oct 11, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] [P1] Preserve Spark-compatible decimal partitioning after collection conversion

Removing the collection rejection makes previously correct mixed-shuffle joins silently lose rows. With spark.comet.convert.rdd.enabled=true, shuffle mode auto, AQE and broadcast joins disabled, and 10 shuffle partitions, join two RDD DataFrames containing matching DECIMAL(38,0) keys 0..99. Keep an ARRAY<INT> column on the left and a NullType column on the right in the output. This head converts the left input and chooses native shuffle, while the right retains JVM columnar shuffle. The join returns 11 rows instead of Spark’s 100. Restoring the base gate returns all 100. MAP<INT,INT> reproduces the same loss.

The existing wide-decimal guard recognizes typed Dataset conversions, and hashesDifferentlyFromSpark protects shuffle-input conversion. Neither protects this RDD conversion, so the broader gate exposes #5994 to collection-bearing inputs that previously worked. Please extend the decimal fallback to the affected source conversions, or make native hashing Spark-compatible, and add a mixed-shuffle regression test.

Evidence: Reproduced after a fresh full-reactor JVM compilation on Spark 4.1.3 with the exact-head CI native library. For each i in 0..99, the left row was (new java.math.BigDecimal(i), Seq(i, i + 1)), alternatively (new java.math.BigDecimal(i), Map(i -> (i + 1))). The right row was (new java.math.BigDecimal(i), null) with the second field declared NullType. Both inputs used two RDD partitions. Typed-Dataset and shuffle-input conversion were disabled. Both head cases failed Spark answer comparison with 11 versus 100 rows. Both passed with the base conversion class restored. Reproduction and logs: /tmp/review6726-root-1791737670-8lvip5yf/source/Review6726CurrentJoinProbe.scala, head.log, and base.log. Spark’s supported-version sources hash wide decimals via unscaledValue().toByteArray; Comet’s hash_array_decimal! hashes fixed-width little-endian bytes.

@viirya viirya Oct 12, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can reproduce this on 48aeec051, but it isn't specific to this PR. With the same setup (AQE and broadcast off, 10 shuffle partitions, DECIMAL(38,0) keys 0..99, the right side an RDD with a NullType column on the JVM columnar shuffle), the join returns 11 of 100 rows in each of these:

  • Left is an RDD with (decimal, int), which the conversion took before this PR.
  • Left is an RDD with (decimal, array<string>), which branch-1.1 already converts.
  • No conversion at all. Left is a native Parquet scan, with spark.comet.convert.rdd.enabled=false.

So this is #3079. Any native shuffle hashing a wide-decimal key next to a JVM columnar one loses rows. A guard on the conversion paths would not cover the last case, while #6005 and #6548 fix it at the shuffle.

This PR does widen what reaches the bug. Array- and map-bearing inputs used to stay on the columnar shuffle on both sides of such a join. @andygrove could this land after #6005 or #6548? Once it's rebased on one of them, a join like sunchao's, with an array<int> column on the converted side, would be a good regression test to add here.

fallbackReasons += s"Unsupported $name of type $dt: Arrow holds the time part of an " +
"interval in nanoseconds, which overflows past about 292 years " +
"(https://github.com/apache/datafusion-comet/issues/5279)"
false
case _ => super.isTypeSupported(dt, name, fallbackReasons)
}
}
Loading
Loading