andygrove commented on code in PR #6730:
URL: https://github.com/apache/datafusion-comet/pull/6730#discussion_r4212746104
##########
spark/src/main/scala/org/apache/spark/sql/comet/execution/arrow/ArrowWriters.scala:
##########
@@ -1384,16 +1627,59 @@ private[arrow] class TimeNanoWriter(val valueVector:
TimeNanoVector)
private[arrow] class ArrayWriter(val valueVector: ListVector, val
elementWriter: ArrowFieldWriter)
extends ArrowFieldWriter {
+ // Pointed at each unsafe array in turn, rather than allocating a view per
value.
+ private val elements = new UnsafeArrayData
+
override def setNull(): Unit = {}
override def setValue(input: SpecializedGetters, ordinal: Int): Unit = {
- val array = input.getArray(ordinal)
+ input.getArray(ordinal) match {
+ case unsafe: UnsafeArrayData =>
+ writeElements(unsafe)
+ case array =>
+ val numElements = array.numElements()
+ valueVector.startNewValue(count)
+ var i = 0
+ while (i < numElements) {
+ elementWriter.write(array, i)
+ i += 1
+ }
+ valueVector.endValue(count, numElements)
+ }
+ }
+
+ override private[arrow] def writeUnsafeRowField(row: UnsafeRow, ordinal:
Int): Unit = {
+ if (row.isNullAt(ordinal)) {
+ setNull()
+ } else {
+ writeUnsafe(row.getBaseObject, row.getBaseOffset, row.getLong(ordinal))
+ }
+ count += 1
+ }
+
+ override private[arrow] def writeArrayElements(array: UnsafeArrayData): Unit
= {
+ val numElements = array.numElements()
var i = 0
- valueVector.startNewValue(count)
- while (i < array.numElements()) {
- elementWriter.write(array, i)
+ while (i < numElements) {
+ if (array.isNullAt(i)) {
+ setNull()
+ } else {
+ writeUnsafe(array.getBaseObject, array.getBaseOffset, array.getLong(i))
+ }
+ count += 1
i += 1
}
+ }
+
+ /** Sets value `count` to the unsafe array that `offsetAndSize` locates from
`baseOffset`. */
+ private def writeUnsafe(base: AnyRef, baseOffset: Long, offsetAndSize:
Long): Unit = {
+ elements.pointTo(base, baseOffset + (offsetAndSize >> 32),
offsetAndSize.toInt)
Review Comment:
Thanks, confirmed. Each of the three views held the last row it read until
the writer was dropped. Rather than detach them, 0623dbac51 takes the reuse
out. Each array, map and struct is read through Spark's getters again, which
allocate a view per value as the generic path does, so nothing a writer holds
outlives the copy.
In the mixed harness the reuse was only worth 1-4 ns per row on nested
types, about a tenth of what this PR gains on them. array<string> goes from 59
ns per row on `main` to 27 with reuse and 30 without, and map<string,string>
from 117 to 50 and 53. Detaching after each value, as suggested, came out no
better than allocating. It was 13% slower on array<array<int>>, since an array
of nested values detaches once per element. So reuse would need a detach once
per row to pay off, and I filed #6765 for that.
The new test, `the row path keeps no reference to an unsafe row once it is
written`, walks the writers' fields after each row, along with those of any
unsafe view they hold, and checks that the row's backing array isn't reachable.
So it doesn't depend on GC, as @viirya suggested. It fails on the previous
head, and on a detaching version that leaves out any one of the three writers'
detach.
--
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]
