RRXXZZYY commented on code in PR #5603:
URL: https://github.com/apache/datafusion-comet/pull/5603#discussion_r3919352922
##########
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:
Fixed in ba8ca2cc. Generated runtime names now skip every original sibling
name before allocation, and the broadcast regression uses duplicate
__comet_runtime_field_0 fields. The focused Spark 4.1 test passed 1/1 with the
native library loaded.
--
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]
