You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gluten.apache.org by "zhztheplayer (via GitHub)" <gi...@apache.org> on 2026/06/15 14:31:13 UTC
[PR] Use Velox TaskCursor and add ThreadManager infrastructure [gluten]
zhztheplayer opened a new pull request, #12302:
URL: https://github.com/apache/gluten/pull/12302
Replace Task-based execution with TaskCursor in WholeStageResultIterator to enable future parallel execution support. Add ThreadInitializer and ThreadManager abstractions to manage native thread lifecycle, plumbed through the Runtime layer. This is a pure refactoring with no behavioral change — serial execution is still the default.
Includes:
- New ThreadInitializer / ThreadManager classes in cpp/core/threads/
- SparkThreadInitializer for JNI thread callbacks
- VeloxThreadManager backend implementation
- TaskCursor-based execution replacing direct Task management
- NumTaskSlotsPerExecutor config wired for IO threads default
- HookedExecutor plumbed with ThreadInitializer
PR authored by me and split from #11852 by AI
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #12302:
URL: https://github.com/apache/gluten/pull/12302#discussion_r3426712188
##########
cpp/core/compute/Runtime.h:
##########
@@ -189,6 +197,7 @@ class Runtime : public std::enable_shared_from_this<Runtime> {
protected:
std::string kind_;
MemoryManager* memoryManager_;
+ ThreadManager* threadManager_;
Review Comment:
Good point. The destruction is guaranteed to be right earlier than destruction of memory manager, by its priority settings:
https://github.com/zhztheplayer/gluten/blob/312f45bf71dcb68d601e39812df30ddca4a90c8f/gluten-arrow/src/main/scala/org/apache/gluten/threads/NativeThreadManager.scala#L57
> initially we run into some leakage from VeloxRuntime and we need to careful on destructor
I was aware of this, and this kind of leakage was fundamentally fixed at once by https://github.com/apache/gluten/pull/11882, specifically, by the introduction of HookedExecutor. HookedExecutor can reuse OS thread resources, while guarantees no background threads are running before we end the Velox task. Precondition is, Velox should keep avoiding running threads off the thread pools we pass to it, and this rule is unlikely to be broken by Velox shortly.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #12302:
URL: https://github.com/apache/gluten/pull/12302#discussion_r3426712188
##########
cpp/core/compute/Runtime.h:
##########
@@ -189,6 +197,7 @@ class Runtime : public std::enable_shared_from_this<Runtime> {
protected:
std::string kind_;
MemoryManager* memoryManager_;
+ ThreadManager* threadManager_;
Review Comment:
Good point. The destruction is guaranteed to be right earlier than destruction of memory manager, by its priority settings:
https://github.com/zhztheplayer/gluten/blob/312f45bf71dcb68d601e39812df30ddca4a90c8f/gluten-arrow/src/main/scala/org/apache/gluten/threads/NativeThreadManager.scala#L57
> initially we run into some leakage from VeloxRuntime and we need to careful on destructor
I was aware of this, and this kind of leakage was fundamentally fixed at once by https://github.com/apache/gluten/pull/11882, specifically, by the introduction of HookedExecutor. HookedExecutor can reuse OS thread resources, while it guarantees no background threads are running before we end the Velox task. Precondition is, Velox should keep avoiding running threads off the thread pools we pass to it, and this rule is unlikely to be broken by Velox shortly.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #12302:
URL: https://github.com/apache/gluten/pull/12302#discussion_r3419951815
##########
cpp/core/jni/JniCommon.h:
##########
@@ -549,3 +550,86 @@ class JavaRssClient : public RssClient {
jmethodID javaPushPartitionData_;
jbyteArray array_;
};
+
+/// Bridges gluten::ThreadInitializer callbacks to a Java-side
+/// NativeThreadInitializer instance via JNI.
+///
+/// On initialize(), attaches the current native thread to the JVM and calls
+/// into the Java initializer so it can install Spark TaskContext or other
+/// thread-local state. On destroy(), calls the Java destroy() method but
+/// does NOT detach the thread — the underlying JVM thread may be reused by
+/// the pool, and detaching prematurely could allow the Java Thread object
+/// to be garbage-collected.
+class SparkThreadInitializer final : public gluten::ThreadInitializer {
+ public:
+ /// @param vm The JavaVM pointer from JNI_OnLoad.
+ /// @param jInitializerLocalRef A local reference to a Java object
+ /// implementing org.apache.gluten.threads.NativeThreadInitializer.
+ /// A global reference is created internally.
+ SparkThreadInitializer(JavaVM* vm, jobject jInitializerLocalRef) : vm_(vm) {
+ JNIEnv* env;
+ attachCurrentThreadAsDaemonOrThrow(vm_, &env);
+ jInitializerGlobalRef_ = env->NewGlobalRef(jInitializerLocalRef);
+ GLUTEN_CHECK(jInitializerGlobalRef_ != nullptr, "Failed to create global reference for native thread initializer.");
+ (void)initializeMethod(env);
+ }
+
+ SparkThreadInitializer(const SparkThreadInitializer&) = delete;
+ SparkThreadInitializer(SparkThreadInitializer&&) = delete;
+ SparkThreadInitializer& operator=(const SparkThreadInitializer&) = delete;
+ SparkThreadInitializer& operator=(SparkThreadInitializer&&) = delete;
+
+ ~SparkThreadInitializer() override {
+ JNIEnv* env;
+ if (vm_->GetEnv(reinterpret_cast<void**>(&env), jniVersion) != JNI_OK) {
+ LOG(WARNING) << "SparkThreadInitializer#~SparkThreadInitializer(): "
+ << "JNIEnv was not attached to current thread";
+ return;
+ }
+ env->DeleteGlobalRef(jInitializerGlobalRef_);
+ }
Review Comment:
If the destructor runs on a thread not attached to the JVM, it returns early and leaks `jInitializerGlobalRef_`. Consider attaching in the destructor (or otherwise ensuring a valid `JNIEnv*`) and deleting the global ref reliably; since this is a destructor, avoid throwing (e.g., attach and log on failure).
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12302:
URL: https://github.com/apache/gluten/pull/12302#issuecomment-4717013777
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12302:
URL: https://github.com/apache/gluten/pull/12302#issuecomment-4727659373
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #12302:
URL: https://github.com/apache/gluten/pull/12302#discussion_r3419951736
##########
cpp/velox/compute/WholeStageResultIterator.h:
##########
@@ -137,12 +139,13 @@ class WholeStageResultIterator : public SplitAwareColumnarBatchIterator {
#endif
const SparkTaskInfo taskInfo_;
folly::Executor* executor_;
- std::shared_ptr<facebook::velox::exec::Task> task_;
+ std::unique_ptr<facebook::velox::exec::TaskCursor> cursor_;
+ facebook::velox::exec::Task* task_ = nullptr;
std::shared_ptr<const facebook::velox::core::PlanNode> veloxPlan_;
/// Spill.
std::string spillStrategy_;
- folly::Executor* spillExecutor_ = nullptr;
+ folly::Executor* spillExecutor_;
Review Comment:
`spillExecutor_` is no longer default-initialized (previously `= nullptr`). If any constructor path or usage assumes a null default, this is undefined behavior. Initialize it to `nullptr` (or set it in the constructor initializer list) to keep pointer state deterministic.
##########
cpp/velox/compute/WholeStageResultIterator.h:
##########
@@ -137,12 +139,13 @@ class WholeStageResultIterator : public SplitAwareColumnarBatchIterator {
#endif
const SparkTaskInfo taskInfo_;
folly::Executor* executor_;
- std::shared_ptr<facebook::velox::exec::Task> task_;
+ std::unique_ptr<facebook::velox::exec::TaskCursor> cursor_;
+ facebook::velox::exec::Task* task_ = nullptr;
Review Comment:
`spillExecutor_` is no longer default-initialized (previously `= nullptr`). If any constructor path or usage assumes a null default, this is undefined behavior. Initialize it to `nullptr` (or set it in the constructor initializer list) to keep pointer state deterministic.
##########
cpp/velox/compute/WholeStageResultIterator.cc:
##########
@@ -307,41 +307,24 @@ std::shared_ptr<velox::core::QueryCtx> WholeStageResultIterator::createNewVeloxQ
}
std::shared_ptr<ColumnarBatch> WholeStageResultIterator::next() {
- if (task_->isFinished()) {
- return nullptr;
- }
- velox::RowVectorPtr vector;
while (true) {
- auto future = velox::ContinueFuture::makeEmpty();
- auto out = task_->next(&future);
- if (!future.valid()) {
- // Not need to wait. Break.
- vector = std::move(out);
- break;
+ if (!cursor_->moveNext()) {
+ return nullptr;
}
- // Velox suggested to wait. This might be because another thread (e.g., background io thread) is spilling the task.
- GLUTEN_CHECK(out == nullptr, "Expected to wait but still got non-null output from Velox task");
- VLOG(2) << "Velox task " << task_->taskId()
- << " is busy when ::next() is called. Will wait and try again. Task state: "
- << taskStateString(task_->state());
- future.wait();
- }
- if (vector == nullptr) {
- return nullptr;
- }
- uint64_t numRows = vector->size();
- if (numRows == 0) {
- return nullptr;
- }
-
- {
- ScopedTimer timer(&loadLazyVectorTime_);
- for (auto& child : vector->children()) {
- child->loadedVector();
+ RowVectorPtr vector = cursor_->current();
+ GLUTEN_CHECK(vector != nullptr, "Cursor returned null vector.");
+ uint64_t numRows = vector->size();
+ if (numRows == 0) {
+ continue;
+ }
Review Comment:
This changes behavior vs the previous implementation, which returned `nullptr` when `numRows == 0`. Now it skips empty batches and continues. If empty vectors can be produced mid-stream, the new behavior is likely correct; however, the PR description says 'pure refactoring with no behavioral change'. Please either restore the previous semantics or explicitly document/justify this behavior change (and ideally add coverage for empty-batch handling).
##########
cpp/core/jni/JniCommon.h:
##########
@@ -549,3 +550,86 @@ class JavaRssClient : public RssClient {
jmethodID javaPushPartitionData_;
jbyteArray array_;
};
+
+/// Bridges gluten::ThreadInitializer callbacks to a Java-side
+/// NativeThreadInitializer instance via JNI.
+///
+/// On initialize(), attaches the current native thread to the JVM and calls
+/// into the Java initializer so it can install Spark TaskContext or other
+/// thread-local state. On destroy(), calls the Java destroy() method but
+/// does NOT detach the thread — the underlying JVM thread may be reused by
+/// the pool, and detaching prematurely could allow the Java Thread object
+/// to be garbage-collected.
+class SparkThreadInitializer final : public gluten::ThreadInitializer {
+ public:
+ /// @param vm The JavaVM pointer from JNI_OnLoad.
+ /// @param jInitializerLocalRef A local reference to a Java object
+ /// implementing org.apache.gluten.threads.NativeThreadInitializer.
+ /// A global reference is created internally.
+ SparkThreadInitializer(JavaVM* vm, jobject jInitializerLocalRef) : vm_(vm) {
+ JNIEnv* env;
+ attachCurrentThreadAsDaemonOrThrow(vm_, &env);
+ jInitializerGlobalRef_ = env->NewGlobalRef(jInitializerLocalRef);
+ GLUTEN_CHECK(jInitializerGlobalRef_ != nullptr, "Failed to create global reference for native thread initializer.");
+ (void)initializeMethod(env);
+ }
+
+ SparkThreadInitializer(const SparkThreadInitializer&) = delete;
+ SparkThreadInitializer(SparkThreadInitializer&&) = delete;
+ SparkThreadInitializer& operator=(const SparkThreadInitializer&) = delete;
+ SparkThreadInitializer& operator=(SparkThreadInitializer&&) = delete;
+
+ ~SparkThreadInitializer() override {
+ JNIEnv* env;
+ if (vm_->GetEnv(reinterpret_cast<void**>(&env), jniVersion) != JNI_OK) {
+ LOG(WARNING) << "SparkThreadInitializer#~SparkThreadInitializer(): "
+ << "JNIEnv was not attached to current thread";
+ return;
+ }
+ env->DeleteGlobalRef(jInitializerGlobalRef_);
+ }
+
+ void initialize(const std::string& threadName) override {
+ JNIEnv* env;
+ attachCurrentThreadAsDaemonOrThrow(vm_, &env);
+ jstring jThreadName = env->NewStringUTF(threadName.c_str());
+ env->CallVoidMethod(jInitializerGlobalRef_, initializeMethod(env), jThreadName);
+ env->DeleteLocalRef(jThreadName);
+ checkException(env);
+ }
+
+ void destroy(const std::string& threadName) override {
+ // IMPORTANT: Do not call vm_.DetachCurrentThread here, otherwise Java side thread
+ // object might be dereferenced and garbage-collected, to break the reuse of thread
+ // resources.
+ JNIEnv* env;
+ attachCurrentThreadAsDaemonOrThrow(vm_, &env);
+ jstring jThreadName = env->NewStringUTF(threadName.c_str());
+ env->CallVoidMethod(jInitializerGlobalRef_, destroyMethod(env), jThreadName);
+ env->DeleteLocalRef(jThreadName);
+ checkException(env);
+ }
+
+ private:
+ jmethodID initializeMethod(JNIEnv* env) {
+ static jmethodID initializeMethod =
+ getMethodIdOrError(env, nativeThreadInitializerClass(env), "initialize", "(Ljava/lang/String;)V");
+ return initializeMethod;
+ }
+
+ jmethodID destroyMethod(JNIEnv* env) {
+ static jmethodID destroyMethod =
+ getMethodIdOrError(env, nativeThreadInitializerClass(env), "destroy", "(Ljava/lang/String;)V");
+ return destroyMethod;
+ }
+
+ jclass nativeThreadInitializerClass(JNIEnv* env) {
+ static jclass javaInitializerClass =
+ createGlobalClassReferenceOrError(env, "Lorg/apache/gluten/threads/NativeThreadInitializer;");
+ return javaInitializerClass;
+ }
Review Comment:
`FindClass`-style lookups typically expect a binary name like `org/apache/.../ClassName`, not a field descriptor like `Lorg/...;`. Using the descriptor here is likely to fail class resolution at runtime. Switch to `org/apache/gluten/threads/NativeThreadInitializer` (without `L` and `;`) if `createGlobalClassReferenceOrError` ultimately uses `JNIEnv::FindClass`.
##########
cpp/core/compute/Runtime.cc:
##########
@@ -40,9 +40,10 @@ void Runtime::registerFactory(const std::string& kind, Runtime::Factory factory,
Runtime* Runtime::create(
const std::string& kind,
MemoryManager* memoryManager,
+ ThreadManager* threadManager,
const std::unordered_map<std::string, std::string>& sessionConf) {
auto& factory = runtimeFactories().get(kind);
- return factory(kind, std::move(memoryManager), sessionConf);
+ return factory(kind, std::move(memoryManager), std::move(threadManager), sessionConf);
Review Comment:
`std::move` on raw pointers is misleading (it does not transfer ownership semantics beyond copying the value). Passing `memoryManager` and `threadManager` directly would be clearer and avoids implying unique ownership transfer.
##########
cpp/core/threads/ThreadInitializer.h:
##########
@@ -0,0 +1,54 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <memory>
+
+namespace gluten {
+
+/// Lifecycle hook invoked on each worker thread managed by a ThreadManager.
+///
+/// When a thread pool (e.g., folly::CPUThreadPoolExecutor) spawns or reaps a
+/// thread, the ThreadInitializer gives the application a chance to attach
+/// per-thread context — such as JNI thread attachment or Spark TaskContext
+/// propagation — before the thread runs native work and to clean up after.
Review Comment:
In this PR, `ThreadInitializer` is invoked from `HookedExecutor::wrap()` around each submitted task (not specifically on thread spawn/reap). Either adjust this documentation to match the actual call pattern (per-task execution) or rework the implementation to run on true thread lifecycle events (e.g., using a thread factory / thread hook) so the documented semantics remain accurate.
##########
cpp/velox/operators/plannodes/CudfVectorStream.h:
##########
@@ -134,6 +134,11 @@ class CudfValueStreamNode final : public facebook::velox::core::PlanNode {
std::shared_ptr<ResultIterator> iterator)
: facebook::velox::core::PlanNode(id), outputType_(outputType), iterator_(std::move(iterator)) {}
+ // Only supports single thread because iterator_ is not guranteed thread-safe.
Review Comment:
Correct typo: `guranteed` -> `guaranteed`.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "zhouyuan (via GitHub)" <gi...@apache.org>.
zhouyuan commented on code in PR #12302:
URL: https://github.com/apache/gluten/pull/12302#discussion_r3426230057
##########
cpp/core/compute/Runtime.h:
##########
@@ -189,6 +197,7 @@ class Runtime : public std::enable_shared_from_this<Runtime> {
protected:
std::string kind_;
MemoryManager* memoryManager_;
+ ThreadManager* threadManager_;
Review Comment:
was trying to understand the lifetime for this `threadManager_`: initially we run into some leakage from VeloxRuntime and we need to careful on destructor maybe we could change this to unique_ptr so it will be released when runtime destruct?
##########
backends-velox/src/test/java/org/apache/gluten/test/MockVeloxBackend.java:
##########
@@ -72,7 +73,8 @@ public Object ask(Object message) throws Exception {
private static SparkConf newSparkConf() {
final SparkConf conf = new SparkConf();
conf.set(GlutenConfig.SPARK_OFFHEAP_SIZE_KEY(), "1g");
- conf.set(VeloxConfig$.MODULE$.COLUMNAR_VELOX_CONNECTOR_IO_THREADS().key(), "0");
+ conf.set(GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR().key(), "1");
Review Comment:
it seems this value is set to -1 on scala side
https://github.com/apache/gluten/blob/3b08a23683cf18fe34c04a99a0609c315c4d8c08/gluten-core/src/main/scala/org/apache/gluten/config/GlutenCoreConfig.scala#L173
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer merged PR #12302:
URL: https://github.com/apache/gluten/pull/12302
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12302:
URL: https://github.com/apache/gluten/pull/12302#issuecomment-4716853881
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12302:
URL: https://github.com/apache/gluten/pull/12302#issuecomment-4716676415
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor and add ThreadManager infrastructure [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12302:
URL: https://github.com/apache/gluten/pull/12302#issuecomment-4709138543
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #12302:
URL: https://github.com/apache/gluten/pull/12302#discussion_r3426705496
##########
cpp/velox/compute/VeloxBackend.cc:
##########
@@ -186,11 +209,24 @@ void VeloxBackend::init(
}
#endif
+ const int32_t numTaskSlotsPerExecutor = [&]() {
+ if (!backendConf_->valueExists(kNumTaskSlotsPerExecutor)) {
+ LOG(WARNING) << kNumTaskSlotsPerExecutor << " is not set. Falling back to 1.";
+ return 1;
+ }
+ return backendConf_->get<int32_t>(kNumTaskSlotsPerExecutor).value();
+ }();
+ GLUTEN_CHECK(
+ numTaskSlotsPerExecutor >= 0,
+ kNumTaskSlotsPerExecutor + " was set to negative number " + std::to_string(numTaskSlotsPerExecutor) +
+ ", this should not happen.");
+
Review Comment:
`kNumTaskSlotsPerExecutor` can be present with GlutenCoreConfig’s default value `-1`, but the current logic treats “key exists” as “configured” and then `GLUTEN_CHECK`s that it is non-negative. This will abort backend init when the config is not explicitly set (or propagated) and remains `-1`. Consider treating negative values (especially `-1`) as “unset” and falling back to 1 (consistent with the missing-key branch).
##########
cpp/velox/compute/WholeStageResultIterator.cc:
##########
@@ -307,41 +307,24 @@ std::shared_ptr<velox::core::QueryCtx> WholeStageResultIterator::createNewVeloxQ
}
std::shared_ptr<ColumnarBatch> WholeStageResultIterator::next() {
- if (task_->isFinished()) {
- return nullptr;
- }
- velox::RowVectorPtr vector;
while (true) {
- auto future = velox::ContinueFuture::makeEmpty();
- auto out = task_->next(&future);
- if (!future.valid()) {
- // Not need to wait. Break.
- vector = std::move(out);
- break;
+ if (!cursor_->moveNext()) {
+ return nullptr;
}
- // Velox suggested to wait. This might be because another thread (e.g., background io thread) is spilling the task.
- GLUTEN_CHECK(out == nullptr, "Expected to wait but still got non-null output from Velox task");
- VLOG(2) << "Velox task " << task_->taskId()
- << " is busy when ::next() is called. Will wait and try again. Task state: "
- << taskStateString(task_->state());
- future.wait();
- }
- if (vector == nullptr) {
- return nullptr;
- }
- uint64_t numRows = vector->size();
- if (numRows == 0) {
- return nullptr;
- }
-
- {
- ScopedTimer timer(&loadLazyVectorTime_);
- for (auto& child : vector->children()) {
- child->loadedVector();
+ RowVectorPtr vector = cursor_->current();
+ GLUTEN_CHECK(vector != nullptr, "Cursor returned null vector.");
+ uint64_t numRows = vector->size();
+ if (numRows == 0) {
+ continue;
+ }
Review Comment:
`next()` now skips 0-row vectors (`continue`) instead of returning `nullptr` as before. This is a semantic change (it can turn a previously-terminating empty batch into continued iteration), which conflicts with the PR description claiming “pure refactoring with no behavioral change”. If this is intentional (e.g., TaskCursor can emit empty batches), it should be explicitly documented and ideally covered by a test; otherwise, consider restoring the prior behavior.
##########
gluten-arrow/src/main/java/org/apache/gluten/threads/NativeThreadManagerJniWrapper.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.gluten.threads;
+
+/**
+ * JNI bridge for native ThreadManager creation and release.
+ *
+ * <p>Each Spark task creates one native ThreadManager (via {@link #create}) that wraps a {@link
+ * NativeThreadInitializer}. The returned handle is passed to the native Runtime factory so it can
+ * install thread lifecycle callbacks on its worker executors.
+ */
+public class NativeThreadManagerJniWrapper {
+ private NativeThreadManagerJniWrapper() {}
+
+ /**
+ * Create a native ThreadManager for the given backend type.
+ *
+ * @param backendType the backend kind string (e.g., "velox").
+ * @param initializer the Java-side callback invoked on worker thread create/destroy.
+ * @return opaque native handle.
+ */
+ public static native long create(String backendType, NativeThreadInitializer initializer);
+
+ /**
+ * Release a native ThreadManager previously created with {@link #create}.
+ *
+ * @param handle the native handle returned by {@link #create}.
+ */
+ public static native void release(long handle);
+}
Review Comment:
This class-level Javadoc and the `initializer` parameter docs describe install of “thread lifecycle callbacks (create/destroy)”, but the callbacks are currently used as per-task hooks around executor submissions. Updating the wording here to match the actual invocation model will reduce confusion for future maintainers.
##########
cpp/velox/compute/WholeStageResultIterator.cc:
##########
@@ -391,22 +371,24 @@ void WholeStageResultIterator::getOrderedNodeIds(
return;
}
- if (isUnionNode) {
- // FIXME: The whole metrics system in gluten-substrait is magic. Passing metrics trees through JNI with a trivial
- // array is possible but requires for a solid design. Apparently we haven't had it. All the code requires complete
- // rework.
- // Union was interpreted as LocalPartition + LocalExchange + 2 fake projects as children in Velox. So we only fetch
- // metrics from the root node.
- std::vector<std::shared_ptr<const velox::core::PlanNode>> unionChildren{};
+ if (isLocalExchangeNode) {
+ // LocalPartition was interpreted as LocalPartition + LocalExchange + 2 fake projects (optional) as children
+ // in SubstraitToVeloxPlan. So we only fetch metrics from the root node.
for (const auto& source : planNode->sources()) {
const auto projectedChild = std::dynamic_pointer_cast<const velox::core::ProjectNode>(source);
- GLUTEN_CHECK(projectedChild != nullptr, "Illegal state");
- const auto projectSources = projectedChild->sources();
- GLUTEN_CHECK(projectSources.size() == 1, "Illegal state");
- const auto projectSource = projectSources.at(0);
- getOrderedNodeIds(projectSource, nodeIds);
+ if (projectedChild != nullptr) {
+ const auto projectSources = projectedChild->sources();
+ GLUTEN_CHECK(projectSources.size() == 1, "Illegal state");
+ const auto projectSource = projectSources.at(0);
+ getOrderedNodeIds(projectSource, nodeIds);
+ } else {
+ getOrderedNodeIds(source, nodeIds);
+ }
+ }
+ if (planNode->sources().size() == 2) {
+ // The LocalPartition maps to a concrete Spark native union transformer operator.
+ nodeIds.emplace_back(planNode->id());
}
- nodeIds.emplace_back(planNode->id());
return;
Review Comment:
`LocalPartitionNode` used for Substrait set-rel (union) can have an arbitrary number of inputs, but this branch only appends the LocalPartition node id when `sources().size() == 2`. For unions with 3+ inputs, this drops the root node from `orderedNodeIds_`, which can break metrics indexing/lookup later (e.g., `collectMetrics` expects stats for every id it records).
##########
gluten-arrow/src/main/java/org/apache/gluten/threads/NativeThreadInitializer.java:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.gluten.threads;
+
+/**
+ * Java-side callback invoked by native code when a managed worker thread is created or destroyed.
+ *
+ * <p>Implementations are responsible for installing per-thread context — such as {@link
+ * org.apache.spark.TaskContext} — so that native worker threads behave as proper child threads of
+ * the Spark task.
+ *
+ * <p>Implementations must be thread-safe; {@link #initialize} and {@link #destroy} may be called
+ * concurrently from different native threads.
+ */
+public interface NativeThreadInitializer {
+ /**
+ * Called when a native worker thread is about to start executing tasks.
+ *
+ * @param threadName a human-readable name identifying the thread.
+ */
+ void initialize(String threadName);
+
+ /**
+ * Called when a native worker thread is about to be returned to the pool or destroyed. Must not
+ * detach the JNI thread — the thread may be reused.
+ *
+ * @param threadName the same name passed to {@link #initialize}.
+ */
+ void destroy(String threadName);
+}
Review Comment:
The Javadoc describes these callbacks as thread create/destroy hooks, but in this PR they are invoked around each *submitted executor task* (see `HookedExecutor::wrap` calling `ThreadInitializer::initialize/destroy` per task). This mismatch is confusing and may lead to incorrect implementations (e.g., assuming one init per thread lifetime). Consider updating the docs and parameter naming to reflect task-scoped invocation on a worker thread.
##########
gluten-arrow/src/main/scala/org/apache/gluten/threads/NativeThreadManager.scala:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.gluten.threads
+
+import org.apache.gluten.exception.GlutenException
+
+import org.apache.spark.task.{TaskResource, TaskResources}
+
+import java.util.concurrent.atomic.AtomicBoolean
+
+/**
+ * Scala wrapper around a native ThreadManager handle.
+ *
+ * Created once per Spark task and registered as a [[TaskResource]] so it is automatically released
+ * when the task completes. The ThreadManager wraps a [[NativeThreadInitializer]] that propagates
+ * task context to native worker threads spawned by folly executors.
+ */
Review Comment:
Scaladoc says the wrapped `NativeThreadInitializer` is invoked on “native worker threads … created/destroyed”, but the implementation wires the initializer into `HookedExecutor`, which invokes it around each submitted task. Aligning this doc (and the `initializer` param doc below) with the per-task invocation model will avoid misuse.
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor and add ThreadManager infrastructure [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12302:
URL: https://github.com/apache/gluten/pull/12302#issuecomment-4709124531
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "zhztheplayer (via GitHub)" <gi...@apache.org>.
zhztheplayer commented on code in PR #12302:
URL: https://github.com/apache/gluten/pull/12302#discussion_r3426601754
##########
backends-velox/src/test/java/org/apache/gluten/test/MockVeloxBackend.java:
##########
@@ -72,7 +73,8 @@ public Object ask(Object message) throws Exception {
private static SparkConf newSparkConf() {
final SparkConf conf = new SparkConf();
conf.set(GlutenConfig.SPARK_OFFHEAP_SIZE_KEY(), "1g");
- conf.set(VeloxConfig$.MODULE$.COLUMNAR_VELOX_CONNECTOR_IO_THREADS().key(), "0");
+ conf.set(GlutenCoreConfig.NUM_TASK_SLOTS_PER_EXECUTOR().key(), "1");
Review Comment:
It's finally set here: https://github.com/apache/gluten/blob/3b08a23683cf18fe34c04a99a0609c315c4d8c08/gluten-core/src/main/scala/org/apache/gluten/GlutenPlugin.scala#L145
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org
Re: [PR] [VL] Use Velox TaskCursor infrastructure [gluten]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #12302:
URL: https://github.com/apache/gluten/pull/12302#issuecomment-4718217463
Run Gluten Clickhouse CI on x86
--
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: commits-unsubscribe@gluten.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@gluten.apache.org
For additional commands, e-mail: commits-help@gluten.apache.org