sunchao commented on code in PR #6730:
URL: https://github.com/apache/datafusion-comet/pull/6730#discussion_r4211468373
##########
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:
[P2] [P2] Release backing storage held by reusable nested views
`elements.pointTo` leaves the reusable view holding the entire input row’s
backing array after its contents have been copied into Arrow.
`StructWriter.fields` and `MapWriter.entries` do the same. When nullable nested
fields have their last non-null values in different rows, each view retains a
different complete row, including unrelated large payloads. Subsequent nulls do
not release those references.
Through the production `RowArrowReader`, 24 streamed unsafe rows with a 2
MiB binary payload and distinct populated nested fields retain 50,337,600 bytes
of consumed input storage at this head. The base retains none. Arrow already
owns the output copies, so these input arrays should be collectible after
conversion. This adds approximately 48 MiB of unnecessary JVM heap residency
during batch construction. The arrays become collectible when the batch writer
dies, so this increases peak residency rather than permanently leaking memory.
Could we detach each reusable view after its synchronous copy, including the
map’s key/value views, and add an input-lifetime regression test?
Evidence: Run `python /tmp/pr6730-61ec-review-wjvupbl7/run-retention.py`. It
separately compiles the base and exact-head `ArrowWriters.scala` with the
production `RowArrowReader`, then runs `ReaderRetentionProbe.scala` using Spark
4.1.3, JDK 17, `-Xmx160m` and Serial GC. The iterator generates 24
`UnsafeProjection(...).copy()` rows containing a 2 MiB binary payload plus
eight array, eight map and eight struct fields, with only one nested field
populated per row. The harness keeps only weak references to consumed backing
arrays. Forced GC inside the final `hasNext`, while batch construction is
active, reports base `(0,0)` surviving arrays/bytes versus head
`(24,50337600)`. After `loadNextBatch` returns, both report `(0,0)`. Outputs
are in `base-reader.log` and `head-reader.log`. Spark’s `pointTo`
implementations confirm the strong backing-object references.
--
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]
