[VL] Use a dedicated CPU thread pool for parallel broadcast hash-table build

[yikf]

What changes are proposed in this pull request?

The parallel broadcast hash-table build path (numThreads > 1 in HashJoinBuilder.nativeBuild) dispatched its CPU-bound build tasks onto VeloxBackend::ioExecutor(). That executor is reserved for async IO tasks (split preloading, DirectBufferedInput).

// FIXME: This reuses the io executor which is supposed to only serve async IO tasks.
auto executor = VeloxBackend::get()->ioExecutor();
Running CPU-heavy hash-table build work on the IO pool makes it contend with
genuine IO work on the same threads.

This PR introduces a dedicated CPU thread pool (VeloxBackend:: broadcastHashTableBuildExecutor()), following the existing getReaderThreadPool() std::call_once pattern, and switches both places that previously reused the io executor:

1. the parallel build task dispatch (folly::via(executor, ...)), and
2. the parallel join-table merge passed to prepareJoinTable(...).

The pool size is controlled by a new native config key spark.gluten.sql.columnar.backend.velox.broadcastHashTableBuildThreads. if miss this config falls back to kNumTaskSlotsPerExecutor, i.e. the same sizing this path previously inherited from the io executor. So default behavior is unchanged — this is a pool-isolation change, not a concurrency change. The key is native-only and passed through via the spark.gluten.sql.columnar.backend.velox. prefix, mirroring how spillThreadNum is handled.

How was this patch tested?

GA.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude claude-opus-4-8

[Copilot]

Copilot review overview

🟡 Changes recommended

Two critical executor initialization issues and a moderate fallback-sizing issue remain unresolved; documentation also needs updating.

Get a fresh assessment by requesting another Copilot review.

Review effort: Lite
Findings: 2

Open (2)

• Function-static guard prevents executor reinitialization · New
• Zero task slots create an invalid executor · New

What changed in this PR

This PR adds a dedicated, configurable CPU executor for parallel broadcast hash-table construction, separating it from Velox’s I/O pool.

Changes:

• Routes parallel build and merge work through the new executor.
• Adds native thread-count configuration.
• Adds executor lifecycle management.

File	Summary
cpp/​velox/​jni/​VeloxJniWrapper.cc	Uses the dedicated executor for hash-table build and merge operations.
cpp/​velox/​config/​VeloxConfig.h	Defines the new pool-size configuration key.
cpp/​velox/​compute/​VeloxBackend.h	Declares executor access and ownership.
cpp/​velox/​compute/​VeloxBackend.cc	Creates and tears down the executor; lifecycle, zero-thread fallback, and compatibility sizing require changes.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[Copilot]

Copilot review overview

🔵 Needs a closer look

Unresolved moderate issues affect executor reinitialization, compatibility, zero-worker behavior, and test coverage.

Review effort: Lite
Findings: 2

Open (2)

• Zero task slots create an invalid executor
• Function-static guard prevents executor reinitialization

[yikf]

@zhztheplayer Could you please take a look if you find a time, thanks.

[Copilot]

Copilot review overview

🔵 Needs a closer look

Address the executor reinitialization and default thread-count compatibility issues; document the new configuration key.

Review effort: Lite
Findings: 1

Open (1)

• Function-static guard prevents executor reinitialization

Resolved since last review (1)

• Zero task slots create an invalid executor

[Copilot]

Copilot review overview

🔵 Needs a closer look

Backend reinitialization and IO-thread fallback behavior need correction before approval.

Review effort: Lite
Findings: 1

Open (1)

• Function-static guard prevents executor reinitialization

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Copilot review overview

🔵 Needs a closer look

Address the IO-thread fallback compatibility issue and add coverage for positive thread-count configuration.

Review effort: Lite
Findings: None

Resolved since last review (1)

• Function-static guard prevents executor reinitialization

[zhztheplayer]

Thanks for the improvement.

cc @JkSelf

[zhztheplayer]

Minor questions:

1. Can we initialize hashTableBuildExecutor_ in VeloxBackend::init()?
2. Should we set the default value as numTaskSlotsPerExecutor?

[yikf]

@zhztheplayer Thanks for your suggestion!

1. hashTableBuildExecutor_ is only used during native build, so lazy initialization is a better choice here, what do you think?

2. Sounds good! Updated accordingly.

[zhztheplayer]

hashTableBuildExecutor_ is only used during native build, so lazy initialization is a better choice here, what do you think?

The thread creation should already be lazy in folly::CPUThreadPoolExecutor, so I assume std::make_unique<folly::CPUThreadPoolExecutor>(numThreads) is a cheap operation?

I asked the question mainly because it's not aligned with initialization code of other executors.

[JkSelf]

Should we set the default value as numTaskSlotsPerExecutor?

Since the driver and executor build hash table share same code path, we need to distinguish the numTaskSlotsPerExecutor and numTaskSlotPerDriver.

[yikf]

the default value has been updated, and we now distinguish between numTaskSlotsPerExecutor and numTaskSlotPerDriver.

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Copilot review overview

🟡 Changes recommended

Resolve the non-positive thread-count fallback before approval.

Review effort: Lite
Findings: 1

Open (1)

• Resolve configured thread fallback before minimum-one clamp · New

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Note

Copilot was unable to run its full agentic suite in this review.

Copilot review overview

15 open findings

• The test config forces broadcastHashTableBuildThreads to "1" but later asserts `buildThreads >… · New
• The test config forces broadcastHashTableBuildThreads to "1" but later asserts `buildThreads >… · New
• Defaulting the dedicated CPU pool size to numTaskSlotsPerExecutor can create a large number of… · New
• The dedicated CPUThreadPoolExecutor is created unconditionally during VeloxBackend::init(),…
• The test only sets kNumTaskSlotsPerExecutor=0 but then asserts ioExecutor() is nullptr. That…
• The default column shows <undefined>, but the description states a concrete default behavior…
• This creates an additional CPU thread pool sized (by default) to numTaskSlotsPerExecutor, which…
• This assertion can be brittle because the suite never pins the broadcast hash-table build…
• The PR description references a different config key name (`spark.gluten.sql.columnar.backend.velox.…
• Use task slots for driver-side default
• Fall back to task slots for non-positive thread settings
• Create the broadcast build thread pool lazily
• The PR description refers to a VeloxBackend::hashTableBuildExecutor() API, but the actual API…
• The documented default here is incomplete relative to the behavior introduced in Scala…
• This introduces a second local convertToInt that duplicates the one already embedded in…

3 resolved since last review

• The executor is always constructed as a folly::CPUThreadPoolExecutor, but is stored as…
• This unconditionally creates (and typically starts) an additional CPU thread pool for every…
• New behavior is introduced for overriding the pool size via kBroadcastHashTableBuildThreads, but…

🧠 Review effort: Lite

[Copilot]

Note

Copilot was unable to run its full agentic suite in this review.

Copilot review overview

19 open findings

• The test forces broadcastHashTableBuildThreads to 1 but then asserts buildThreads > 1. With a… · New
• The test forces broadcastHashTableBuildThreads to 1 but then asserts buildThreads > 1. With a… · New
• The test config forces broadcastHashTableBuildThreads to "1" but later asserts `buildThreads >…
• The test config forces broadcastHashTableBuildThreads to "1" but later asserts `buildThreads >…
• This creates an additional CPUThreadPoolExecutor unconditionally during backend init. With the… · New
• The default column is set to <undefined>, but the description specifies a concrete defaulting… · New
• Defaulting the dedicated CPU pool size to numTaskSlotsPerExecutor can create a large number of…
• The dedicated CPUThreadPoolExecutor is created unconditionally during VeloxBackend::init(),…
• The test only sets kNumTaskSlotsPerExecutor=0 but then asserts ioExecutor() is nullptr. That…
• The default column shows <undefined>, but the description states a concrete default behavior…
• This creates an additional CPU thread pool sized (by default) to numTaskSlotsPerExecutor, which…
• This assertion can be brittle because the suite never pins the broadcast hash-table build…
• The PR description references a different config key name (`spark.gluten.sql.columnar.backend.velox.…
• Use task slots for driver-side default
• Fall back to task slots for non-positive thread settings
• Create the broadcast build thread pool lazily
• The PR description refers to a VeloxBackend::hashTableBuildExecutor() API, but the actual API…
• The documented default here is incomplete relative to the behavior introduced in Scala…
• This introduces a second local convertToInt that duplicates the one already embedded in…

🧠 Review effort: Lite

[Copilot]

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Copilot review overview

16 open findings

• The test config forces broadcastHashTableBuildThreads to "1" but later asserts `buildThreads >…
• This creates an additional CPU thread pool unconditionally during backend init. Even if parallel… · New
• This test relies on dynamic_cast to folly::CPUThreadPoolExecutor, which depends on RTTI being… · New
• The test only sets kNumTaskSlotsPerExecutor=0 but then asserts ioExecutor() is nullptr. That…
• The default column shows <undefined>, but the description states a concrete default behavior…
• This creates an additional CPU thread pool sized (by default) to numTaskSlotsPerExecutor, which…
• This assertion can be brittle because the suite never pins the broadcast hash-table build…
• The PR description references a different config key name (`spark.gluten.sql.columnar.backend.velox.…
• Use task slots for driver-side default
• Fall back to task slots for non-positive thread settings
• Create the broadcast build thread pool lazily
• The PR description says this follows the existing std::call_once pattern (like… · New
• The 'Default' column says <undefined>, but the description states a concrete default behavior… · New
• The PR description refers to a VeloxBackend::hashTableBuildExecutor() API, but the actual API…
• The documented default here is incomplete relative to the behavior introduced in Scala…
• This introduces a second local convertToInt that duplicates the one already embedded in…

7 resolved since last review

• The test forces broadcastHashTableBuildThreads to 1 but then asserts buildThreads > 1. With a…
• The test forces broadcastHashTableBuildThreads to 1 but then asserts buildThreads > 1. With a…
• The test config forces broadcastHashTableBuildThreads to "1" but later asserts `buildThreads >…
• The default column is set to <undefined>, but the description specifies a concrete defaulting…
• This creates an additional CPUThreadPoolExecutor unconditionally during backend init. With the…
• Defaulting the dedicated CPU pool size to numTaskSlotsPerExecutor can create a large number of…
• The dedicated CPUThreadPoolExecutor is created unconditionally during VeloxBackend::init(),…

🧠 Review effort: Lite

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Copilot review overview

18 open findings

• The PR description says this follows the existing getReaderThreadPool() std::call_once pattern,… · New
• In this PR, broadcastHashTableBuildThreads is registered via buildStaticConf(...) (static).… · New
• This test relies on dynamic_cast to folly::CPUThreadPoolExecutor, which depends on RTTI being…
• This creates an additional CPU thread pool unconditionally during backend init. Even if parallel…
• The test only sets kNumTaskSlotsPerExecutor=0 but then asserts ioExecutor() is nullptr. That…
• The default column shows <undefined>, but the description states a concrete default behavior…
• This creates an additional CPU thread pool sized (by default) to numTaskSlotsPerExecutor, which…
• This assertion can be brittle because the suite never pins the broadcast hash-table build…
• The PR description references a different config key name (`spark.gluten.sql.columnar.backend.velox.…
• Use task slots for driver-side default
• Fall back to task slots for non-positive thread settings
• Create the broadcast build thread pool lazily
• This call fetches broadcastHashTableBuildExecutor() again even though an executor variable is… · New
• The 'Default' column says <undefined>, but the description states a concrete default behavior…
• The PR description says this follows the existing std::call_once pattern (like…
• The PR description refers to a VeloxBackend::hashTableBuildExecutor() API, but the actual API…
• The documented default here is incomplete relative to the behavior introduced in Scala…
• This introduces a second local convertToInt that duplicates the one already embedded in…

1 resolved since last review

• The test config forces broadcastHashTableBuildThreads to "1" but later asserts `buildThreads >…

🧠 Review effort: Lite

[github-actions]

Run Gluten Clickhouse CI on x86

[Copilot]

Note

Copilot was unable to run its full agentic suite in this review.

Copilot review overview

13 open findings

• This creates a dedicated thread pool unconditionally (minimum 1 thread) for every backend instance,… · New
• The test only sets kNumTaskSlotsPerExecutor=0 but then asserts ioExecutor() is nullptr. That…
• The default column shows <undefined>, but the description states a concrete default behavior…
• This creates an additional CPU thread pool sized (by default) to numTaskSlotsPerExecutor, which…
• This assertion can be brittle because the suite never pins the broadcast hash-table build…
• The PR description references a different config key name (`spark.gluten.sql.columnar.backend.velox.…
• Use task slots for driver-side default
• Fall back to task slots for non-positive thread settings
• Create the broadcast build thread pool lazily
• Other executors in this backend appear to use an explicit queue configuration (e.g.,… · New
• The PR description refers to a VeloxBackend::hashTableBuildExecutor() API, but the actual API…
• The documented default here is incomplete relative to the behavior introduced in Scala…
• This introduces a second local convertToInt that duplicates the one already embedded in…

7 resolved since last review

• In this PR, broadcastHashTableBuildThreads is registered via buildStaticConf(...) (static).…
• The PR description says this follows the existing getReaderThreadPool() std::call_once pattern,…
• This test relies on dynamic_cast to folly::CPUThreadPoolExecutor, which depends on RTTI being…
• This creates an additional CPU thread pool unconditionally during backend init. Even if parallel…
• This call fetches broadcastHashTableBuildExecutor() again even though an executor variable is…
• The 'Default' column says <undefined>, but the description states a concrete default behavior…
• The PR description says this follows the existing std::call_once pattern (like…

🧠 Review effort: Lite

[Copilot]

Note

Copilot was unable to run its full agentic suite in this review.

Copilot review overview

16 open findings

• Using a literal value like \"placeholder\" for resourcePath is unclear and can make failures… · New
• This unconditionally creates an additional CPU thread pool sized (by default) to the executor task… · New
• The production API exposes folly::Executor*, but the test relies on… · New
• This creates a dedicated thread pool unconditionally (minimum 1 thread) for every backend instance,…
• The test only sets kNumTaskSlotsPerExecutor=0 but then asserts ioExecutor() is nullptr. That…
• The default column shows <undefined>, but the description states a concrete default behavior…
• This creates an additional CPU thread pool sized (by default) to numTaskSlotsPerExecutor, which…
• This assertion can be brittle because the suite never pins the broadcast hash-table build…
• The PR description references a different config key name (`spark.gluten.sql.columnar.backend.velox.…
• Use task slots for driver-side default
• Fall back to task slots for non-positive thread settings
• Create the broadcast build thread pool lazily
• Other executors in this backend appear to use an explicit queue configuration (e.g.,…
• The PR description refers to a VeloxBackend::hashTableBuildExecutor() API, but the actual API…
• The documented default here is incomplete relative to the behavior introduced in Scala…
• This introduces a second local convertToInt that duplicates the one already embedded in…

🧠 Review effort: Lite

[Copilot]

Warning

Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.

Copilot review overview

21 open findings

• The PR description states 'default behavior is unchanged — this is a pool-isolation change, not a… · New
• This eagerly creates an additional thread pool sized (by default) similarly to the existing… · New
• The accessor returns a generic folly::Executor*, but the implementation always constructs a… · New
• This test relies on dynamic_cast (RTTI) to validate the executor type and inspect numThreads(). If… · New
• The production API exposes folly::Executor*, but the test relies on…
• This unconditionally creates an additional CPU thread pool sized (by default) to the executor task…
• Using a literal value like \"placeholder\" for resourcePath is unclear and can make failures…
• This creates a dedicated thread pool unconditionally (minimum 1 thread) for every backend instance,…
• The test only sets kNumTaskSlotsPerExecutor=0 but then asserts ioExecutor() is nullptr. That…
• The default column shows <undefined>, but the description states a concrete default behavior…
• This creates an additional CPU thread pool sized (by default) to numTaskSlotsPerExecutor, which…
• This assertion can be brittle because the suite never pins the broadcast hash-table build…
• The PR description references a different config key name (`spark.gluten.sql.columnar.backend.velox.…
• Use task slots for driver-side default
• Fall back to task slots for non-positive thread settings
• Create the broadcast build thread pool lazily
• Using a literal value like \"placeholder\" for resourcePath makes the intent unclear and can be… · New
• Other executors in this backend appear to use an explicit queue configuration (e.g.,…
• The PR description refers to a VeloxBackend::hashTableBuildExecutor() API, but the actual API…
• The documented default here is incomplete relative to the behavior introduced in Scala…

And 1 more that still need to be addressed.

🧠 Review effort: Lite

[philo-he]

Looks good. Thanks.

[JkSelf]

LGTM. Thanks for your fix.