Weixin-Xu commented on code in PR #13492:
URL: https://github.com/apache/gluten/pull/13492#discussion_r4421414424
##########
backends-bolt/src/main/scala/org/apache/gluten/execution/ColumnarRangeExec.scala:
##########
@@ -98,17 +100,17 @@ case class ColumnarRangeExec(range: LogicalRange) extends
ColumnarRangeBaseExec
batchSize
)
- val vectors =
ArrowWritableColumnVector.allocateColumns(numRows, schema)
-
+ val root = ArrowColumnVectors.allocateRoot(schema)
Review Comment:
Both multiplications now happen in BigInt: BigInt(i) * step when generating
values and BigInt(numRows) * step when advancing the range. Added regression
coverage for extreme positive and negative steps.
##########
backends-bolt/src-uniffle/main/java/org/apache/spark/shuffle/writer/BoltUniffleColumnarShuffleWriter.java:
##########
@@ -224,29 +222,36 @@ public long spill(MemoryTarget self, Spiller.Phase phase,
long size) {
return;
}
long startTime = System.nanoTime();
- GlutenSplitResult splitResult;
+ BoltSplitResult splitResult;
try {
splitResult = shuffleWriterJniWrapper.stop(nativeShuffleWriter);
} catch (IOException e) {
throw new RssException(e);
}
columnarDep.metrics().get("shuffleWallTime").get().add(System.nanoTime() -
startTime);
+ // AQE needs the uncompressed size, including batches flushed by stop().
+ columnarDep
+ .metrics()
+ .get("dataSize")
+ .get()
+ .add(Arrays.stream(splitResult.getRawPartitionLengths()).sum());
+ // Bolt's evict time includes compression; the common writer reported push
time separately.
+ long pushTime =
+ Math.max(0L, splitResult.getTotalPushTime() -
splitResult.getTotalCompressTime());
columnarDep
.metrics()
.get("splitTime")
Review Comment:
The joined result now uses independently owned input vectors, so its
reference counts no longer depend on when the upstream writer releases its
batch. Added regression coverage that retains the upstream input while joining
and offloading the result, then checks that releasing either batch does not
invalidate the other.
--
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]
