[GLUTEN-11895][VL] Fix SIGSEGV on IOThreadPool threads during HDFS scan

backends-velox/src/test/java/org/apache/gluten/jni/JniThreadDetachTest.java

@@ -0,0 +1,110 @@
+/*
+ * 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.jni;
+
+import org.apache.gluten.backendsapi.BackendsApiManager;
+import org.apache.gluten.runtime.Runtime;
+import org.apache.gluten.runtime.Runtimes;
+import org.apache.gluten.test.VeloxBackendTestBase;
+import org.apache.gluten.vectorized.ColumnarBatchInIterator;
+
+import org.apache.spark.sql.vectorized.ColumnarBatch;
+import org.apache.spark.task.TaskResources$;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Regression test for SIGSEGV in CPUThreadPool threads during HDFS scan.
+ *
+ * <p>Root cause: JniColumnarBatchIterator destructor called DetachCurrentThread(), which poisoned
+ * libhdfs.so's TLS-cached JNIEnv*. The next HDFS call on the same thread used the stale env,
+ * causing SIGSEGV in jni_NewStringUTF.
+ *
+ * <p>This test reproduces the exact crash: on a native std::thread (simulating CPUThreadPool), it
+ * saves the JNIEnv* (like libhdfs caches in TLS), destroys a real JniColumnarBatchIterator, then
+ * reuses the saved env for a JNI call. With the buggy code, this triggers SIGSEGV and the JVM
+ * crashes. With the fix, it works normally.
+ */
+public class JniThreadDetachTest extends VeloxBackendTestBase {
+
+  /**
+   * Native helper in JniTestHelper.cc. Spawns a std::thread and reproduces:
+   *
+   * <ol>
+   *   <li>Attach thread, save env (simulates libhdfs TLS cache)
+   *   <li>Create/destroy real JniColumnarBatchIterator (destructor under test)
+   *   <li>Reuse saved env for FindClass (simulates libhdfs's next HDFS call)
+   * </ol>
+   *
+   * With the fix: returns true. With the bug: SIGSEGV crashes the JVM at step 3.
+   */
+  private static native boolean nativeTestIteratorDestructorKeepsThreadAttached(
+      long runtimeHandle, Object jColumnarBatchItr);
+
+  @Test
+  public void testIteratorDestructorDoesNotDetachThread() {
+    AtomicBoolean result = new AtomicBoolean(false);
+    AtomicReference<Throwable> thrown = new AtomicReference<>(null);
+
+    TaskResources$.MODULE$.runUnsafe(
+        () -> {
+          try {
+            String backendName = BackendsApiManager.getBackendName();
+            Runtime runtime = Runtimes.contextInstance(backendName, "JniThreadDetachTest");
+            long runtimeHandle = runtime.getHandle();
+
+            Iterator<ColumnarBatch> emptyIter = Collections.emptyIterator();
+            ColumnarBatchInIterator batchItr = new ColumnarBatchInIterator(backendName, emptyIter);
+
+            boolean ok = nativeTestIteratorDestructorKeepsThreadAttached(runtimeHandle, batchItr);
+            result.set(ok);
+          } catch (Throwable t) {
+            thrown.set(t);
+          }
+          return null;
+        });
+
+    if (thrown.get() != null) {
+      Assert.fail(
+          "Test setup failed (exception in TaskResources scope): " + thrown.get().getMessage());
+    }
+    Assert.assertTrue(
+        "JNI call on native thread failed after JniColumnarBatchIterator destructor.",
+        result.get());
+  }
+
+  /**
+   * Native helper in JniTestHelper.cc. Creates a JniAwareThreadFactory, runs a task on it, destroys
+   * the executor, then verifies the thread was properly attached and JNI calls succeeded (no crash,
+   * executor destroyed cleanly).
+   *
+   * <p>With the fix: returns true. With a regression (e.g. detach removed): JavaThread objects
+   * accumulate silently — this test catches the crash case and documents the correct lifecycle.
+   */
+  private static native boolean nativeTestSpillThreadDetachesCleanly();
+
+  @Test
+  public void testSpillThreadDetachesCleanly() {
+    boolean ok = nativeTestSpillThreadDetachesCleanly();
+    Assert.assertTrue("Spill thread JNI lifecycle test failed.", ok);
+  }
+}

cpp/core/jni/JniCommon.cc

@@ -26,7 +26,7 @@ void gluten::JniCommonState::ensureInitialized(JNIEnv* env) {
   initialized_ = true;
 }
 
-void gluten::JniCommonState::assertInitialized() {
+void gluten::JniCommonState::assertInitialized() const {
   if (!initialized_) {
     throw gluten::GlutenException("Fatal: JniCommonState::Initialize(...) was not called before using the utility");
   }
@@ -95,7 +95,11 @@ gluten::JniColumnarBatchIterator::~JniColumnarBatchIterator() {
   attachCurrentThreadAsDaemonOrThrow(vm_, &env);
   env->DeleteGlobalRef(jColumnarBatchItr_);
   env->DeleteGlobalRef(serializedColumnarBatchIteratorClass_);
-  vm_->DetachCurrentThread();
+  // Do NOT call DetachCurrentThread() here.
+  // libhdfs.so caches JNIEnv* in thread-local storage after AttachCurrentThread.
+  // If we detach, libhdfs's TLS cache becomes stale — the next HDFS call via
+  // libhdfs returns the stale env, causing SIGSEGV in jni_NewStringUTF.
+  // Daemon-attached threads are safe to leave attached; they won't block JVM shutdown.
 }
 
 std::shared_ptr<gluten::ColumnarBatch> gluten::JniColumnarBatchIterator::next() {

cpp/core/jni/JniCommon.h

@@ -21,6 +21,9 @@
 #include <arrow/ipc/writer.h>
 #include <execinfo.h>
 #include <jni.h>
+#include <thread>
+
+#include <folly/executors/thread_factory/ThreadFactory.h>
 
 #include "compute/ProtobufUtils.h"
 #include "compute/Runtime.h"
@@ -156,12 +159,17 @@ class JniCommonState {
 
   void ensureInitialized(JNIEnv* env);
 
-  void assertInitialized();
+  void assertInitialized() const;
 
   void close();
 
   jmethodID runtimeAwareCtxHandle();
 
+  JavaVM* vm() const {
+    assertInitialized();
+    return vm_;
+  }
+
  private:
   void initialize(JNIEnv* env);
 
@@ -179,6 +187,56 @@ inline JniCommonState* getJniCommonState() {
   return &jniCommonState;
 }
 
+/// A folly::ThreadFactory for spill thread pools. Attaches each thread to the
+/// JVM as a daemon at creation and calls DetachCurrentThread inside the thread
+/// function body — after all work completes but before any pthread_key destructor
+/// fires — to prevent unbounded JavaThread accumulation.
+///
+/// INVARIANT: threads created by this factory must never call libhdfs. libhdfs
+/// registers hdfsThreadDestructor via pthread_key on first HDFS call; that
+/// destructor calls DetachCurrentThread at actual thread exit. Calling it
+/// earlier (inside the thread body) would invalidate libhdfs's cached JNIEnv*,
+/// causing SIGSEGV on the next HDFS call.
+///
+/// REQUIRES: JniCommonState::ensureInitialized() must have been called before
+/// constructing this factory (i.e. after JNI_OnLoad completes).
+class JniAwareThreadFactory : public folly::ThreadFactory {
+ public:
+  JniAwareThreadFactory() : vm_(getJniCommonState()->vm()) {}
+
+  std::thread newThread(folly::Func&& func) override {
+    return std::thread([vm = vm_, f = std::move(func)]() mutable {
+      JNIEnv* env = nullptr;
+      bool weAttached = (vm->GetEnv(reinterpret_cast<void**>(&env), jniVersion) == JNI_EDETACHED);
+      if (weAttached) {
+        if (vm->AttachCurrentThreadAsDaemon(reinterpret_cast<void**>(&env), nullptr) != JNI_OK) {
+          LOG(WARNING) << "JniAwareThreadFactory: failed to attach thread to JVM";
+          weAttached = false;
+        }
+      }
+      // RAII guard: ensures DetachCurrentThread is called even if f() throws.
+      struct DetachGuard {
+        JavaVM* vm;
+        bool active;
+        ~DetachGuard() {
+          if (active) {
+            vm->DetachCurrentThread();
+          }
+        }
+      } guard{vm, weAttached};
+      f();
+    });
+  }
+
+  const std::string& getNamePrefix() const override {
+    static const std::string kEmpty;
+    return kEmpty;
+  }
+
+ private:
+  JavaVM* vm_;
+};
+
 Runtime* getRuntime(JNIEnv* env, jobject runtimeAware);
 
 // Safe version of JNI {Get|Release}<PrimitiveType>ArrayElements routines.

cpp/core/jni/JniWrapper.cc

@@ -103,7 +103,11 @@ class JavaInputStreamAdaptor final : public arrow::io::InputStream {
     env->CallVoidMethod(jniIn_, jniByteInputStreamClose);
     checkException(env);
     env->DeleteGlobalRef(jniIn_);
-    vm_->DetachCurrentThread();
+    // Do NOT call DetachCurrentThread() here.
+    // libhdfs.so caches JNIEnv* in thread-local storage after AttachCurrentThread.
+    // If we detach, libhdfs's TLS cache becomes stale — the next HDFS call via
+    // libhdfs returns the stale env, causing SIGSEGV in jni_NewStringUTF.
+    // Daemon-attached threads are safe to leave attached; they won't block JVM shutdown.
     closed_ = true;
     return arrow::Status::OK();
   }

cpp/velox/CMakeLists.txt

@@ -157,6 +157,7 @@ set(VELOX_SRCS
     jni/JniFileSystem.cc
     jni/JniUdf.cc
     jni/VeloxJniWrapper.cc
+    jni/JniTestHelper.cc
     jni/JniHashTable.cc
     memory/BufferOutputStream.cc
     memory/VeloxColumnarBatch.cc

cpp/velox/compute/WholeStageResultIterator.cc

@@ -19,6 +19,7 @@
 #include "VeloxPlanConverter.h"
 #include "VeloxRuntime.h"
 #include "config/VeloxConfig.h"
+#include "jni/JniCommon.h"
 #include "utils/ConfigExtractor.h"
 #include "velox/connectors/hive/HiveConfig.h"
 #include "velox/connectors/hive/HiveConnectorSplit.h"
@@ -92,7 +93,14 @@ WholeStageResultIterator::WholeStageResultIterator(
   spillStrategy_ = veloxCfg_->get<std::string>(kSpillStrategy, kSpillStrategyDefaultValue);
   auto spillThreadNum = veloxCfg_->get<uint32_t>(kSpillThreadNum, kSpillThreadNumDefaultValue);
   if (spillThreadNum > 0) {
-    spillExecutor_ = std::make_shared<folly::CPUThreadPoolExecutor>(spillThreadNum);
+    // INVARIANT: spillExecutor_ threads must never call libhdfs.
+    // JniAwareThreadFactory calls DetachCurrentThread at thread exit (inside the
+    // thread fn body, before any pthread_key destructor). If libhdfs were used on
+    // these threads, hdfsThreadDestructor would fire afterward with a stale JNIEnv*,
+    // causing SIGSEGV. Spill always uses local or heap-over-local filesystem.
+    spillExecutor_ = std::make_shared<folly::CPUThreadPoolExecutor>(
+        spillThreadNum,
+        std::make_shared<gluten::JniAwareThreadFactory>());
   }
   getOrderedNodeIds(veloxPlan_, orderedNodeIds_);