This is an automated email from the ASF dual-hosted git repository.
philo-he pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 844241f500 [VL] Use a dedicated CPU thread pool for parallel broadcast
hash-table build (#13115)
844241f500 is described below
commit 844241f500d65b31b41e7d9f9616324e2f48be9c
Author: Kaifei Yi <[email protected]>
AuthorDate: Sat Oct 17 05:09:06 2026 +0800
[VL] Use a dedicated CPU thread pool for parallel broadcast hash-table
build (#13115)
---
.../org/apache/gluten/config/VeloxConfig.scala | 10 ++++
.../joins/GlutenParallelBroadcastJoinSuite.scala | 67 ++++++++++++++++++++++
cpp/velox/compute/VeloxBackend.cc | 6 ++
cpp/velox/compute/VeloxBackend.h | 8 +++
cpp/velox/config/VeloxConfig.h | 7 +++
cpp/velox/jni/VeloxJniWrapper.cc | 8 +--
cpp/velox/tests/CMakeLists.txt | 1 +
cpp/velox/tests/VeloxBackendTest.cc | 48 ++++++++++++++++
docs/velox-configuration.md | 1 +
9 files changed, 152 insertions(+), 4 deletions(-)
diff --git
a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
index 6d66080a23..f2d188e818 100644
--- a/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
+++ b/backends-velox/src/main/scala/org/apache/gluten/config/VeloxConfig.scala
@@ -249,6 +249,16 @@ object VeloxConfig extends ConfigRegistry {
.intConf
.createOptional
+ val COLUMNAR_VELOX_BROADCAST_HASH_TABLE_BUILD_THREADS =
+
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.")
+ .intConf
+ .createOptional
+
val COLUMNAR_VELOX_BROADCAST_HASH_TABLE_BUILD_TARGET_BYTES =
buildStaticConf("spark.gluten.velox.broadcast.build.targetBytesPerThread")
.doc(
diff --git
a/backends-velox/src/test/scala/org/apache/spark/sql/execution/joins/GlutenParallelBroadcastJoinSuite.scala
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/joins/GlutenParallelBroadcastJoinSuite.scala
new file mode 100644
index 0000000000..ce15734871
--- /dev/null
+++
b/backends-velox/src/test/scala/org/apache/spark/sql/execution/joins/GlutenParallelBroadcastJoinSuite.scala
@@ -0,0 +1,67 @@
+/*
+ * 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,
VeloxWholeStageTransformerSuite}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.execution.ColumnarBroadcastExchangeExec
+
+class GlutenParallelBroadcastJoinSuite extends VeloxWholeStageTransformerSuite
{
+
+ override protected val resourcePath: String = "placeholder"
+ override protected val fileFormat: String = "parquet"
+
+ 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")
+
.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)
+ }
+ }
+ }
+ }
+ }
+}
diff --git a/cpp/velox/compute/VeloxBackend.cc
b/cpp/velox/compute/VeloxBackend.cc
index a3e1dbd9cc..c131c008f4 100644
--- a/cpp/velox/compute/VeloxBackend.cc
+++ b/cpp/velox/compute/VeloxBackend.cc
@@ -14,6 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+#include <algorithm>
#include <filesystem>
#include "VeloxBackend.h"
@@ -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));
+ broadcastHashTableBuildExecutor_ =
std::make_unique<folly::CPUThreadPoolExecutor>(broadcastHashTableBuildThreads);
+
initJolFilesystem();
velox::dwio::common::registerFileSinks();
@@ -472,6 +477,7 @@ void VeloxBackend::tearDown() {
spillExecutor_.reset();
ioExecutor_.reset();
ssdCacheExecutor_.reset();
+ broadcastHashTableBuildExecutor_.reset();
globalMemoryManager_.reset();
// dump cache stats on exit if enabled
diff --git a/cpp/velox/compute/VeloxBackend.h b/cpp/velox/compute/VeloxBackend.h
index 09fad04dc4..5e129b9059 100644
--- a/cpp/velox/compute/VeloxBackend.h
+++ b/cpp/velox/compute/VeloxBackend.h
@@ -23,6 +23,7 @@
#include <boost/uuid/uuid_io.hpp>
#include <folly/executors/ThreadPoolExecutor.h>
#include <filesystem>
+#include <mutex>
#include "velox/common/caching/AsyncDataCache.h"
#include "velox/common/config/Config.h"
@@ -53,6 +54,12 @@ class VeloxBackend {
ReaderThreadPool* getReaderThreadPool();
+ /// A dedicated CPU thread pool for the parallel broadcast hash-table build.
The build is
+ /// CPU-bound, so it must not share ioExecutor() which is reserved for async
IO tasks.
+ folly::Executor* broadcastHashTableBuildExecutor() const {
+ return broadcastHashTableBuildExecutor_.get();
+ }
+
std::shared_ptr<facebook::velox::config::ConfigBase> getBackendConf() const {
return backendConf_;
}
@@ -135,6 +142,7 @@ class VeloxBackend {
std::shared_ptr<facebook::velox::config::ConfigBase> backendConf_;
std::unique_ptr<ReaderThreadPool> readerThreadPool_;
+ std::unique_ptr<folly::Executor> broadcastHashTableBuildExecutor_;
};
} // namespace gluten
diff --git a/cpp/velox/config/VeloxConfig.h b/cpp/velox/config/VeloxConfig.h
index dcb806274b..91069e86a2 100644
--- a/cpp/velox/config/VeloxConfig.h
+++ b/cpp/velox/config/VeloxConfig.h
@@ -169,6 +169,13 @@ const std::string kVeloxAsyncTimeoutOnTaskStopping =
"spark.gluten.sql.columnar.backend.velox.asyncTimeoutOnTaskStopping";
const int32_t kVeloxAsyncTimeoutOnTaskStoppingDefault = 30000; // 30s
+// By default, executors use the number of task slots per executor, with a
+// minimum of one thread. When driver-side broadcast hash-table build is
enabled
+// and Spark is not running in local mode, Scala-side logic in VeloxListenerApi
+// overrides the driver default to use spark.driver.cores instead.
+const std::string kBroadcastHashTableBuildThreads =
+ "spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads";
+
// udf
const std::string kVeloxUdfLibraryPaths =
"spark.gluten.sql.columnar.backend.velox.internal.udfLibraryPaths";
diff --git a/cpp/velox/jni/VeloxJniWrapper.cc b/cpp/velox/jni/VeloxJniWrapper.cc
index 1c91c11c18..12e1872376 100644
--- a/cpp/velox/jni/VeloxJniWrapper.cc
+++ b/cpp/velox/jni/VeloxJniWrapper.cc
@@ -1065,9 +1065,9 @@ JNIEXPORT jlong JNICALL
Java_org_apache_gluten_vectorized_HashJoinBuilder_native
return gluten::getHashTableObjStore()->save(builder);
}
- // Use thread pool (executor) instead of creating threads directly.
- // FIXME: This reuses the io executor which is supposed to only serve async
IO tasks.
- auto executor = VeloxBackend::get()->ioExecutor();
+ // Use a dedicated CPU thread pool for the parallel hash-table build. The
build is CPU-bound,
+ // so it must not run on the io executor which is reserved for async IO
tasks.
+ auto executor = VeloxBackend::get()->broadcastHashTableBuildExecutor();
std::vector<std::shared_ptr<gluten::HashTableBuilder>>
hashTableBuilders(numThreads);
std::vector<std::unique_ptr<facebook::velox::exec::BaseHashTable>>
otherTables(numThreads);
@@ -1138,7 +1138,7 @@ JNIEXPORT jlong JNICALL
Java_org_apache_gluten_vectorized_HashJoinBuilder_native
facebook::velox::exec::BaseHashTable::kNoSpillInputStartPartitionBit,
hashTableBuilders[0]->joinBuildVectorHasherMaxNumDistinct(),
hashTableBuilders[0]->dropDuplicates(),
- allowParallelJoinBuild ? VeloxBackend::get()->ioExecutor() : nullptr);
+ allowParallelJoinBuild ?
VeloxBackend::get()->broadcastHashTableBuildExecutor() : nullptr);
for (int i = 1; i < numThreads; ++i) {
if (hashTableBuilders[i]->joinHasNullKeys()) {
diff --git a/cpp/velox/tests/CMakeLists.txt b/cpp/velox/tests/CMakeLists.txt
index 6a10641fbf..369c92f5d4 100644
--- a/cpp/velox/tests/CMakeLists.txt
+++ b/cpp/velox/tests/CMakeLists.txt
@@ -140,6 +140,7 @@ add_velox_test(
add_velox_test(spark_functions_test SOURCES SparkFunctionTest.cc
FunctionTest.cc)
add_velox_test(runtime_test SOURCES RuntimeTest.cc)
+add_velox_test(velox_backend_test SOURCES VeloxBackendTest.cc)
add_velox_test(velox_memory_test SOURCES MemoryManagerTest.cc)
add_velox_test(buffer_outputstream_test SOURCES BufferOutputStreamTest.cc)
add_velox_test(velox_whole_stage_dumper_test SOURCES
diff --git a/cpp/velox/tests/VeloxBackendTest.cc
b/cpp/velox/tests/VeloxBackendTest.cc
new file mode 100644
index 0000000000..65bacf6f1f
--- /dev/null
+++ b/cpp/velox/tests/VeloxBackendTest.cc
@@ -0,0 +1,48 @@
+/*
+ * 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();
+ 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);
+}
+
+} // namespace gluten
diff --git a/docs/velox-configuration.md b/docs/velox-configuration.md
index 6cc3995645..1966c511fb 100644
--- a/docs/velox-configuration.md
+++ b/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 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.
[...]
|
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.cacheEnabled
| âš“ Static | false | Enable Velox cache, default off. It's
recommended to enablesoft-affinity as well when enable velox cache.
[...]
| spark.gluten.sql.columnar.backend.velox.cachePrefetchMinPct
| âš“ Static | 0 | Set prefetch cache min pct for velox
file scan
[...]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
