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: [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]
