Copilot commented on code in PR #13368:
URL: https://github.com/apache/gluten/pull/13368#discussion_r4314007731
##########
backends-velox/src/test/scala/org/apache/spark/sql/execution/joins/GlutenParallelBroadcastJoinSuite.scala:
##########
@@ -32,36 +33,113 @@ class GlutenParallelBroadcastJoinSuite extends
VeloxWholeStageTransformerSuite {
.set("spark.gluten.velox.broadcast.build.targetBytesPerThread", "8KB")
.set("spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads",
"4")
- for (buildOnDriver <- Seq(false, true)) {
- test(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)
- if (buildOnDriver) {
- assert(exchanges.head.metrics("serializedHashTableSize").value > 0)
+ private val threadsKey =
+ "spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads"
+ private val taskSlotsKey = "spark.gluten.numTaskSlotsPerExecutor"
+
+ test("broadcast build threads: resolve process defaults without modifying
SparkConf") {
+ val conf = new SparkConf(false)
+ .setMaster("yarn")
+ .set("spark.driver.cores", "6")
+ .set("spark.executor.cores", "16")
+ .set("spark.task.cpus", "2")
+ .set(taskSlotsKey, "8")
+ val original = conf.getAll.toMap
+
+ assert(VeloxListenerApi.parseConf(conf, isDriver = true)(threadsKey) ==
"6")
+ assert(conf.getAll.toMap == original)
+ assert(VeloxListenerApi.parseConf(conf, isDriver = false)(threadsKey) ==
"8")
+ assert(conf.getAll.toMap == original)
+ }
+
+ test("broadcast build threads: explicit thread count takes precedence on
both " +
+ "driver and executor") {
+ val conf = new SparkConf(false)
+ .set("spark.driver.cores", "6")
+ .set(taskSlotsKey, "8")
+ .set(threadsKey, "3")
+
+ Seq(true, false).foreach {
+ isDriver => assert(VeloxListenerApi.parseConf(conf,
isDriver)(threadsKey) == "3")
+ }
+ assert(conf.get(threadsKey) == "3")
+ }
+
+ test("broadcast build threads: missing resource settings and spark.master
default " +
+ "to one thread") {
+ Seq(true, false).foreach {
+ isDriver =>
+ assert(VeloxListenerApi.parseConf(new SparkConf(false),
isDriver)(threadsKey) == "1")
+ }
+ }
+
+ test(
+ "broadcast build threads: driver defaults to one core independently of
executor task slots") {
+ val conf = new SparkConf(false).set(taskSlotsKey, "8")
+ assert(VeloxListenerApi.parseConf(conf, isDriver = true)(threadsKey) ==
"1")
+ }
+
+ test("broadcast build threads: local pool uses master cores instead of task
slots") {
+ val masters = Seq(
+ "local" -> 1,
+ "local[6]" -> 6,
+ "local[6,2]" -> 6,
+ "local[*]" -> Runtime.getRuntime.availableProcessors(),
+ "local[*,2]" -> Runtime.getRuntime.availableProcessors()
+ )
Review Comment:
Using `Runtime.getRuntime.availableProcessors()` in assertions can be
brittle across CI environments (e.g., container CPU limits, differing JVM
visibility) and can drift from Spark’s own parsing of `local[*]`. To make the
test robust and aligned with production logic, compute the expected value using
the same code path used in `parseConf` for local mode (e.g.,
`SparkResourceUtil.getExecutorCores(conf)` after setting `master`).
##########
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();
+ }();
+ broadcastHashTableBuildExecutor_ =
+ std::make_unique<folly::CPUThreadPoolExecutor>(std::max<int32_t>(1,
broadcastHashTableBuildThreads));
Review Comment:
Logging this as `WARNING` whenever the setting is absent can be noisy in
native embedding scenarios (where omitting the key is expected per the comment
in `VeloxConfig.h`). Consider downgrading this to `INFO`/`VLOG`, or gating it
(e.g., log once) so normal defaulting doesn’t produce warning-level logs.
##########
cpp/velox/tests/VeloxBackendTest.cc:
##########
@@ -18,31 +18,56 @@
#include <folly/executors/CPUThreadPoolExecutor.h>
#include <folly/futures/Future.h>
#include <gtest/gtest.h>
+#include <optional>
+#include <string>
#include <thread>
+#include <unordered_map>
#include "compute/VeloxBackend.h"
#include "config/VeloxConfig.h"
namespace gluten {
-class VeloxBackendTest : public ::testing::Test {
+struct BroadcastBuildThreadsTestParam {
+ int taskSlots;
+ std::optional<int> buildThreads;
+ int expectedThreads;
+};
+
+class VeloxBackendTest : public
::testing::TestWithParam<BroadcastBuildThreadsTestParam> {
protected:
- static void SetUpTestSuite() {
- VeloxBackend::create(AllocationListener::noop(),
{{kNumTaskSlotsPerExecutor, "0"}});
+ void SetUp() override {
+ const auto& param = GetParam();
+ std::unordered_map<std::string, std::string>
conf{{kNumTaskSlotsPerExecutor, std::to_string(param.taskSlots)}};
+ if (param.buildThreads.has_value()) {
+ conf.emplace(kBroadcastHashTableBuildThreads,
std::to_string(param.buildThreads.value()));
+ }
+ VeloxBackend::create(AllocationListener::noop(), conf);
}
- static void TearDownTestSuite() {
+ void TearDown() override {
VeloxBackend::get()->tearDown();
}
};
-TEST_F(VeloxBackendTest, broadcastHashTableBuildExecutorWithZeroTaskSlots) {
+TEST_P(VeloxBackendTest, broadcastHashTableBuildExecutorThreadCount) {
Review Comment:
The parameter and test names are a bit ambiguous relative to the PR’s
driver-vs-executor defaulting semantics (these C++ tests are validating the
native fallback behavior when the key is absent). Consider renaming to make
intent explicit, e.g. `NativeBroadcastBuildThreadsFallbackParam` and
`broadcastHashTableBuildThreads_nativeFallbackToTaskSlots`, so readers don’t
confuse this with the Spark-side per-process default resolution.
##########
cpp/velox/tests/VeloxBackendTest.cc:
##########
@@ -18,31 +18,56 @@
#include <folly/executors/CPUThreadPoolExecutor.h>
#include <folly/futures/Future.h>
#include <gtest/gtest.h>
+#include <optional>
+#include <string>
#include <thread>
+#include <unordered_map>
#include "compute/VeloxBackend.h"
#include "config/VeloxConfig.h"
namespace gluten {
-class VeloxBackendTest : public ::testing::Test {
+struct BroadcastBuildThreadsTestParam {
+ int taskSlots;
+ std::optional<int> buildThreads;
+ int expectedThreads;
+};
Review Comment:
The parameter and test names are a bit ambiguous relative to the PR’s
driver-vs-executor defaulting semantics (these C++ tests are validating the
native fallback behavior when the key is absent). Consider renaming to make
intent explicit, e.g. `NativeBroadcastBuildThreadsFallbackParam` and
`broadcastHashTableBuildThreads_nativeFallbackToTaskSlots`, so readers don’t
confuse this with the Spark-side per-process default resolution.
--
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]
