andygrove commented on code in PR #6726:
URL: https://github.com/apache/datafusion-comet/pull/6726#discussion_r4242035301
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometSparkToColumnarExec.scala:
##########
@@ -137,14 +135,4 @@ object CometSparkToColumnarExec extends
CometSink[SparkPlan] with DataTypeSuppor
op: SparkPlan): CometNativeExec = {
CometScanWrapper(nativeOp, CometSparkToColumnarExec(op))
}
-
- 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
Review Comment:
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.
##########
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala:
##########
@@ -4150,43 +4150,190 @@ class CometExecSuite extends CometTestBase {
})
}
- test("SparkToColumnar admits only binary string arrays and maps through the
collection gate") {
- for (nullable <- Seq(false, true); containsNull <- Seq(false, true)) {
- for (dataType <- Seq(
- ArrayType(StringType, containsNull),
- MapType(StringType, StringType, containsNull))) {
- val field = StructField("tags", dataType, nullable)
- Seq(
- StructType(Seq(field)),
- StructType(Seq(StructField("nested", StructType(Seq(field))))))
- .foreach { schema =>
- assert(CometSparkToColumnarExec.isSchemaSupported(schema,
ListBuffer.empty))
- }
- }
- }
- val unsupported = Seq(
+ test("SparkToColumnar admits arrays and maps of every type the shared rule
supports") {
+ val struct = StructType(Seq(StructField("i", IntegerType),
StructField("s", StringType)))
+ val admitted = Seq(
+ ArrayType(StringType),
ArrayType(IntegerType),
+ ArrayType(DecimalType(38, 10)),
+ ArrayType(TimestampType),
ArrayType(BinaryType),
ArrayType(ArrayType(StringType)),
- ArrayType(StructType(Seq(StructField("s", StringType)))),
- MapType(StringType, IntegerType),
- MapType(IntegerType, StringType),
- MapType(StringType, ArrayType(StringType)),
- MapType(StringType, StructType(Seq(StructField("s", StringType))))) ++
- (if (isSpark40Plus)
+ ArrayType(struct),
+ ArrayType(MapType(StringType, IntegerType)),
+ MapType(StringType, StringType),
+ MapType(IntegerType, IntegerType),
+ MapType(DateType, DecimalType(20, 2)),
+ MapType(StringType, ArrayType(IntegerType)),
+ MapType(IntegerType, struct),
+ MapType(struct, StringType),
+ MapType(ArrayType(IntegerType), IntegerType),
+ MapType(StringType, MapType(IntegerType, StringType)))
+ for (nullable <- Seq(false, true); containsNull <- Seq(false, true);
dataType <- admitted) {
+ val collection = dataType match {
+ case a: ArrayType => a.copy(containsNull = containsNull)
+ case m: MapType => m.copy(valueContainsNull = containsNull)
+ }
+ val field = StructField("c", collection, nullable)
+ Seq(StructType(Seq(field)), StructType(Seq(StructField("nested",
StructType(Seq(field))))))
+ .foreach { schema =>
+ val reasons = ListBuffer.empty[String]
+ assert(CometSparkToColumnarExec.isSchemaSupported(schema, reasons),
s"$collection")
+ assert(reasons.isEmpty, reasons)
+ }
+ }
+
+ // The shared rule looks inside a collection and declines these for what
it finds there. A
+ // decline of the whole collection would record no reason, which is how
the two differ.
+ val duplicate = StructType(Seq(StructField("a", LongType),
StructField("a", LongType)))
+ val declined = Seq(
Review Comment:
Added `ArrayType(CalendarIntervalType)` and `MapType(StringType,
CalendarIntervalType)` to the list, plus a check that a top-level interval is
declined with the #5279 reason.
##########
docs/source/user-guide/latest/datasources.md:
##########
@@ -73,12 +73,13 @@ 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,
+timestamps and calendar intervals, and arrays, maps and structs of them,
nested to any depth.
Review Comment:
Dropped them from the list, and added a short paragraph saying calendar
intervals aren't converted at any level, with the 292-year limit and a link to
#5279. `tuning/shuffle.md` had a stale sentence from #6607 too: it said the
shuffle-input conversion rules out arrays and maps other than `array<string>`
and `map<string,string>`. That's fixed as well.
##########
spark/src/test/scala/org/apache/comet/exec/CometExecSuite.scala:
##########
@@ -4150,43 +4150,190 @@ class CometExecSuite extends CometTestBase {
})
}
- test("SparkToColumnar admits only binary string arrays and maps through the
collection gate") {
- for (nullable <- Seq(false, true); containsNull <- Seq(false, true)) {
- for (dataType <- Seq(
- ArrayType(StringType, containsNull),
- MapType(StringType, StringType, containsNull))) {
- val field = StructField("tags", dataType, nullable)
- Seq(
- StructType(Seq(field)),
- StructType(Seq(StructField("nested", StructType(Seq(field))))))
- .foreach { schema =>
- assert(CometSparkToColumnarExec.isSchemaSupported(schema,
ListBuffer.empty))
- }
- }
- }
- val unsupported = Seq(
+ test("SparkToColumnar admits arrays and maps of every type the shared rule
supports") {
+ val struct = StructType(Seq(StructField("i", IntegerType),
StructField("s", StringType)))
+ val admitted = Seq(
+ ArrayType(StringType),
ArrayType(IntegerType),
+ ArrayType(DecimalType(38, 10)),
+ ArrayType(TimestampType),
ArrayType(BinaryType),
ArrayType(ArrayType(StringType)),
- ArrayType(StructType(Seq(StructField("s", StringType)))),
- MapType(StringType, IntegerType),
- MapType(IntegerType, StringType),
- MapType(StringType, ArrayType(StringType)),
- MapType(StringType, StructType(Seq(StructField("s", StringType))))) ++
- (if (isSpark40Plus)
+ ArrayType(struct),
+ ArrayType(MapType(StringType, IntegerType)),
+ MapType(StringType, StringType),
+ MapType(IntegerType, IntegerType),
+ MapType(DateType, DecimalType(20, 2)),
+ MapType(StringType, ArrayType(IntegerType)),
+ MapType(IntegerType, struct),
+ MapType(struct, StringType),
+ MapType(ArrayType(IntegerType), IntegerType),
+ MapType(StringType, MapType(IntegerType, StringType)))
+ for (nullable <- Seq(false, true); containsNull <- Seq(false, true);
dataType <- admitted) {
+ val collection = dataType match {
+ case a: ArrayType => a.copy(containsNull = containsNull)
+ case m: MapType => m.copy(valueContainsNull = containsNull)
+ }
+ val field = StructField("c", collection, nullable)
+ Seq(StructType(Seq(field)), StructType(Seq(StructField("nested",
StructType(Seq(field))))))
+ .foreach { schema =>
+ val reasons = ListBuffer.empty[String]
+ assert(CometSparkToColumnarExec.isSchemaSupported(schema, reasons),
s"$collection")
+ assert(reasons.isEmpty, reasons)
+ }
+ }
+
+ // The shared rule looks inside a collection and declines these for what
it finds there. A
+ // decline of the whole collection would record no reason, which is how
the two differ.
+ val duplicate = StructType(Seq(StructField("a", LongType),
StructField("a", LongType)))
+ val declined = Seq(
+ ArrayType(duplicate),
+ MapType(LongType, duplicate),
+ ArrayType(ArrayType(duplicate)),
+ ArrayType(NullType),
+ MapType(IntegerType, ArrayType(NullType)),
+ ArrayType(DayTimeIntervalType()),
+ MapType(StringType, YearMonthIntervalType())) ++
+ (if (isSpark40Plus) {
Seq(
- DataType.fromDDL("ARRAY<STRING COLLATE UTF8_LCASE>"),
- DataType.fromDDL("MAP<STRING COLLATE UTF8_LCASE,STRING>"),
- DataType.fromDDL("MAP<STRING,STRING COLLATE UTF8_LCASE>"))
- else Seq.empty)
- unsupported.foreach { dataType =>
+ "ARRAY<STRING COLLATE UTF8_LCASE>",
+ "ARRAY<ARRAY<STRING COLLATE UTF8_LCASE>>",
+ "ARRAY<STRUCT<s: STRING COLLATE UTF8_LCASE>>",
+ "MAP<STRING COLLATE UTF8_LCASE,STRING>",
+ "MAP<STRING,STRING COLLATE UTF8_LCASE>",
+ "MAP<INT,ARRAY<STRING COLLATE UTF8_LCASE>>").map(DataType.fromDDL)
+ } else Seq.empty)
+ declined.foreach { dataType =>
val schema = StructType(Seq(StructField("value", dataType)))
- assert(!CometSparkToColumnarExec.isSchemaSupported(schema,
ListBuffer.empty), dataType)
- assert(
- !CometSparkToColumnarExec.isSchemaSupported(
- StructType(Seq(StructField("nested", schema))),
- ListBuffer.empty),
- dataType)
+ Seq(schema, StructType(Seq(StructField("nested", schema)))).foreach {
declinedSchema =>
+ val reasons = ListBuffer.empty[String]
+ assert(!CometSparkToColumnarExec.isSchemaSupported(declinedSchema,
reasons), dataType)
+ assert(reasons.nonEmpty, s"$dataType was declined without a reason")
+ }
+ }
+ }
+
+ // A column of each array and map shape the conversion admits, as its type,
an expression over
+ // `id` that builds it, and one that reads an element of it. The column has,
in some rows, a
+ // null collection, an empty one, and elements that are null.
+ private val sparkToColumnarShapes: Seq[(String, String, String)] = {
Review Comment:
Added `SparkToColumnar converts arrays and maps whose elements and values
are not null`. It reads a typed `Seq[Int]`/`Map[Int, Int]` Dataset and an RDD
with the same schema, asserts that the schema still has `containsNull = false`
and `valueContainsNull = false`, and runs `size`, `slice`, `map_entries`, and
on 3.5+ `array_insert` and `array_prepend` natively, then a native shuffle. I
left Parquet out, since Spark builds a file relation with
`dataSchema.asNullable`, so a Parquet read never has non-null children.
--
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]
