andygrove commented on code in PR #5603:
URL: https://github.com/apache/datafusion-comet/pull/5603#discussion_r3899504780
##########
spark/src/main/scala/org/apache/comet/vector/NativeUtil.scala:
##########
@@ -322,6 +330,80 @@ class NativeUtil extends AutoCloseable {
}
object NativeUtil {
+
+ /**
+ * Create a vector whose physical struct children remain positional when the
exported Arrow
+ * schema contains duplicate names. Arrow's default struct factory indexes
children by name and
+ * collapses such fields.
+ */
+ private[comet] def createVector(field: Field, allocator: BufferAllocator):
FieldVector = {
+ val runtimeField = fieldForAllocation(field)
+ field.getType match {
+ case _: ArrowType.List | _: ArrowType.LargeList | _:
ArrowType.FixedSizeList =>
+ val vector = new RenamedListVector(runtimeField, field, allocator)
+ vector.initializeChildrenFromFields(runtimeField.getChildren)
+ vector
+ case _: ArrowType.Map =>
+ val vector = new RenamedMapVector(runtimeField, field, allocator)
+ vector.initializeChildrenFromFields(runtimeField.getChildren)
+ vector
+ case _: ArrowType.Struct =>
+ val vector = new RenamedStructVector(runtimeField, field, allocator)
+ vector.initializeChildrenFromFields(runtimeField.getChildren)
+ vector
+ case _ => field.createVector(allocator).asInstanceOf[FieldVector]
+ }
+ }
+
+ private def fieldForAllocation(field: Field): Field = {
+ val children =
field.getChildren.asScala.map(fieldForAllocation).toIndexedSeq
+ val runtimeChildren = field.getType match {
+ case _: ArrowType.Struct if children.map(_.getName).distinct.size !=
children.size =>
+ children.zipWithIndex.map { case (child, ordinal) =>
+ new Field(s"__comet_runtime_field_$ordinal", child.getFieldType,
child.getChildren)
+ }
+ case _ => children
+ }
+ new Field(field.getName, field.getFieldType, runtimeChildren.asJava)
+ }
+
+ /**
+ * Pin `getField()` to the imported Field so FFI keeps the original child
labels. ListVector's
+ * runtime data-vector label is `"$data$"`; struct runtime names may be
private and unique.
+ */
+ private final class RenamedListVector(
+ runtimeField: Field,
+ exportField: Field,
+ allocator: BufferAllocator)
+ extends ListVector(runtimeField, allocator, null) {
+ override def getField: Field = exportField
+ }
+
+ private final class RenamedMapVector(
+ runtimeField: Field,
+ exportField: Field,
+ allocator: BufferAllocator)
+ extends MapVector(runtimeField, allocator, null) {
+ override def getField: Field = exportField
+ }
+
+ private final class RenamedStructVector(
+ runtimeField: Field,
+ exportField: Field,
+ allocator: BufferAllocator)
+ extends StructVector(
+ runtimeField.getName,
+ allocator,
+ runtimeField.getFieldType,
+ null,
+ AbstractStructVector.ConflictPolicy.CONFLICT_ERROR,
+ true) {
+ private var constructed = false
Review Comment:
What is this flag guarding? The only thing it can be is a `getField()` call
from inside the superclass constructor, when `exportField` is still null. But I
walked Arrow 18.3's `StructVector` / `NonNullableStructVector` /
`AbstractStructVector` constructor chain and none of them call `getField()`,
and `RenamedListVector` and `RenamedMapVector` just above have no equivalent
guard.
If a superclass constructor really does reach `getField()` here, a comment
saying so would stop someone deleting this later. If it does not, can it go?
Separately, Arrow 18.3 has `StructVector(Field, allocator, callBack,
conflictPolicy, allowConflictPolicyChanges)`, so you can pass `runtimeField`
directly and still get `CONFLICT_ERROR` without dropping to the `name` /
`fieldType` constructor.
##########
spark/src/test/scala/org/apache/comet/CometExpressionSuite.scala:
##########
@@ -2480,15 +2480,10 @@ class CometExpressionSuite extends CometTestBase with
AdaptiveSparkPlanHelper {
val path = new Path(dir.toURI.toString, "test.parquet")
makeParquetFileAllPrimitiveTypes(path, dictionaryEnabled =
dictionaryEnabled, 10000)
withParquetTable(path.toString, "tbl") {
+ checkSparkAnswerAndOperator("SELECT named_struct('a', _1, 'a', _2)
FROM tbl")
Review Comment:
`spark/src/test/resources/sql-tests/expressions/struct/create_named_struct.sql`
already exists, and the SQL file framework runs each query through both engines
and asserts native execution with no Scala. Would it be worth moving the
duplicate-name cases there? That gets them running under 3.4 and 3.5 as well,
which matters given this was only validated against 4.0 locally.
A few cases that would be good to cover wherever they land:
- `SELECT struct(a, a) FROM tbl`. `CreateStruct` lowers to
`CreateNamedStruct` and names each field after its child expression, so this
goes through the same serde and is currently untested.
- A duplicate-name struct inside an array or a map, for example `SELECT
array(named_struct('a', _1, 'a', _2)) FROM tbl`. That exercises the branch of
`fieldForAllocation` where a `ListVector` or `MapVector` root has to carry a
renamed subtree while the pinned root field keeps the original names. None of
the current tests reach it, and it is the case most likely to break.
- Three or more duplicates, and all-null rows.
- A case that crosses a shuffle (`ORDER BY` or `repartition`) rather than
going straight to the driver.
##########
spark/src/main/scala/org/apache/comet/vector/NativeUtil.scala:
##########
@@ -322,6 +330,80 @@ class NativeUtil extends AutoCloseable {
}
object NativeUtil {
+
+ /**
+ * Create a vector whose physical struct children remain positional when the
exported Arrow
+ * schema contains duplicate names. Arrow's default struct factory indexes
children by name and
+ * collapses such fields.
+ */
+ private[comet] def createVector(field: Field, allocator: BufferAllocator):
FieldVector = {
+ val runtimeField = fieldForAllocation(field)
+ field.getType match {
+ case _: ArrowType.List | _: ArrowType.LargeList | _:
ArrowType.FixedSizeList =>
+ val vector = new RenamedListVector(runtimeField, field, allocator)
+ vector.initializeChildrenFromFields(runtimeField.getChildren)
+ vector
+ case _: ArrowType.Map =>
+ val vector = new RenamedMapVector(runtimeField, field, allocator)
+ vector.initializeChildrenFromFields(runtimeField.getChildren)
+ vector
+ case _: ArrowType.Struct =>
+ val vector = new RenamedStructVector(runtimeField, field, allocator)
+ vector.initializeChildrenFromFields(runtimeField.getChildren)
+ vector
+ case _ => field.createVector(allocator).asInstanceOf[FieldVector]
Review Comment:
Routing all imports through `createVector` also pins `getField()` on every
list, map and struct column, not just the duplicate-name ones. Before this PR
the import path used `field.createVector(allocator)` for all of them.
That is a visible change for columns that have nothing to do with this fix.
Arrow's `ListVector.getField()` rebuilds children from the data vector, which
`addOrGetVector` names `$data$`, so imported list columns used to report a
child named `$data$` and now report whatever name native sent. That may well be
the right thing, but it is a separate change, it is untested, and the PR
description does not mention it.
Could `createVector` short-circuit to `field.createVector(allocator)` when
the field tree contains no duplicate struct names? That fixes the per-batch
cost above and keeps the blast radius on the case this PR is actually about.
##########
spark/src/test/scala/org/apache/comet/vector/NativeUtilSuite.scala:
##########
@@ -365,4 +366,47 @@ class NativeUtilSuite extends CometTestBase {
nativeUtil.close()
}
}
+
+ test("importVector preserves duplicate struct fields positionally") {
+ val allocator = new RootAllocator(Long.MaxValue)
+ val nativeUtil = new NativeUtil
+ val children = Arrays.asList(
+ new Field("a", FieldType.nullable(ArrowType.Bool.INSTANCE),
Collections.emptyList[Field]()),
+ new Field(
+ "a",
+ FieldType.nullable(new ArrowType.Int(8, true)),
+ Collections.emptyList[Field]()))
+ val field = new Field("value",
FieldType.nullable(ArrowType.Struct.INSTANCE), children)
+ val source = NativeUtil.createVector(field,
allocator).asInstanceOf[StructVector]
+ var imported: CometVector = null
+
+ try {
+ source.allocateNew()
+ val bool = source.getChildByOrdinal(0).asInstanceOf[BitVector]
+ val byte = source.getChildByOrdinal(1).asInstanceOf[TinyIntVector]
+ bool.setSafe(0, 1)
+ byte.setSafe(0, 7)
+ bool.setValueCount(1)
+ byte.setValueCount(1)
+ source.setIndexDefined(0)
+ source.setValueCount(1)
+
+ val array = ArrowArray.allocateNew(allocator)
+ val schema = ArrowSchema.allocateNew(allocator)
+ Data.exportVector(allocator, source, null, array, schema)
+ source.close()
Review Comment:
`source.close()` runs here and again at line 406.
Also, the `ArrowArray` and `ArrowSchema` allocated at 394 and 395 are not
closed if anything between here and the import throws, so `allocator.close()`
in the `finally` will raise a leak error that hides the real failure. Wrapping
them in `Using` or closing them in the `finally` would make a failure in this
test readable.
##########
spark/src/main/scala/org/apache/comet/serde/structs.scala:
##########
@@ -31,7 +31,9 @@ import
org.apache.comet.CometSparkSessionExtensions.withFallbackReason
import org.apache.comet.DataTypeSupport
import org.apache.comet.serde.QueryPlanSerde.{exprToProtoInternal,
serializeDataType}
-object CometCreateNamedStruct extends CometExpressionSerde[CreateNamedStruct] {
+object CometCreateNamedStruct
+ extends CometExpressionSerde[CreateNamedStruct]
+ with CodegenDispatchFallback {
private val duplicateNamesReason =
Review Comment:
Now that `CodegenDispatchFallback` is mixed in, `GenerateDocs` renders
`getUnsupportedReasons()` under the header "The following cases have no native
implementation and always run in the JVM using Spark's code-generated
implementation (inside the Comet pipeline)" (`GenerateDocs.scala:400`). So the
generated `struct.md` will say duplicate names run in the JVM and then bullet
that they are "not supported", which reads as a contradiction.
Could the doc-facing string be split from the `Unsupported(...)` note, the
way `ComparisonUtils.nonDefaultCollationDocReason` does it in
`predicates.scala`? Something along the lines of "Duplicate field names are
routed through the JVM codegen dispatcher (Spark's own `doGenCode`)".
##########
spark/src/main/scala/org/apache/comet/vector/NativeUtil.scala:
##########
@@ -322,6 +330,80 @@ class NativeUtil extends AutoCloseable {
}
object NativeUtil {
+
+ /**
+ * Create a vector whose physical struct children remain positional when the
exported Arrow
+ * schema contains duplicate names. Arrow's default struct factory indexes
children by name and
+ * collapses such fields.
+ */
+ private[comet] def createVector(field: Field, allocator: BufferAllocator):
FieldVector = {
+ val runtimeField = fieldForAllocation(field)
Review Comment:
`importVector` runs this for every column of every batch coming back from
native, so `createVector` is now on the hottest JVM path in Comet.
`fieldForAllocation` is computed here before the match, which means a plain
`IntVector` column pays a full recursive `Field` tree rebuild (with the
`asScala` / `toIndexedSeq` / `asJava` conversions and a fresh `Field` per node)
and then discards it in the `case _` branch below. `fieldForAllocation` also
allocates a new `Field` at every node even when nothing needed renaming.
Could `fieldForAllocation` return the original `field` by identity when no
struct in the tree has duplicate names, so this becomes a cheap walk instead of
a rebuild? Related, line 279 builds a fresh closure per column per batch, which
could be hoisted to a `val` on `NativeUtil`.
##########
docs/source/user-guide/latest/expressions.md:
##########
@@ -624,7 +624,7 @@ The type-name conversion functions (`bigint`, `binary`,
`boolean`, `date`, `deci
| Function | Status | Implementation | Notes |
| --- | --- | --- | --- |
-| `named_struct` | ✅ | Native | Duplicate field names fall back |
+| `named_struct` | ✅ | Hybrid | Duplicate field names route through the JVM
codegen dispatcher |
| `struct` | ✅ | Native | |
Review Comment:
The `struct` row should move to `Hybrid` too.
`GenerateDocs.buildFunctionNameToKind` resolves both `named_struct` and
`struct` through Spark's `FunctionRegistry` to `CometCreateNamedStruct`, and
`CreateStruct` lowers to `CreateNamedStruct` naming each field after its child
expression, so `struct(a, a)` produces duplicate names and takes exactly this
path.
--
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]
