Copilot commented on code in PR #13115:
URL: https://github.com/apache/gluten/pull/13115#discussion_r4229666885
##########
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, 1));
+ broadcastHashTableBuildExecutor_ =
std::make_unique<folly::CPUThreadPoolExecutor>(broadcastHashTableBuildThreads);
Review Comment:
The PR description says the CPU thread pool is 'lazily-created' following a
`std::call_once` pattern, but the code creates it eagerly during
`VeloxBackend::init()`. Either update the PR description to reflect eager
creation, or change the implementation to actually lazily initialize the
executor on first use (e.g., in `broadcastHashTableBuildExecutor()` via
`std::call_once`), which would also align with the newly added `<mutex>`
include in the header.
##########
docs/velox-configuration.md:
##########
@@ -17,6 +17,7 @@ nav_order: 16
| 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.broadcastNLJ.fullOuterRewriteThreshold
| 🔄 Dynamic | 10MB | Maximum per-side plan size in bytes for
rewriting a full outer broadcast nested loop join into a left outer join and an
existence join followed by union. The rewrite is applied only when both sides
have known statistics and each side is at or below this threshold. Set to -1 to
disable the rewrite.
|
+| 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 documented default says it matches 'maximum task slots per Spark
executor', but `VeloxListenerApi` applies a different default on the driver
(`getDriverCores`) vs executors (`getTaskSlots`). Also the table default column
is `<undefined>` even though a defaulting behavior is described. Please update
this row to reflect the driver-vs-executor behavior (and/or clarify where the
default is computed) so operators aren’t misled.
##########
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, 1));
Review Comment:
The implementation doesn’t match the documented/PR-intended defaulting
behavior. Here the native default is effectively 1 thread (and `<= 0` becomes 1
due to `std::max`), but the PR description/docs state `<= 0` should fall back
to `kNumTaskSlotsPerExecutor` (min 1). This affects users who explicitly set 0
to request fallback, and any path where the Scala listener doesn’t populate the
key. Consider reading the configured value into a variable, and if it’s `<= 0`,
derive the thread count from `kNumTaskSlotsPerExecutor` (then clamp to at least
1); otherwise use the configured positive value.
##########
gluten-core/src/main/scala/org/apache/spark/util/SparkResourceUtil.scala:
##########
@@ -75,6 +75,21 @@ object SparkResourceUtil extends Logging {
cores
}
+ def getDriverCores(conf: SparkConf): Int = {
+ val master = conf.get("spark.master")
+
+ def convertToInt(threads: String): Int = {
+ if (threads == "*") Runtime.getRuntime.availableProcessors() else
threads.toInt
+ }
Review Comment:
This introduces a second local `convertToInt` that duplicates the one
already embedded in `getExecutorCores`. Since both parse the same
`SparkMasterRegex` patterns, consider extracting a shared private helper (e.g.,
a `parseLocalMasterThreads(master: String): Option[Int]` or `convertToInt` at
object scope) to reduce duplication and keep the parsing logic consistent over
time.
##########
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")
Review Comment:
This test asserts `buildThreads > 1` later, but it doesn’t explicitly
configure
`spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads` (and
in some environments the computed default may be 1, e.g., `local` master or
constrained CI). To avoid environment-dependent flakes, set
`spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads`
explicitly (e.g., 2) in `sparkConf` or within `withSQLConf` for the test.
--
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]
