andygrove commented on code in PR #5766:
URL: https://github.com/apache/datafusion-comet/pull/5766#discussion_r3952969995
##########
spark/src/main/scala/org/apache/comet/serde/literals.scala:
##########
@@ -41,27 +43,20 @@ object CometLiteral extends CometExpressionSerde[Literal]
with Logging {
"Not all data types are supported for literal values")
override def getSupportLevel(expr: Literal): SupportLevel = {
+ val serializable = expr.dataType match {
+ // A null literal carries only its type on the wire, so any type
serializeDataType handles.
+ case _ if expr.value == null => supportedDataType(expr.dataType,
allowComplex = true)
+ case ArrayType(elementType, _) =>
listLiteralElementSupported(elementType)
Review Comment:
`hasUnwritableValue` only runs inside `canExpandComplexLiteral`, so it
covers the rebuilt `Create*` tree but not the two paths that serialize a string
literal directly. Both of those go through `UTF8String.toString`, which
substitutes U+FFFD:
```sql
SELECT _1, hex(element_at(array(CAST(X'FF' AS STRING), 'b', 'c'), _1)) FROM
tbl
-- spark: FF, 62, 63 comet: EFBFBD, 62, 63
SELECT _1, hex(concat(CAST(X'FF' AS STRING), CAST(_1 AS STRING))) FROM tbl
-- spark: FF31, FF32, FF33 comet: EFBFBD31, EFBFBD32, EFBFBD33
```
The second one has nothing to do with complex literals at all, it is the
scalar `_: StringType` branch of `convert`. The overlong form and a nested
`ARRAY<ARRAY<STRING>>` behave the same way. A plain answer comparison calls all
of these a match, because both sides render as the replacement character, so it
takes `hex()` to see it.
Both reproduce on the merge base, so they are pre-existing rather than
something this PR introduced, but you now have `isValidUtf8` in hand and the
fix looks like one more conjunct here:
```scala
if ((serializable && !hasUnwritableValue(expr.value, expr.dataType)) ||
canExpandComplexLiteral(expr)) {
```
with the `expr.value == null` arm short-circuiting first as it already does.
Since `string_val` on the wire is a proto `string` it cannot carry the raw
bytes, so declining looks like the only option that keeps Spark's semantics. Is
there a reason to leave these two paths for a separate PR?
##########
spark/src/main/scala/org/apache/comet/serde/arrays.scala:
##########
@@ -690,7 +667,12 @@ object CometElementAt extends
CometExpressionSerde[ElementAt] {
// evaluated on the selected rows (DataFusion's CaseExpr filters the batch
before the THEN
// branch), reproducing the short-circuit. Mirrors the CASE-WHEN idiom in
CometArrayAppend /
// CometSize; the ELSE null literal carries the result type, as in
CometArraysZip.
- if (expr.failOnError && expr.left.nullable) {
+ //
+ // The guard serializes `left` a second time and runs the THEN branch over
a different row
+ // selection, so a stateful operand (`rand()`,
`monotonically_increasing_id()`) would advance
+ // its state twice and silently move values and NULLs. Restrict it to
deterministic operands;
+ // the rest keep the pre-existing eager-key behaviour.
+ if (expr.failOnError && expr.left.nullable && expr.left.deterministic) {
Review Comment:
The `expr.left.deterministic` guard is right, and the reasoning above it
applies verbatim to three other serdes that build the same `CASE WHEN <child>
IS NOT NULL` over a second serialization of the same child. On this branch,
with a 16-row table and a native projection asserted:
```sql
SELECT _1, size(IF(monotonically_increasing_id() % 2 = 0, array(1),
CAST(NULL AS ARRAY<INT>))) FROM tbl
```
Spark returns `1` for every row whose array is non-NULL. Comet returns `-1`
on 5 of the 16 rows. `arrays_zip` over the same operand returns `[null,2]`
where Spark returns `[1,2]`, and `map_from_arrays` returns `NULL` where Spark
returns `Map(1 -> 2)`. All three reproduce on the merge base, so they are not
regressions from this PR, but they are the same bug you are fixing here, and
they are silent wrong answers rather than errors.
`CometArrayAppend` has the same shape. It is unreachable on Spark 4.x since
`ArrayAppend` is `RuntimeReplaceable` there, but it is live on 3.4 and 3.5, so
I could not exercise it and that one is by inspection.
Would you extend the same narrow decline to those four, or file it so the
four cases do not drift apart? `CometSize` and `CometArraysZip` are not ANSI
short-circuits, so the guard is not removable for them the way it is here.
Declining a nondeterministic child looks like the cheapest safe option, and for
`CometSize` specifically `coalesce(size(x), <legacy sentinel>)` would get there
with a single serialization.
##########
spark/src/main/scala/org/apache/comet/serde/literals.scala:
##########
@@ -261,47 +272,131 @@ object CometLiteral extends
CometExpressionSerde[Literal] with Logging {
* `CometCreateMap` hands the whole rebuilt `CreateMap` to the JVM
codegen dispatcher, so
* Spark's own code builds the struct.
* - Folded maps with duplicate keys, see [[hasDuplicateMapKeys]].
+ * - Values no Arrow encoding of the rebuilt tree can carry, see
[[hasUnwritableValue]].
*/
private def expandComplexLiteral(expr: Literal): Option[Expression] = {
if (!canExpandComplexLiteral(expr)) return None
expr.dataType match {
case ArrayType(et, containsNull) =>
val arr = expr.value.asInstanceOf[ArrayData]
- val elements = (0 until arr.numElements())
- .map(i => withNullability(literalAt(arr, i, et), containsNull))
- Some(CreateArray(elements, useStringTypeWhenEmpty = false))
+ if (arr.numElements() == 0) {
+ Some(emptyTypedArray(et, containsNull))
+ } else {
+ val elements = (0 until arr.numElements())
+ .map(i => withNullability(literalAt(arr, i, et), containsNull))
+ Some(CreateArray(elements, useStringTypeWhenEmpty = false))
+ }
case MapType(kt, vt, valueContainsNull) =>
val mapData = expr.value.asInstanceOf[MapData]
- val keys = mapData.keyArray()
- val values = mapData.valueArray()
- val children = (0 until keys.numElements()).flatMap(i =>
- Seq(
- literalAt(keys, i, kt),
- withNullability(literalAt(values, i, vt), valueContainsNull)))
- Some(CreateMap(children, useStringTypeWhenEmpty = false))
+ if (mapData.numElements() == 0) {
+ // `MapFromArrays` recovers the declared key/value types from its
two array children;
+ // a childless `CreateMap` would report `MapType(NullType,
NullType)`. Arrow map keys
+ // are never null, so the key array is non-nullable.
+ Some(
+ MapFromArrays(
+ emptyTypedArray(kt, containsNull = false),
+ emptyTypedArray(vt, valueContainsNull)))
+ } else {
+ val keys = mapData.keyArray()
+ val values = mapData.valueArray()
+ val children = (0 until keys.numElements()).flatMap(i =>
+ Seq(
+ literalAt(keys, i, kt),
+ withNullability(literalAt(values, i, vt), valueContainsNull)))
+ Some(CreateMap(children, useStringTypeWhenEmpty = false))
+ }
case _ => None
}
}
+ /**
+ * An empty `ArrayType(elementType, containsNull)` value. `CometCreateArray`
turns a childless
+ * `CreateArray` into an empty `ArrayType(NullType)` list literal (its
documented workaround for
+ * DataFusion's zero-argument `make_array`), and the cast then stamps the
declared element type
+ * onto it. Only widens an empty container's type, so it cannot change any
value.
+ */
+ private def emptyTypedArray(elementType: DataType, containsNull: Boolean):
Expression =
Review Comment:
`literals.scala` is at 494 lines for nine small private methods, and this PR
adds around 90 more comment lines. The parts that carry real information are
worth keeping, and I would not want to lose them: why `MapFromArrays` instead
of a childless `CreateMap`, why `StringType` is matched as a stable identifier,
where the interval bound comes from. The parts that restate the signature could
go. This is a two-line expression under a five-line doc, and `isValidUtf8`'s
first sentence repeats its own name. Trimming those would make the load-bearing
comments easier to find.
##########
spark/src/main/scala/org/apache/comet/serde/literals.scala:
##########
@@ -261,47 +272,131 @@ object CometLiteral extends
CometExpressionSerde[Literal] with Logging {
* `CometCreateMap` hands the whole rebuilt `CreateMap` to the JVM
codegen dispatcher, so
* Spark's own code builds the struct.
* - Folded maps with duplicate keys, see [[hasDuplicateMapKeys]].
+ * - Values no Arrow encoding of the rebuilt tree can carry, see
[[hasUnwritableValue]].
*/
private def expandComplexLiteral(expr: Literal): Option[Expression] = {
if (!canExpandComplexLiteral(expr)) return None
expr.dataType match {
case ArrayType(et, containsNull) =>
val arr = expr.value.asInstanceOf[ArrayData]
- val elements = (0 until arr.numElements())
- .map(i => withNullability(literalAt(arr, i, et), containsNull))
- Some(CreateArray(elements, useStringTypeWhenEmpty = false))
+ if (arr.numElements() == 0) {
+ Some(emptyTypedArray(et, containsNull))
+ } else {
+ val elements = (0 until arr.numElements())
+ .map(i => withNullability(literalAt(arr, i, et), containsNull))
+ Some(CreateArray(elements, useStringTypeWhenEmpty = false))
+ }
case MapType(kt, vt, valueContainsNull) =>
val mapData = expr.value.asInstanceOf[MapData]
- val keys = mapData.keyArray()
- val values = mapData.valueArray()
- val children = (0 until keys.numElements()).flatMap(i =>
- Seq(
- literalAt(keys, i, kt),
- withNullability(literalAt(values, i, vt), valueContainsNull)))
- Some(CreateMap(children, useStringTypeWhenEmpty = false))
+ if (mapData.numElements() == 0) {
+ // `MapFromArrays` recovers the declared key/value types from its
two array children;
+ // a childless `CreateMap` would report `MapType(NullType,
NullType)`. Arrow map keys
+ // are never null, so the key array is non-nullable.
+ Some(
+ MapFromArrays(
+ emptyTypedArray(kt, containsNull = false),
+ emptyTypedArray(vt, valueContainsNull)))
+ } else {
+ val keys = mapData.keyArray()
+ val values = mapData.valueArray()
+ val children = (0 until keys.numElements()).flatMap(i =>
+ Seq(
+ literalAt(keys, i, kt),
+ withNullability(literalAt(values, i, vt), valueContainsNull)))
+ Some(CreateMap(children, useStringTypeWhenEmpty = false))
+ }
case _ => None
}
}
+ /**
+ * An empty `ArrayType(elementType, containsNull)` value. `CometCreateArray`
turns a childless
+ * `CreateArray` into an empty `ArrayType(NullType)` list literal (its
documented workaround for
+ * DataFusion's zero-argument `make_array`), and the cast then stamps the
declared element type
+ * onto it. Only widens an empty container's type, so it cannot change any
value.
+ */
+ private def emptyTypedArray(elementType: DataType, containsNull: Boolean):
Expression =
+ Cast(CreateArray(Nil, useStringTypeWhenEmpty = false),
ArrayType(elementType, containsNull))
+
/**
* Cheap admission test that mirrors [[expandComplexLiteral]] without
materializing the rebuilt
* `Create*` tree, so `getSupportLevel` can probe a large folded literal
without allocating the
- * N `Literal`s `convert` would immediately rebuild. Declines a null value
or empty top-level
- * container (no children to recover the element type), a folded map with
duplicate keys (see
- * [[hasDuplicateMapKeys]]), any unsupported or non-orderable map key type
at any nesting level
- * (see [[mapKeyTypesExpandable]]), and an array of structs
([[needsExpansion]] stops at a
- * `StructType`). [[expandComplexLiteral]] gates on this, so the two cannot
diverge.
+ * N `Literal`s `convert` would immediately rebuild. Declines a null value,
a folded map with
+ * duplicate keys (see [[hasDuplicateMapKeys]]), any unsupported or
non-orderable map key type
+ * at any nesting level (see [[mapKeyTypesExpandable]]), an array of structs
([[needsExpansion]]
+ * stops at a `StructType`), and a value Arrow cannot hold
([[hasUnwritableValue]]).
+ * [[expandComplexLiteral]] gates on this, so the two cannot diverge.
*/
private def canExpandComplexLiteral(expr: Literal): Boolean = {
if (expr.value == null || !mapKeyTypesExpandable(expr.dataType)) return
false
- expr.dataType match {
- case ArrayType(et, _) if needsExpansion(et) =>
- expr.value.asInstanceOf[ArrayData].numElements() > 0
+ val expandableShape = expr.dataType match {
+ case ArrayType(et, _) => needsExpansion(et)
case MapType(kt, _, _) =>
- val mapData = expr.value.asInstanceOf[MapData]
- mapData.numElements() > 0 && !hasDuplicateMapKeys(mapData.keyArray(),
kt)
+ !hasDuplicateMapKeys(expr.value.asInstanceOf[MapData].keyArray(), kt)
case _ => false
}
+ // Last, because it is the only check here that walks the whole folded
value.
+ expandableShape && !hasUnwritableValue(expr.value, expr.dataType)
Review Comment:
The doc on this method still says it is a cheap probe that lets
`getSupportLevel` avoid materializing the value, but `hasUnwritableValue` now
walks every scalar in it, and `canExpandComplexLiteral` runs from both
`getSupportLevel` and `expandComplexLiteral`. Each string also costs a fresh
`CharsetDecoder` and a `byte[]` copy out of `getBytes`.
This is planning-time only and I did not measure it as a problem, so it may
well be fine. But since the doc comment on `isValidUtf8` already notes
`UTF8String.isValid` exists on 4.0+, routing through the version shim with the
decoder as the 3.x fallback would drop both allocations and let the doc stay
true.
##########
spark/src/main/scala/org/apache/comet/serde/literals.scala:
##########
@@ -226,6 +223,22 @@ object CometLiteral extends CometExpressionSerde[Literal]
with Logging {
listLiteralBuilder
}
+ /**
+ * Element types [[makeListLiteral]] has a branch for. Anything else has to
be expanded or
+ * declined before `convert` reaches it, or the missing branch raises a
`MatchError` mid-plan.
+ * The arms mirror that match exactly, down to `StringType` being matched as
a stable identifier
+ * so that a non-default collation (whose `equals` compares `collationId`)
is declined here
+ * rather than falling into the missing branch.
+ */
+ private def listLiteralElementSupported(dataType: DataType): Boolean =
dataType match {
Review Comment:
This mirror is carrying the invariant the whole PR exists to establish, and
a drift reproduces the `[INTERNAL_ERROR]` planning failure rather than a
fallback. I confirmed that on the merge base: `array(make_interval(1))`,
`array('a' COLLATE UTF8_LCASE)` and `array(TIME'12:00:00')` all fail planning
there and all fall back cleanly here.
Right now the invariant rests on a comment on each side. Could a small
table-driven test pin it? For each `dt` in a fixed list of Spark types, assert
that `listLiteralElementSupported(dt)` agrees with whether a one-element
`makeListLiteral` succeeds. That way the next `makeListLiteral` arm, or the
next Spark release that adds an `AtomicType`, fails a test instead of a query
plan.
##########
spark/src/test/resources/sql-tests/expressions/map/create_map.sql:
##########
@@ -28,3 +28,14 @@ SELECT map(k, v) FROM test_create_map
query
SELECT map(1, 'a', 2, 'b'), map('x', array(1, 2), 'y', array(3))
+
+-- Arrow addresses a struct vector's children by name, so the two `x` fields
collapse into one
+-- child and the dispatcher's generated writer NPEs on the missing ordinal-1
vector. Spark keeps
+-- both values. A struct that is only a map value never reaches
`CometCreateNamedStruct`'s check.
+-- https://github.com/apache/datafusion-comet/issues/5544
+query expect_fallback(codegen dispatch: unsupported output type)
+SELECT map(1, named_struct('x', 10, 'x', 20))
Review Comment:
This covers the dispatcher path, but since the SQL harness excludes
`ConstantFolding` it does not cover the folded `Literal` that #5544 actually
reports, and the `array(map(...))` wrapper from the report is not covered
anywhere. Both do fall back correctly on this branch, and both NPE on the merge
base, so this is only a missing regression test.
Would you add the two folded spellings to `CometArrayExpressionSuite`,
alongside the other folded-literal cases that are there precisely because
folding is on?
```scala
Seq(
"array(map(1, named_struct('x', 10, 'x', 20)))" -> "ArrayType",
"map(1, named_struct('x', 10, 'x', 20))" -> "MapType").foreach { case (v,
dt) =>
checkSparkAnswerAndFallbackReason(s"SELECT _1 AS id, $v AS v FROM tbl",
s"Unsupported data type $dt")
}
```
A case-differing control would be worth having too. `named_struct('x', 10,
'X', 20)` still runs natively, which documents that `fieldNames.distinct` is
deliberately exact-name rather than case-insensitive, and that this matches how
Arrow keys a struct vector's 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]
