andygrove commented on code in PR #5051:
URL: https://github.com/apache/datafusion-comet/pull/5051#discussion_r3677934494
##########
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:
You are right on both counts, and your analysis of the `ColumnVector`
contract matches what I found — `ColumnarArray` + the standard accessors means
it works for any correctly-implemented `ColumnVector`, which is why the fix
also covers Iceberg and other external columnar sources without naming them.
The duplication is fixed in c7bc49e5: the copy loop is now
`CometArrowConverters.writeColumns(root, batch, startRow, numRows)`, and
`SparkColumnarArrowReader.loadNextBatch` calls it too. That also collapses the
two copies of the zero-column `setRowCount` workaround back into one.
On your `readerBatchIter` sketch — I went a slightly different way and want
to flag why, since it is a real tradeoff:
- It converts **unconditionally**, so a Comet-native cached plan (the
primary case for this feature) would get a full element-wise copy through the
generic accessors instead of the current zero-copy Arrow IPC write. That is the
regression I most wanted to avoid, so `encodeBatches` keeps a per-batch
`Utils.isArrowBacked` check and only converts foreign vectors.
- The reader wraps the whole partition iterator, so the Arrow-vs-not
decision becomes per-partition rather than per-batch. In practice a partition
is homogeneous, but making it per-batch costs nothing.
Your version does have one genuine advantage mine gives up:
`SparkColumnarArrowReader` reuses one stable VSR across batches, whereas
`columnarBatchToArrowBatch` allocates a fresh root per batch. If that shows up
in cache-materialization profiles it would be worth revisiting — happy to file
a follow-up.
One thing your sketch surfaced that I did **not** change: your code uses
`conf.sessionLocalTimeZone` for the Arrow schema, which matches the
pre-existing row path (`convertInternalRowToCachedBatch`), while my columnar
path uses `CometArrowStream.NATIVE_TIMEZONE` (UTC) to match
`CometSparkToColumnarExec`. So the same `CometCachedBatch` format is currently
written with two timezone conventions depending on the input path. Timestamps
are UTC micros either way so this is metadata-only today, but it is
inconsistent and I would rather fix it deliberately. Do you have a preference
for which one the cache format should standardise on?
--
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]
