sunchao commented on code in PR #5452:
URL: https://github.com/apache/datafusion-comet/pull/5452#discussion_r3845993193
##########
spark/src/main/scala/org/apache/comet/serde/literals.scala:
##########
@@ -215,4 +225,82 @@ object CometLiteral extends CometExpressionSerde[Literal]
with Logging {
}
listLiteralBuilder
}
+
+ /**
+ * True when a non-null Literal of this type is not encodable in the native
`Literal` proto and
+ * `convert` should try to rebuild it from primitive-typed Literals. The
native proto today
+ * carries scalars and nested `ListLiteral`s (arrays of arrays / arrays of
scalars). It does not
+ * carry Map or Struct values, so any Literal whose type contains a
`MapType` or a `StructType`
+ * needs to be expanded before serialization.
+ */
+ private def needsExpansion(dataType: DataType): Boolean = dataType match {
+ case _: MapType | _: StructType => true
+ case ArrayType(et, _) => needsExpansion(et)
+ case _ => false
+ }
+
+ /**
+ * True when the Literal is a non-null complex value that we can rebuild
from primitive Literals
+ * via [[expandComplexLiteral]]. Empty top-level containers are excluded
because a synthesized
+ * `Create[Array|Map]` with no children cannot recover the original element
type.
+ */
+ private def canExpandComplexLiteral(expr: Literal): Boolean = {
+ val value = expr.value
+ if (value == null || !needsExpansion(expr.dataType)) return false
+ expr.dataType match {
+ case _: ArrayType => value.asInstanceOf[ArrayData].numElements() > 0
+ case _: MapType => value.asInstanceOf[MapData].numElements() > 0
+ case _: StructType => true
+ case _ => false
+ }
+ }
+
+ /**
+ * Rebuild a folded complex Literal as an equivalent tree of `CreateArray` /
`CreateMap` /
+ * `CreateNamedStruct` over primitive-typed Literals. Callers must gate on
+ * [[canExpandComplexLiteral]] so `dataType` is one of the three complex
cases and top-level
+ * containers are non-empty.
+ */
+ private def expandComplexLiteral(value: Any, dataType: DataType): Expression
=
+ dataType match {
+ case ArrayType(et, _) =>
+ val arr = value.asInstanceOf[ArrayData]
+ val elems = (0 until arr.numElements()).map { i =>
+ if (arr.isNullAt(i)) Literal(null, et)
+ else asNullable(Literal(arr.get(i, et), et))
+ }
+ CreateArray(elems, useStringTypeWhenEmpty = false)
+ case MapType(kt, vt, _) =>
+ val mapData = value.asInstanceOf[MapData]
+ val keys = mapData.keyArray()
+ val vals = mapData.valueArray()
+ val children = (0 until keys.numElements()).flatMap { i =>
+ val k = if (keys.isNullAt(i)) Literal(null, kt) else
Literal(keys.get(i, kt), kt)
+ val v =
+ if (vals.isNullAt(i)) Literal(null, vt)
+ else asNullable(Literal(vals.get(i, vt), vt))
+ Seq(k, v)
+ }
+ CreateMap(children, useStringTypeWhenEmpty = false)
Review Comment:
[P2] Preserve existing map entries without reapplying constructor
deduplication
A folded `MapData` need not have been produced by `CreateMap`. For example,
Spark folds `from_json('{"a":1,"a":2}', 'MAP<STRING,INT>')` into a map literal
containing keys `[a,a]` and values `[1,2]`, and Spark/the base serializer
execute it successfully. Rebuilding that value as `CreateMap` sends it through
`ArrayBasedMapBuilder` again, so projecting it alongside a column from a
Parquet table now throws `[DUPLICATED_MAP_KEY]` under the default policy. I
reproduced the regression on Spark 3.5.9 and 4.1.3 with normal folding and a
native Comet projection. Please preserve the literal's existing entries
directly, or retain fallback when reconstruction would change their semantics.
##########
spark/src/main/scala/org/apache/comet/serde/literals.scala:
##########
@@ -215,4 +225,82 @@ object CometLiteral extends CometExpressionSerde[Literal]
with Logging {
}
listLiteralBuilder
}
+
+ /**
+ * True when a non-null Literal of this type is not encodable in the native
`Literal` proto and
+ * `convert` should try to rebuild it from primitive-typed Literals. The
native proto today
+ * carries scalars and nested `ListLiteral`s (arrays of arrays / arrays of
scalars). It does not
+ * carry Map or Struct values, so any Literal whose type contains a
`MapType` or a `StructType`
+ * needs to be expanded before serialization.
+ */
+ private def needsExpansion(dataType: DataType): Boolean = dataType match {
+ case _: MapType | _: StructType => true
+ case ArrayType(et, _) => needsExpansion(et)
+ case _ => false
+ }
+
+ /**
+ * True when the Literal is a non-null complex value that we can rebuild
from primitive Literals
+ * via [[expandComplexLiteral]]. Empty top-level containers are excluded
because a synthesized
+ * `Create[Array|Map]` with no children cannot recover the original element
type.
+ */
+ private def canExpandComplexLiteral(expr: Literal): Boolean = {
+ val value = expr.value
+ if (value == null || !needsExpansion(expr.dataType)) return false
+ expr.dataType match {
+ case _: ArrayType => value.asInstanceOf[ArrayData].numElements() > 0
+ case _: MapType => value.asInstanceOf[MapData].numElements() > 0
+ case _: StructType => true
+ case _ => false
+ }
+ }
+
+ /**
+ * Rebuild a folded complex Literal as an equivalent tree of `CreateArray` /
`CreateMap` /
+ * `CreateNamedStruct` over primitive-typed Literals. Callers must gate on
+ * [[canExpandComplexLiteral]] so `dataType` is one of the three complex
cases and top-level
+ * containers are non-empty.
+ */
+ private def expandComplexLiteral(value: Any, dataType: DataType): Expression
=
+ dataType match {
+ case ArrayType(et, _) =>
+ val arr = value.asInstanceOf[ArrayData]
+ val elems = (0 until arr.numElements()).map { i =>
+ if (arr.isNullAt(i)) Literal(null, et)
+ else asNullable(Literal(arr.get(i, et), et))
+ }
+ CreateArray(elems, useStringTypeWhenEmpty = false)
+ case MapType(kt, vt, _) =>
+ val mapData = value.asInstanceOf[MapData]
+ val keys = mapData.keyArray()
+ val vals = mapData.valueArray()
+ val children = (0 until keys.numElements()).flatMap { i =>
+ val k = if (keys.isNullAt(i)) Literal(null, kt) else
Literal(keys.get(i, kt), kt)
+ val v =
+ if (vals.isNullAt(i)) Literal(null, vt)
+ else asNullable(Literal(vals.get(i, vt), vt))
+ Seq(k, v)
+ }
+ CreateMap(children, useStringTypeWhenEmpty = false)
+ case StructType(fields) =>
+ val row = value.asInstanceOf[InternalRow]
+ val children = fields.zipWithIndex.flatMap { case (f, i) =>
+ val v =
+ if (row.isNullAt(i)) Literal(null, f.dataType)
+ else asNullable(Literal(row.get(i, f.dataType), f.dataType))
+ Seq(Literal(f.name), v)
+ }
+ CreateNamedStruct(children.toSeq)
Review Comment:
[P2] Preserve scalar/batch cardinality for reconstructed structs
With normal constant folding, `SELECT id, named_struct('a', 1) FROM t` now
reaches native `CreateNamedStruct` instead of retaining Spark projection
fallback. For all-scalar children, its evaluator calls
`ColumnarValue::values_to_arrays` and returns a one-row `StructArray`,
regardless of the input batch size. With three rows in one Parquet batch, the
query fails with `Array length 1 does not match expected length 3`; arrays
containing such structs similarly fail the `make_array` row-count check. I
reproduced this on Spark 3.5.9 and 4.1.3, while the exact base literal
serializer returns all three rows correctly. Please preserve scalar
semantics/broadcast to the input batch size, or retain fallback for these
literals until the native constructor supports them.
##########
spark/src/main/scala/org/apache/comet/serde/literals.scala:
##########
@@ -215,4 +225,82 @@ object CometLiteral extends CometExpressionSerde[Literal]
with Logging {
}
listLiteralBuilder
}
+
+ /**
+ * True when a non-null Literal of this type is not encodable in the native
`Literal` proto and
+ * `convert` should try to rebuild it from primitive-typed Literals. The
native proto today
+ * carries scalars and nested `ListLiteral`s (arrays of arrays / arrays of
scalars). It does not
+ * carry Map or Struct values, so any Literal whose type contains a
`MapType` or a `StructType`
+ * needs to be expanded before serialization.
+ */
+ private def needsExpansion(dataType: DataType): Boolean = dataType match {
+ case _: MapType | _: StructType => true
+ case ArrayType(et, _) => needsExpansion(et)
+ case _ => false
+ }
+
+ /**
+ * True when the Literal is a non-null complex value that we can rebuild
from primitive Literals
+ * via [[expandComplexLiteral]]. Empty top-level containers are excluded
because a synthesized
+ * `Create[Array|Map]` with no children cannot recover the original element
type.
+ */
+ private def canExpandComplexLiteral(expr: Literal): Boolean = {
+ val value = expr.value
+ if (value == null || !needsExpansion(expr.dataType)) return false
+ expr.dataType match {
+ case _: ArrayType => value.asInstanceOf[ArrayData].numElements() > 0
+ case _: MapType => value.asInstanceOf[MapData].numElements() > 0
+ case _: StructType => true
+ case _ => false
+ }
+ }
+
+ /**
+ * Rebuild a folded complex Literal as an equivalent tree of `CreateArray` /
`CreateMap` /
+ * `CreateNamedStruct` over primitive-typed Literals. Callers must gate on
+ * [[canExpandComplexLiteral]] so `dataType` is one of the three complex
cases and top-level
+ * containers are non-empty.
+ */
+ private def expandComplexLiteral(value: Any, dataType: DataType): Expression
=
+ dataType match {
+ case ArrayType(et, _) =>
+ val arr = value.asInstanceOf[ArrayData]
+ val elems = (0 until arr.numElements()).map { i =>
+ if (arr.isNullAt(i)) Literal(null, et)
+ else asNullable(Literal(arr.get(i, et), et))
+ }
+ CreateArray(elems, useStringTypeWhenEmpty = false)
+ case MapType(kt, vt, _) =>
+ val mapData = value.asInstanceOf[MapData]
+ val keys = mapData.keyArray()
+ val vals = mapData.valueArray()
+ val children = (0 until keys.numElements()).flatMap { i =>
+ val k = if (keys.isNullAt(i)) Literal(null, kt) else
Literal(keys.get(i, kt), kt)
+ val v =
+ if (vals.isNullAt(i)) Literal(null, vt)
+ else asNullable(Literal(vals.get(i, vt), vt))
+ Seq(k, v)
+ }
+ CreateMap(children, useStringTypeWhenEmpty = false)
+ case StructType(fields) =>
+ val row = value.asInstanceOf[InternalRow]
+ val children = fields.zipWithIndex.flatMap { case (f, i) =>
+ val v =
+ if (row.isNullAt(i)) Literal(null, f.dataType)
+ else asNullable(Literal(row.get(i, f.dataType), f.dataType))
Review Comment:
[P2] Preserve struct-field nullability in the native representation
`KnownNullable` is discarded by `CometKnownNullable.convert`, and native
`CreateNamedStruct::fields` infers each field's nullability from its serialized
child. With normal folding, `array(named_struct('a', CAST(NULL AS INT)),
named_struct('a', 1))` is a single literal with a unified nullable struct type,
so the outer `CometCreateArray` type guard passes. Recursive expansion then
produces native struct fields with `nullable=true` and `nullable=false`.
DataFusion 54.1 preserves those per-input flags during struct coercion, and
Arrow panics in `MutableArrayData` with `Arrays with inconsistent types`. This
reproduces on Spark 3.5.9 and 4.1.3 and the base serializer safely falls back.
Please carry/normalize the field nullability on the native wire, or decline
expansion when it cannot be preserved; the Catalyst-only wrapper does not
prevent this panic.
--
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]
