comphead commented on code in PR #7124:
URL: https://github.com/apache/datafusion-comet/pull/7124#discussion_r4332014828
##########
spark/src/main/scala/org/apache/comet/vector/NativeUtil.scala:
##########
@@ -142,12 +142,22 @@ class NativeUtil extends AutoCloseable {
// Don't need to deallocate them here.
val arrowSchema = ArrowSchema.wrap(schemaAddrs(index))
val arrowArray = ArrowArray.wrap(arrayAddrs(index))
- Data.exportVector(
- allocator,
- getFieldVector(valueVector, "export"),
- provider,
- arrowArray,
- arrowSchema)
+ narrowLargeOffsets(valueVector) match {
+ case Some(narrow) =>
+ // A PyArrow UDF can return large_string or large_binary, which
the native side
+ // does not plan for. Export a 32-bit copy instead, as
serialization does. The
+ // export retains the copy's buffers, so the copy itself is
closed here and its
+ // memory is freed when native releases the array.
+ try Data.exportVector(allocator, narrow, provider, arrowArray,
arrowSchema)
+ finally narrow.close()
Review Comment:
A question on the design. Native columnar-to-row already casts every column
to the planned type in `maybe_cast_to_schema_type` (`columnar_to_row.rs`), and
the schema it gets comes from `serializeDataType(StringType)` and `BinaryType`,
so a `LargeUtf8` or `LargeBinary` column should reach the default arm and
Arrow's cast. For these types that cast should only rebuild the offsets and
reuse the data buffer, and it errors if an offset exceeds `i32`.
`narrowOffsets` copies the payload, the validity bitmap and the offsets, with a
per-element loop, and holds those buffers until native releases the array. Is
the JVM-side copy needed here, or could `exportBatch` pass the large vectors
through and leave the cast to native? If there is a reason it has to be
narrowed in the JVM, such as a path that skips `maybe_cast_to_schema_type`, a
line in the comment would help. I have not run this, so the native cast arm is
read from source only.
##########
spark/src/main/scala/org/apache/spark/sql/comet/CometMapInBatchExec.scala:
##########
@@ -106,8 +106,9 @@ case class CometMapInBatchExec(
columnarBatchIter.map { batch =>
// Python returns a single struct column; flatten to the user's output
columns. The runner
// produces Comet vectors, so the struct's children are already
CometVectors that downstream
- // consumers (a stacked CometMapInBatchExec, or NativeUtil.exportBatch
for a native Comet
- // operator) can use directly.
+ // consumers (a stacked CometMapInBatchExec, the Arrow C Stream in
CometNativeArrowSource for
Review Comment:
This comment line is 101 characters and `FileLineLengthChecker` in
`dev/scalastyle-config.xml` allows 100, so scalastyle should reject it. I
measured the length and did not run scalastyle. Could you rewrap it?
--
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]
