peterxcli commented on code in PR #5539:
URL: https://github.com/apache/datafusion-comet/pull/5539#discussion_r3886125731
##########
spark/src/main/scala/org/apache/comet/vector/NativeUtil.scala:
##########
@@ -254,15 +254,29 @@ class NativeUtil {
def importVector(arrays: Array[ArrowArray], schemas: Array[ArrowSchema]):
Seq[CometVector] = {
val arrayVectors = mutable.ArrayBuffer.empty[CometVector]
- (0 until arrays.length).foreach { i =>
- val arrowSchema = schemas(i)
- val arrowArray = arrays(i)
+ try {
+ (0 until arrays.length).foreach { i =>
+ val arrowSchema = schemas(i)
+ val arrowArray = arrays(i)
- arrayVectors += CometVector.getVector(
- importer.importVector(arrowArray, arrowSchema, dictionaryProvider),
- dictionaryProvider)
+ arrayVectors += CometVector.getVector(
Review Comment:
[P1] Keep the current `FieldVector` rollback-owned until publication
`arrayVectors` only owns earlier columns here. If `importer.importVector`
fails after `field.createVector` / `importArray` has attached buffers, or if it
succeeds and `CometVector.getVector` throws, the current `FieldVector` is
unreachable. The C structs are already consumed, so `releaseArrowStructs`
cannot recover it. I reproduced the latter with a valid `UInt4Vector`:
allocator close reports 48 leaked bytes.
Please close the raw `FieldVector` on `ArrowImporter.importVector` failure
and retain rollback ownership across wrapping and `+=`; commit only after it is
appended. Add a failure-after-import regression—the current test fails during
schema import before a `FieldVector` exists.
##########
spark/src/main/scala/org/apache/comet/vector/NativeUtil.scala:
##########
@@ -254,15 +254,29 @@ class NativeUtil {
def importVector(arrays: Array[ArrowArray], schemas: Array[ArrowSchema]):
Seq[CometVector] = {
val arrayVectors = mutable.ArrayBuffer.empty[CometVector]
- (0 until arrays.length).foreach { i =>
- val arrowSchema = schemas(i)
- val arrowArray = arrays(i)
+ try {
+ (0 until arrays.length).foreach { i =>
+ val arrowSchema = schemas(i)
+ val arrowArray = arrays(i)
- arrayVectors += CometVector.getVector(
- importer.importVector(arrowArray, arrowSchema, dictionaryProvider),
- dictionaryProvider)
+ arrayVectors += CometVector.getVector(
+ importer.importVector(arrowArray, arrowSchema, dictionaryProvider),
+ dictionaryProvider)
+ }
+ arrayVectors.toSeq
+ } catch {
+ case failure: Throwable =>
+ val firstUnconsumed = arrayVectors.length
+ arrayVectors.foreach { vector =>
Review Comment:
Non-blocking: Arrow 18.3 already provides
`org.apache.arrow.util.AutoCloseables`; since `CometVector` is `AutoCloseable`,
this loop can be `AutoCloseables.close(failure, arrayVectors.toSeq: _*)`. It
closes every resource and attaches cleanup failures to the original. Caveat: it
catches `Exception` rather than every `Throwable`, so keep the manual loop if
continuing after a close-time `Error` is intentional. No new Comet helper is
needed.
--
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]
