Copilot commented on code in PR #13115:
URL: https://github.com/apache/gluten/pull/13115#discussion_r4244658145
##########
cpp/velox/compute/VeloxBackend.cc:
##########
@@ -254,6 +255,10 @@ void VeloxBackend::init(
std::make_unique<folly::CPUThreadPoolExecutor>(ioThreads,
folly::CPUThreadPoolExecutor::makeLifoSemQueue());
}
+ const auto broadcastHashTableBuildThreads =
+ std::max<int32_t>(1,
backendConf_->get<int32_t>(kBroadcastHashTableBuildThreads,
numTaskSlotsPerExecutor));
Review Comment:
This creates an additional CPU thread pool sized (by default) to
`numTaskSlotsPerExecutor`, which can materially increase total thread count and
CPU contention (task threads + this pool) on busy executors. If thatβs
intentional, it would help to (a) document the potential oversubscription and
recommended sizing in the config docs, and/or (b) consider a more conservative
computed default or an option to disable creating the pool when parallel join
build is effectively off.
##########
gluten-ut/spark35/src/test/backends-velox/org/apache/spark/sql/execution/joins/GlutenParallelBroadcastJoinSuite.scala:
##########
@@ -0,0 +1,64 @@
+/*
+ * 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.spark.sql.execution.joins
+
+import org.apache.gluten.execution.BroadcastHashJoinExecTransformer
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.GlutenSQLTestsTrait
+import org.apache.spark.sql.execution.ColumnarBroadcastExchangeExec
+
+class GlutenParallelBroadcastJoinSuite extends GlutenSQLTestsTrait {
+
+ override def sparkConf: SparkConf = super.sparkConf
+ .set("spark.sql.adaptive.enabled", "false")
+ .set("spark.gluten.sql.columnar.backend.velox.IOThreads", "0")
+ .set("spark.gluten.velox.broadcast.build.targetBytesPerThread", "8KB")
+
+ for (buildOnDriver <- Seq(false, true)) {
+ testGluten(s"parallel broadcast hash table build with IO disabled,
driver=$buildOnDriver") {
+ withSQLConf(
+
"spark.gluten.sql.columnar.backend.velox.driverSideBroadcastHashTableBuild" ->
+ buildOnDriver.toString,
+ "spark.gluten.velox.minTableRowsForParallelJoinBuild" -> "1"
+ ) {
+ withTable("parallel_broadcast_build") {
+ spark.range(4096)
+ .selectExpr("id", "sha2(cast(id as string), 256) AS value")
+ .write.saveAsTable("parallel_broadcast_build")
+
+ val result = spark.sql("""
+ |SELECT /*+ BROADCAST(b) */ p.id, b.value
+ |FROM parallel_broadcast_build p JOIN
parallel_broadcast_build b
+ |ON p.id = b.id
+ |""".stripMargin)
+ val expected =
spark.table("parallel_broadcast_build").collect().toSeq
+ checkAnswer(result, expected)
+
+ val plan = result.queryExecution.executedPlan
+ assert(plan.collect { case j: BroadcastHashJoinExecTransformer => j
}.nonEmpty)
+ val exchanges = plan.collect { case e: ColumnarBroadcastExchangeExec
=> e }
+ assert(exchanges.size == 1)
+ assert(exchanges.head.metrics("buildThreads").value > 1)
Review Comment:
This test hard-requires `buildThreads > 1`, but the thread-pool default is
derived from executor task slots. In environments with a single task slot
(e.g., local[1] or constrained CI), the metric can legitimately be 1 and the
test will fail. Consider setting
`spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads`
explicitly to a value > 1 for this suite, or make the assertion conditional on
the configured pool size (e.g., assert `>= 1` always, and assert `> 1` only
when the configured threads are > 1).
##########
docs/velox-configuration.md:
##########
@@ -16,6 +16,7 @@ nav_order: 16
| spark.gluten.sql.columnar.backend.velox.abandonPartialAggregationMinPct
| π Dynamic | 90 | If partial aggregation aggregationPct
greater than this value, partial aggregation may be early abandoned. Note: this
option only works when flushable partial aggregation is enabled. Ignored when
spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false.
|
| spark.gluten.sql.columnar.backend.velox.abandonPartialAggregationMinRows
| π Dynamic | 100000 | If partial aggregation input rows
number greater than this value, partial aggregation may be early abandoned.
Note: this option only works when flushable partial aggregation is enabled.
Ignored when
spark.gluten.sql.columnar.backend.velox.flushablePartialAggregation=false.
|
| spark.gluten.sql.columnar.backend.velox.asyncTimeoutOnTaskStopping
| β Static | 30000ms | Timeout in milliseconds when waiting
for runtime-scoped async work to finish during teardown.
|
+| spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads
| β Static | <undefined> | Size of the dedicated CPU thread pool
for the parallel broadcast hash-table build. By default, the value is the same
as the maximum task slots per Spark executor, with a minimum of one thread.
|
Review Comment:
The default column shows `<undefined>`, but the description states a
concrete default behavior (derived from task slots, min 1). To avoid confusion
for users, update the default column to reflect the real default (e.g.,
`numTaskSlotsPerExecutor (min 1)`) or otherwise clarify how the effective
default is computed.
--
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]
