mbutrovich commented on code in PR #5046:
URL: https://github.com/apache/datafusion-comet/pull/5046#discussion_r3806406329
##########
spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/RowArrowReader.scala:
##########
@@ -55,7 +55,7 @@ private[comet] class RowArrowReader(
}
val startNs = System.nanoTime()
- val writer = ArrowWriter.create(getVectorSchemaRoot)
+ val writer = ArrowWriter.create(getVectorSchemaRoot,
maxRecordsPerBatch.toInt)
var rowCount = 0L
while (rowIter.hasNext &&
(maxRecordsPerBatch <= 0 || rowCount < maxRecordsPerBatch)) {
Review Comment:
The `maxRecordsPerBatch <= 0` branch here looks like it's no longer safe now
that `write(row)` routes fixed-width fields through `writeUnsafe`.
`create(root, maxRecordsPerBatch.toInt)` on line 58 pre-allocates each
fixed-width vector to exactly `maxRecordsPerBatch` capacity, and
`FixedWidthArrowFieldWriter.writeUnsafe` calls Arrow's non-resizing `set()`
with no bounds check. If `maxRecordsPerBatch` is ever `0`, `ArrowWriter.create`
allocates fixed-width vectors at capacity 0 (its `require(fixedWidthCapacity >=
0, ...)` on `ArrowWriters.scala:43` permits 0), but this loop treats `0` as
"unlimited" and keeps writing through the unsafe setter anyway. Under
`arrow.enable_unsafe_memory_access=true` (the default in non-debug builds),
that's an out-of-bounds native write, not an exception.
This isn't reachable today because `CometConf.COMET_BATCH_SIZE` has
`checkValue(v => v > 0, ...)`, so both production callers
(`CometSparkToColumnarExec`, `CometLocalTableScanExec`) always pass a positive
value. But `CometArrowConverters.rowToArrowBatchIter` hit this same class of
problem and picked up `require(maxRecordsPerBatch > 0, ...)` at
`CometArrowConverters.scala:58`. `RowArrowReader` shares the same invariant (a
writer built once per batch, sized for the whole batch, then written unsafely)
and doesn't have the matching guard. `SparkColumnarArrowReader` is fine here
since `writeCol`/`writeColNoNull` self-heal capacity via their own `reAlloc`
loop regardless of what `create` allocated, but `RowArrowReader.write(row)` has
no such self-healing.
Could this get the same `require` (or have the dead `<= 0` branch removed,
since `maxRecordsPerBatch` is always positive in practice)? Right now it's
unreachable only because of a config check upstream in `CometConf`, and there's
no compiler-enforced link between the two.
--
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]
