fix: load the bundled native library only once per class loader

spark/src/main/java/org/apache/comet/NativeBase.java

@@ -49,6 +49,10 @@ public abstract class NativeBase {
 
   private static final String libraryToLoad = System.mapLibraryName(NATIVE_LIB_NAME);
   private static boolean loaded = false;
+  // The bundled libcomet this class loader unpacked and loaded, or null. Unlike `loaded`, never
+  // reset: unpacking it again yields a new temporary file, which the JVM loads as a second
+  // library with its own uninitialized native state, and JNI methods can then bind to either copy.
+  private static File bundledLibrary = null;
   private static volatile Throwable loadErr = null;
   private static final String searchPattern = "libcomet-";
   private static final AtomicBoolean released = new AtomicBoolean(false);
@@ -76,6 +80,11 @@ static synchronized void setLoaded(boolean b) {
     loaded = b;
   }
 
+  // Only for testing
+  static synchronized File bundledLibrary() {
+    return bundledLibrary;
+  }
+
   static synchronized void load() {
     if (loaded) {
       return;
@@ -113,6 +122,11 @@ static synchronized void load() {
    * Use the bundled native libraries. Functionally equivalent to <code>System.loadLibrary</code>.
    */
   private static void bundleLoadLibrary() {
+    if (bundledLibrary != null) {
+      loaded = true;
+      return;
+    }
+
     String resourceName = resourceName();
     InputStream is = NativeBase.class.getResourceAsStream(resourceName);
     if (is == null) {
@@ -133,6 +147,7 @@ private static void bundleLoadLibrary() {
       Files.copy(is, tempLib.toPath(), StandardCopyOption.REPLACE_EXISTING);
       System.load(tempLib.getAbsolutePath());
       loaded = true;
+      bundledLibrary = tempLib;
     } catch (IOException e) {
       throw new IllegalStateException("Cannot unpack libcomet: " + e);
     } finally {

spark/src/test/scala/org/apache/comet/CometSparkSessionExtensionsSuite.scala

@@ -114,6 +114,22 @@ class CometSparkSessionExtensionsSuite extends CometTestBase {
     }
   }
 
+  test("reloading NativeBase does not load a second copy of the native library") {
+    // The bundled library is unpacked to a new temporary file each time it is loaded. To the JVM
+    // a second copy is a distinct library with its own uninitialized native state, and a JNI
+    // method first called afterwards can bind to it (#6096).
+    val before = NativeBase.bundledLibrary()
+    try {
+      NativeBase.setLoaded(false)
+      NativeBase.load()
+      assert(NativeBase.isLoaded)
+    } finally {
+      NativeBase.setLoaded(true)
+    }
+    val after = NativeBase.bundledLibrary()
+    assert(after eq before, s"reload unpacked another copy of the native library: $after")
+  }
+
   test("Arrow properties") {
     NativeBase.setLoaded(false)
     NativeBase.load()