Copilot commented on code in PR #13368:
URL: https://github.com/apache/gluten/pull/13368#discussion_r4313844561
##########
backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxListenerApi.scala:
##########
@@ -312,6 +312,22 @@ object VeloxListenerApi {
var parsed: Map[String, String] =
GlutenConfigUtil.parseConfig(conf.getAll.toMap)
+ val broadcastBuildThreadsKey =
COLUMNAR_VELOX_BROADCAST_HASH_TABLE_BUILD_THREADS.key
+ if (!parsed.contains(broadcastBuildThreadsKey)) {
+ val defaultThreads = if (isDriver) {
+ if (inLocalMode(conf)) {
+ SparkResourceUtil.getExecutorCores(conf)
+ } else {
+ conf.getInt("spark.driver.cores", 1)
+ }
+ } else {
+ conf.getInt(GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR.key, 1)
+ }
+ // Resolve defaults per process without writing the driver's value back
to SparkConf,
+ // which is also propagated to executors. Local mode shares the driver's
thread pool.
Review Comment:
In local mode you comment that the pool is shared, but the local-mode
default is only applied when `isDriver == true`. If `parseConf(..., isDriver =
false)` is ever used in local mode, it will default to task slots and can
diverge from the driver’s local-master-based sizing. Consider restructuring to
check `inLocalMode(conf)` first (so both driver and executor resolve to local
master cores), or otherwise enforce/document that the executor path is never
invoked in local mode. Adding a Scala test for `isDriver = false` with a
`local[*]` master would prevent regressions.
##########
cpp/velox/compute/VeloxBackend.cc:
##########
@@ -255,9 +255,16 @@ 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));
- broadcastHashTableBuildExecutor_ =
std::make_unique<folly::CPUThreadPoolExecutor>(broadcastHashTableBuildThreads);
+ const int32_t broadcastHashTableBuildThreads = [&]() {
+ if (!backendConf_->valueExists(kBroadcastHashTableBuildThreads)) {
+ LOG(WARNING) << kBroadcastHashTableBuildThreads << " is not set. Falling
back to " << kNumTaskSlotsPerExecutor
+ << " (" << numTaskSlotsPerExecutor << ").";
+ return numTaskSlotsPerExecutor;
+ }
+ return backendConf_->get<int32_t>(kBroadcastHashTableBuildThreads).value();
+ }();
Review Comment:
`kBroadcastHashTableBuildThreads` is documented/used as an optional setting;
logging at WARNING every time it is unset can create noisy logs (and in some
deployments, trigger alerting). Consider lowering this to INFO/VLOG, or
rate-limiting (e.g., log only once per process) while still keeping the
fallback behavior.
##########
backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala:
##########
@@ -253,9 +253,11 @@ object VeloxConfig extends ConfigRegistry {
buildStaticConf("spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads")
.doc(
"Size of the dedicated CPU thread pool for the parallel broadcast
hash-table build. " +
- "By default, the thread pool is sized to the maximum task slots per
Spark executor " +
- "on executors, or to spark.driver.cores on the driver (used when " +
- "driver-side broadcast hash-table build is enabled), with a minimum
of one thread.")
+ "If unset, the size is resolved separately for each process at
backend initialization: " +
+ "spark.gluten.numTaskSlotsPerExecutor on executors, and
spark.driver.cores on the " +
+ "driver. In local mode, the shared pool uses the number of cores
specified by " +
+ "spark.master (for example, local[N] uses N). An explicit value
applies to both " +
+ "driver and executors. The pool always has at least one thread.")
Review Comment:
This docstring reads as if the default resolution is always performed at
backend initialization, but the native backend code path still falls back to
`spark.gluten.numTaskSlotsPerExecutor` when the setting is absent (and relies
on JVM-side `VeloxListenerApi.parseConf` to apply driver/local defaults).
Consider clarifying here that per-process defaults are resolved by the
Spark-side configuration parsing (and that non-Spark/native callers default to
task slots unless they explicitly pass the resolved value).
--
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]
