andygrove commented on code in PR #5051:
URL: https://github.com/apache/datafusion-comet/pull/5051#discussion_r3867237889
##########
spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/CometArrowConverters.scala:
##########
@@ -74,4 +74,45 @@ object CometArrowConverters extends Logging {
}
}
}
+
+ /**
+ * Copy a Spark `ColumnarBatch` whose columns are not Arrow-backed (e.g.
+ * `On/OffHeapColumnVector` from Spark's vectorized Parquet reader, or a
third-party connector's
+ * vectors) into a freshly allocated Arrow `ColumnarBatch` of `CometVector`s.
+ *
+ * The input batch is not consumed or closed; the caller owns the returned
batch and must close
+ * it. Values are copied element-wise, since Spark's `ColumnVector`
implementations do not
+ * expose Arrow buffers.
+ */
+ def columnarBatchToArrowBatch(
Review Comment:
Following up here because merging latest `apache/main` changed the outcome
of this thread.
The shared copy loop we settled on, `CometArrowConverters.writeColumns`, is
gone. #5046 and #5442 landed on main in the meantime and moved that
responsibility onto `ArrowWriter` itself as `ArrowWriter.writeColumns(input,
startRow, numRows)`, with a bulk-copy path for fixed-width columns rather than
the element-wise loop this PR had. So the deduplication you asked for now lives
upstream of both callers, and I resolved the conflict by deleting this PR's
version rather than keeping a second one:
`SparkColumnarArrowReader.loadNextBatch` takes main's version wholesale, and
`columnarBatchToArrowBatch` now calls `ArrowWriter.create(root, numRows)`,
`writeColumns`, `finish()`.
One detail worth recording, since it was a deliberate deletion rather than
an oversight. This PR's `writeColumns` ended with an explicit
`root.setRowCount(numRows)`, because `ArrowWriter` derived the root row count
from per-column writes and a zero-column batch (Spark's count-from-metadata
scan, `numRows > 0` with `numCols == 0`) would otherwise have produced
`rowCount == 0` and silently dropped the rows. Main's `writeColumns` sets
`count = numRows` directly, so `finish()` gets it right with no columns present
and the workaround is no longer needed. `CometInMemoryCacheSuite` has a test
for that exact case ("supports empty projection scan"), and it still passes, so
this is covered rather than taken on trust.
The leak guard from the other thread is unaffected and still wraps both call
sites. All 21 tests in the suite pass on the merged branch.
--
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]
