Copilot commented on code in PR #13115:
URL: https://github.com/apache/gluten/pull/13115#discussion_r4229912768
##########
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:
This does not implement the documented/PR-stated behavior that a value <= 0
falls back to `kNumTaskSlotsPerExecutor` (with a minimum of 1). With the
current logic, `<= 0` becomes `1`, and the implicit default is also `1` when
the key is absent, which can change behavior in non-Spark/native entrypoints
(and prevents users from using `0`/negative to request fallback semantics).
Consider reading the configured value with a sentinel default (e.g., 0), and if
the value is <= 0, derive the pool size from the configured task slots (then
clamp to at least 1).
##########
cpp/velox/tests/VeloxBackendTest.cc:
##########
@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+
+#include <folly/executors/CPUThreadPoolExecutor.h>
+#include <folly/futures/Future.h>
+#include <gtest/gtest.h>
+#include <thread>
+
+#include "compute/VeloxBackend.h"
+#include "config/VeloxConfig.h"
+
+namespace gluten {
+
+class VeloxBackendTest : public ::testing::Test {
+ protected:
+ static void SetUpTestSuite() {
+ VeloxBackend::create(AllocationListener::noop(),
{{kNumTaskSlotsPerExecutor, "0"}});
+ }
+
+ static void TearDownTestSuite() {
+ VeloxBackend::get()->tearDown();
+ }
+};
+
+TEST_F(VeloxBackendTest, broadcastHashTableBuildExecutorWithZeroTaskSlots) {
+ auto* backend = VeloxBackend::get();
+ ASSERT_EQ(backend->ioExecutor(), nullptr);
+
+ auto* executor =
dynamic_cast<folly::CPUThreadPoolExecutor*>(backend->broadcastHashTableBuildExecutor());
+ ASSERT_NE(executor, nullptr);
+ ASSERT_EQ(executor->numThreads(), 1);
+ const auto callerThread = std::this_thread::get_id();
+ EXPECT_NE(folly::via(executor, [] { return std::this_thread::get_id();
}).get(), callerThread);
+}
+
Review Comment:
This test covers the 'zero task slots => min 1 thread' case, but it does not
cover the newly documented behavior that an explicit
`broadcastHashTableBuildThreads <= 0` falls back to task slots. Once the native
fallback logic is corrected, consider adding cases for (a)
`broadcastHashTableBuildThreads = 0/-1` with non-zero task slots to assert it
picks task slots, and (b) a positive explicit value to assert it is honored.
##########
cpp/velox/config/VeloxConfig.h:
##########
@@ -169,6 +169,11 @@ const std::string kVeloxAsyncTimeoutOnTaskStopping =
"spark.gluten.sql.columnar.backend.velox.asyncTimeoutOnTaskStopping";
const int32_t kVeloxAsyncTimeoutOnTaskStoppingDefault = 30000; // 30s
+// Size of the dedicated CPU thread pool for the parallel broadcast hash-table
build.
+// Defaults to the number of task slots per executor, with a minimum of one
thread.
+const std::string kBroadcastHashTableBuildThreads =
+ "spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads";
Review Comment:
The PR description references a different config key name
(`spark.gluten.sql.columnar.backend.velox.hashTableBuildThreads`), while the
code/docs use
`spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads`.
Please align the PR description with the implemented key (or rename the key if
the description is the intended API), since this affects how users
discover/configure the feature.
##########
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 assertion relies on environment-dependent defaults (e.g., driver cores
/ task slots) producing more than one build thread. To make the test
deterministic and avoid failures on single-core or constrained CI environments,
set `spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads`
explicitly to a value > 1 within the test, or relax the assertion to account
for environments where the computed default is 1.
--
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]
