sunchao commented on code in PR #5603:
URL: https://github.com/apache/datafusion-comet/pull/5603#discussion_r3918896606
##########
spark/src/main/scala/org/apache/comet/vector/NativeUtil.scala:
##########
@@ -322,6 +333,190 @@ 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)
+ createPinnedVector(runtimeField, field, allocator)
+ }
+
+ /**
+ * Preserve Arrow's default allocation path unless a duplicate-name struct
needs positional
+ * runtime children. This is called for every imported column of every
native batch.
+ */
+ private[comet] def createVectorForImport(
+ field: Field,
+ allocator: BufferAllocator): FieldVector = {
+ val runtimeField = fieldForAllocation(field)
+ if (runtimeField eq field) {
+ field.createVector(allocator).asInstanceOf[FieldVector]
+ } else {
+ createPinnedVector(runtimeField, field, allocator)
+ }
+ }
+
+ /** Build an IPC root with the same duplicate-safe allocation used by C Data
imports. */
+ def createVectorSchemaRootForImport(
+ schema: Schema,
+ allocator: BufferAllocator): VectorSchemaRoot = {
+ val fields = schema.getFields
+ val vectors = new ArrayList[FieldVector](fields.size())
+ try {
+ var ordinal = 0
+ while (ordinal < fields.size()) {
+ vectors.add(createVectorForImport(fields.get(ordinal), allocator))
+ ordinal += 1
+ }
+ new VectorSchemaRoot(schema, vectors, 0)
+ } catch {
+ case failure: Throwable =>
+ AutoCloseables.close(failure, vectors)
+ throw failure
+ }
+ }
+
+ /**
+ * Build a C Stream root whose physical and advertised schemas use the same
duplicate-safe field
+ * names. Arrow's C Data exporter reconstructs nested vectors from the
advertised schema and
+ * otherwise collapses duplicate struct children before loading the record
batch.
+ */
+ def createVectorSchemaRootForExport(
+ schema: Schema,
+ allocator: BufferAllocator): VectorSchemaRoot = {
+ val fields = schema.getFields
+ val runtimeFields = new ArrayList[Field](fields.size())
+ val vectors = new ArrayList[FieldVector](fields.size())
+ try {
+ var ordinal = 0
+ while (ordinal < fields.size()) {
+ val runtimeField = fieldForAllocation(fields.get(ordinal))
+ runtimeFields.add(runtimeField)
+
vectors.add(runtimeField.createVector(allocator).asInstanceOf[FieldVector])
+ ordinal += 1
+ }
+ new VectorSchemaRoot(new Schema(runtimeFields), vectors, 0)
Review Comment:
[P2] Keep stream runtime names disjoint from logical field names
Could this preserve ordinal mapping when a user names both children
`__comet_runtime_field_0`? This root advertises `__comet_runtime_field_0` and
`__comet_runtime_field_1`. `ScanStream::build_record_batch` then casts back to
the declared struct type, but Arrow 58.4.0 matches both target names to source
child 0 and returns that child's values twice. In the existing broadcast-join
regression, replacing both `'x'` labels with `'__comet_runtime_field_0'` has
this path from `(1, 10)` to `(1, 1)`. A focused probe using the pinned Arrow
cast reproduced the value and null duplication. I did not execute the Spark
query. Please make exported runtime names disjoint from every original child
name, or restore the logical schema strictly by ordinal, and cover this case in
the broadcast regression.
--
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]
