liuneng1994 commented on code in PR #13492:
URL: https://github.com/apache/gluten/pull/13492#discussion_r4409629555
##########
backends-bolt/src-delta/main/scala/org/apache/gluten/component/BoltDeltaComponent.scala:
##########
@@ -37,7 +37,10 @@ class BoltDeltaComponent extends Component {
val legacy = injector.gluten.legacy
legacy.injectTransform {
c =>
- val offload = Seq(OffloadDeltaScan(), OffloadDeltaProject(),
OffloadDeltaFilter())
+ val offload = Seq(
+ OffloadDeltaScan(enableNativeDmlRowIndexScan = true),
Review Comment:
[P1] Keep deletion-vector scans on Spark until Bolt supports their native
splits
On Spark 3.5+, a Delta scan selecting files with deletion vectors produces
`DeltaLocalFilesNode` splits. `BoltPlanConverter::parseLocalFiles` has neither
a `kDelta` case nor deletion-vector handling, so these splits become `UNKNOWN`
rather than Parquet with DV filtering. The shared
`DeltaPostTransformRules.nativeDeletionVectorRule` also removes Spark's
synthetic DV filter once the scan is offloaded.
Could we keep these scans on Spark until Bolt implements both the split
representation and DV filtering? Passing `false` here only guards DV-writing
DML row-index scans; ordinary reads of DV-bearing files need a backend
capability guard too.
##########
backends-bolt/src/main/scala/org/apache/spark/api/python/ColumnarArrowEvalPythonExec.scala:
##########
@@ -368,44 +373,29 @@ case class ColumnarArrowEvalPythonExec(
val joinedVectors = (0 until inputCb.numCols).toArray.map(
i => inputCb.column(i)) ++ (0 until
outputCb.numCols).toArray.map(
i => outputCb.column(i))
- (0 until joinedVectors.length).foreach(
- i => {
-
adjustRefCnt(joinedVectors(i).asInstanceOf[ArrowWritableColumnVector], 1)
- })
val numRows = inputCb.numRows
numOutputBatches += 1
numOutputRows += numRows
val batch = new ColumnarBatch(joinedVectors, numRows)
Review Comment:
[P1] Potential mixed reference counts in the joined Arrow batch
`inputBatchIter` retains the input columns for the cache, while the upstream
iterator can still own its reference. If a Python result arrives before the
writer advances far enough to release that upstream batch, the input columns
can have refcount 2 while the newly decoded output columns have refcount 1.
Joining those columns directly creates a batch that `ColumnarBatches.offload`
cannot accept: `getRefCntHeavy` requires identical reference counts and throws
`IllegalStateException` otherwise.
Could we give the joined result independent, consistently owned vectors
instead of relying on writer timing? A test with a delayed writer and a fast
Python result would help cover this interleaving. This is a static,
timing-dependent concern, not a reproduced failure.
##########
backends-bolt/src/main/scala/org/apache/gluten/execution/ColumnarPartialGenerateExec.scala:
##########
@@ -335,24 +256,11 @@ case class ColumnarPartialGenerateExec(generateExec:
GenerateExec, child: SparkP
return Iterator.empty
}
- val colSizes = Array.fill(generateExec.generatorOutput.length)(0L)
- rowResults.foreach {
- row =>
- for (i <- variableWidthFields) {
- colSizes(i) = colSizes(i) + fieldsSizeGetter(i)(row, i)
- }
- }
-
- val rightResultVectors: Array[ArrowWritableColumnVector] =
- ArrowWritableColumnVector.allocateColumns(rowResults.length, colSizes,
rightSchema)
- val rightTargetRow = new ArrowColumnarRow(rightResultVectors)
-
- rowResults.foreach(row => writeRowUnsafe(row, rightTargetRow))
- rightTargetRow.finishWriteRow()
+ val rightTargetBatch = ArrowColumnVectors.fromRows(rightSchema,
rowResults.iterator)
Review Comment:
[P1] Terminate the generator when the partition ends, including trailing
empty batches
There is an existing termination edge case in the function whose output
conversion is changed here: `generator.terminate()` runs only when the
`hasNext` argument is false, while `doExecuteColumnar` skips empty input
batches without calling this function. With a nonempty batch followed by a
final empty batch, the nonempty batch sees `hasNext = true`, and the empty one
is skipped, so termination never happens. A Hive UDTF that buffers rows until
`close()` loses its final output and close-time cleanup.
Could termination be driven by partition exhaustion independently of batch
contents? A nonempty batch followed by an empty batch should exercise this case.
##########
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)
+ val vector = root.getVector(0).asInstanceOf[BigIntVector]
+ vector.allocateNew(numRows)
for (i <- 0 until numRows) {
val value = current + i * step
Review Comment:
[P1] Perform the range multiplications in BigInt before adding them
Although `current` is a `BigInt`, `i * step` is evaluated as `Long` before
the addition; the same applies to `numRows * step` when advancing `current`.
For `range(Long.MinValue, Long.MaxValue, Long.MaxValue, 1)`, the expected
values are `Long.MinValue`, `-1`, and `Long.MaxValue - 1`. Multiplication by 2
already overflows here, so the third value is wrong and advancing the batch can
move `current` backwards instead of exhausting the range.
Please use `BigInt(i) * step` and `BigInt(numRows) * step`. This is existing
arithmetic in the batch-generation function being migrated, and should be
covered with an extreme-step regression test.
##########
backends-bolt/src/main/scala/org/apache/spark/api/python/ColumnarArrowEvalPythonExec.scala:
##########
@@ -368,44 +373,29 @@ case class ColumnarArrowEvalPythonExec(
val joinedVectors = (0 until inputCb.numCols).toArray.map(
i => inputCb.column(i)) ++ (0 until
outputCb.numCols).toArray.map(
i => outputCb.column(i))
- (0 until joinedVectors.length).foreach(
- i => {
-
adjustRefCnt(joinedVectors(i).asInstanceOf[ArrowWritableColumnVector], 1)
- })
val numRows = inputCb.numRows
Review Comment:
[P1] Match Python output to input by rows, not by batch index
The surrounding `zipWithIndex` / `inputCbCache(batchId)` logic assumes one
Python output batch per input batch, but a scalar iterator pandas UDF only
needs to preserve the total number of rows. For example, it can concatenate two
2-row input batches and yield one 4-row output batch. Here that output is
joined with only the first input batch and the result is assigned 2 rows,
dropping the remaining output. Splitting an input batch into smaller outputs
can instead overrun an output vector or exhaust the input cache.
Could we align input/output slices by row count, or explicitly fall back for
iterator UDFs until that is supported? This assumption predates the changed
ownership code, but remains in the restored execution path.
##########
backends-bolt/src/test/scala/org/apache/gluten/execution/BoltTPCHSuite.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.execution
+
+import org.apache.gluten.config.{BoltConfig, GlutenConfig}
+
+import org.apache.spark.SparkConf
+
+/** TPC-H fixture shared by Bolt's table-format suites, matching Velox's
default subtype. */
+abstract class BoltTPCHSuite extends BoltWholeStageTransformerSuite {
+ protected val rootPath: String = getClass.getResource("/").getPath
+ override protected val resourcePath: String = "/tpch-data-parquet"
+ override protected val fileFormat: String = "parquet"
+ protected val tpchQueries: String =
+ rootPath +
"../../../../tools/gluten-it/common/src/main/resources/tpch-queries"
+ protected val queriesResults: String = rootPath + "queries-output"
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set("spark.shuffle.manager",
"org.apache.spark.shuffle.sort.ColumnarShuffleManager")
+ .set("spark.sql.files.maxPartitionBytes", "1g")
+ .set("spark.sql.shuffle.partitions", "1")
+ .set(BoltConfig.COLUMNAR_BOLT_MEM_INIT_CAPACITY.key, "1m")
+ .set("spark.memory.offHeap.size", "2g")
+ .set("spark.unsafe.exceptionOnMemoryLeak", "true")
+ .set("spark.sql.autoBroadcastJoinThreshold", "-1")
+ .set(GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "true")
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ createTPCHNotNullTables()
Review Comment:
[P2] The new setup call exposes a double resource-path resolution in
BoltTPCHDeltaSuite
`BoltTPCHDeltaSuite` sets `resourcePath` to
`getClass.getResource("/tpch-data-parquet").getFile`, which is an absolute
filesystem path. Its override of `createTPCHNotNullTables()` then passes that
path back to `getClass.getResource(resourcePath).getFile`. `getResource`
interprets it as a classpath resource name, so the second lookup returns null
in a normal test checkout and this new `beforeAll` call fails with
`NullPointerException` before any inherited TPC-H query runs.
Please keep `resourcePath` as the classpath name, or use the already
resolved filesystem path directly in the Delta fixture.
##########
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:
[P1] Do not assume every shuffle dependency exposes splitTime
A dependency with at least the configured sort-partition threshold (4000 by
default) can be classified as `SortShuffleWriterType`.
`BoltMetricsApi.genShuffleExchangeMetrics` creates `sortTime` and `c2rTime` for
that type, but not `splitTime`. This unconditional `.get("splitTime").get()`
therefore throws after the native writer stops, before `sendRestBlockAndWait()`
and `sendCommit()` run.
Could we make metric selection consistent with the dependency's writer type,
or ensure this Bolt RSS path constructs a dependency with the matching metrics?
Please include a Uniffle test above the sort threshold.
##########
cpp/bolt/jni/BoltJniWrapper.cc:
##########
@@ -946,15 +946,14 @@ JNIEXPORT jlong JNICALL
Java_org_apache_gluten_shuffle_BoltShuffleWriterJniWrapp
std::shared_ptr<RssClient> rssClient = nullptr;
if (partitionPusher != nullptr) {
- jclass celebornPartitionPusherClass =
- createGlobalClassReferenceOrError(env,
"Lorg/apache/spark/shuffle/CelebornPartitionPusher;");
- jmethodID celebornPushPartitionDataMethod =
- getMethodIdOrError(env, celebornPartitionPusherClass,
"pushPartitionData", "(I[BI)I");
+ jclass partitionPusherClass = env->GetObjectClass(partitionPusher);
Review Comment:
[P2] Release the protobuf byte-array elements in the shared writer-creation
path
Just above this changed pusher lookup, `ParseFromArray` receives a pointer
returned by `GetByteArrayElements(shuffleWriterInfoProto, nullptr)`, but the
pointer is never passed to `ReleaseByteArrayElements`. Each writer creation can
therefore leave a JNI copy allocated or a Java array pinned, including
exceptional exits. Routing Uniffle through this function expands the use of the
existing leak.
Could we use `gluten::getByteArrayElementsSafe`, as the other
protobuf-reading methods in this file do, so the elements are released on every
exit?
##########
cpp/bolt/jni/BoltJniWrapper.cc:
##########
@@ -1006,8 +1005,10 @@ JNIEXPORT void JNICALL
Java_org_apache_gluten_shuffle_BoltShuffleWriterJniWrappe
// The column batch maybe BoltColumnBatch or
ArrowCStructColumnarBatch(FallbackRangeShuffleWriter)
auto batch = ObjectStore::retrieve<ColumnarBatch>(batchHandle);
+ const auto inputBytes = batch->numBytes();
Review Comment:
[P2] Avoid flattening every shuffle batch solely for unused byte accounting
`BoltColumnarBatch::numBytes()` calls `ensureFlattened()`, which
materializes dictionary/constant children and loads lazy vectors. The local and
Celeborn writers discard `write()`'s return value, and Uniffle now uses it only
in a debug log; its `dataSize` metric comes from the final raw partition
lengths instead. Nevertheless, this unconditional call forces all those shuffle
paths to allocate flattened vectors before splitting, destroying compact
encodings for accounting the callers do not use.
Could we remove this calculation, or make it opt-in and use a non-mutating
size estimate where genuinely needed?
##########
backends-bolt/src/test/scala/org/apache/gluten/execution/BoltTPCHSuite.scala:
##########
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.gluten.execution
+
+import org.apache.gluten.config.{BoltConfig, GlutenConfig}
+
+import org.apache.spark.SparkConf
+
+/** TPC-H fixture shared by Bolt's table-format suites, matching Velox's
default subtype. */
+abstract class BoltTPCHSuite extends BoltWholeStageTransformerSuite {
+ protected val rootPath: String = getClass.getResource("/").getPath
+ override protected val resourcePath: String = "/tpch-data-parquet"
+ override protected val fileFormat: String = "parquet"
+ protected val tpchQueries: String =
+ rootPath +
"../../../../tools/gluten-it/common/src/main/resources/tpch-queries"
+ protected val queriesResults: String = rootPath + "queries-output"
+
+ override protected def sparkConf: SparkConf = {
+ super.sparkConf
+ .set("spark.shuffle.manager",
"org.apache.spark.shuffle.sort.ColumnarShuffleManager")
+ .set("spark.sql.files.maxPartitionBytes", "1g")
+ .set("spark.sql.shuffle.partitions", "1")
+ .set(BoltConfig.COLUMNAR_BOLT_MEM_INIT_CAPACITY.key, "1m")
+ .set("spark.memory.offHeap.size", "2g")
+ .set("spark.unsafe.exceptionOnMemoryLeak", "true")
+ .set("spark.sql.autoBroadcastJoinThreshold", "-1")
+ .set(GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "true")
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ createTPCHNotNullTables()
+ }
+
+ (1 to 22).foreach {
+ query =>
+ test(s"TPC-H q$query") {
+ // Like Velox's default subtype, execute every query with
noFallBack=true
+ // without a backend-specific approved plan or a golden result file.
+ runTPCHQuery(query, tpchQueries, queriesResults, compareResult =
false, noFallBack = true) {
Review Comment:
[P2] Explicitly enforce the no-fallback assertion in the new TPC-H tests
This suite inherits `BoltWholeStageTransformerSuite`, whose constructor
calls `disableFallbackCheck`. `GlutenQueryComparisonTest.checkDataFrame`
therefore skips `checkFallBack` even when `runTPCHQuery` receives `noFallBack =
true`. The custom callback is empty as well, so all 22 queries can fall back
entirely to Spark and these tests still pass without demonstrating Bolt
execution compatibility.
Could this suite override `needCheckFallback` to return true, or explicitly
invoke `GlutenQueryComparisonTest.checkFallBack` in the callback?
##########
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:
[P2] Bound the remaining row count before converting it to Int
The `numRows` calculation immediately above this allocation calls `toInt` on
the remaining `BigInt` count before applying the 1000-row batch limit. For a
single-partition `range(0L, 4294967296L, 1L, 1)`, the first count wraps to 0,
and subsequent counts wrap negative; `.max(1)` turns these into one-row Arrow
batches. This makes a valid large range spend billions of iterations allocating
single-row batches.
Please apply the bound in `BigInt` first, e.g.
`remaining.min(BigInt(batchSize)).max(BigInt(1)).toInt`. This is existing
batch-sizing logic in the function being migrated.
##########
backends-bolt/src/main/scala/org/apache/spark/api/python/ColumnarArrowEvalPythonExec.scala:
##########
@@ -103,9 +101,20 @@ class ColumnarArrowPythonRunner(
if (reader != null && batchLoaded) {
batchLoaded = reader.loadNextBatch()
if (batchLoaded) {
- val batch = new ColumnarBatch(vectors)
- batch.setNumRows(root.getRowCount)
- batch
+ // The reader reuses root across batches. Give each result its
own vectors and
+ // retained buffer references so prefetch and reader.close
cannot invalidate it.
+ val recordBatch = new VectorUnloader(root).getRecordBatch
+ val outputRoot = VectorSchemaRoot.create(root.getSchema,
allocator)
+ try {
+ new VectorLoader(outputRoot).load(recordBatch)
+ ArrowColumnVectors.toBatch(outputRoot)
Review Comment:
[P2] Potential leak of independently allocated, prefetched Python output
Spark's `ReaderIterator.hasNext` can invoke `read()` and cache a result
before `next()` transfers it downstream. If a limit, cancellation, or
downstream exception ends iteration at that point, this new `outputRoot` never
reaches `recyclePayload(_.close())`. The task-completion listener closes only
`reader` and its original `root`, not the independently allocated output
vectors.
Could the reader track producer-owned output batches and close any
unconsumed batch during teardown, removing it from that tracking when ownership
is transferred? This is a static lifecycle concern, not a reproduced leak; an
early-stop test after prefetch would validate it.
--
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]
