Add lifecycle cleanup for table function states

UDF.MD

@@ -218,6 +218,38 @@ Table function callback must implement `DuckDBTableFunction` interface:
  - `localInit` (optional) - called once for every thread before the function execution, returns an object used to track thread-local state during function execution
  - `apply` - has access to bind data, global init data and local init date, writes results into output vectors, returns a number of written row, is called by the engine repeatedly until it return zero rows
 
+States that own resources can opt into deterministic cleanup by implementing
+`DuckDBTableFunctionState`:
+
+```java
+final class SourceState implements DuckDBTableFunctionState {
+    private final AutoCloseable resource;
+
+    SourceState(AutoCloseable resource) {
+        this.resource = resource;
+    }
+
+    @Override
+    public void close() throws Exception {
+        resource.close();
+    }
+}
+```
+
+After a state is successfully installed as bind, global-init, or local-init
+data, DuckDB owns its lifecycle and calls `close()` when that state is
+destroyed. This can happen after normal exhaustion, a `LIMIT`, an error,
+cancellation, or early JDBC result/statement/connection closure. Only objects
+implementing `DuckDBTableFunctionState` opt in; an object implementing only
+`AutoCloseable`, and `null`, are left unchanged.
+
+Cleanup can run on native worker threads. Local states can be closed
+concurrently, and no global ordering is guaranteed between bind, global-init,
+and local states. Make `close()` fast, idempotent, and thread-safe when it
+manages shared resources. It must not execute SQL or re-enter the same
+connection. Exceptions from `close()` are handled on a best-effort basis and
+are not propagated to the query, so record any diagnostics inside the state.
+
 To enable parallel function execution it is necessary to set desired number of threads using `info.setMaxThreads()` method during the global init call.
 Depending on the query it also may be necessary to set `preserve_insertion_order=FALSE` option ([reference](https://duckdb.org/docs/current/configuration/overview#global-configuration-options)) in connection string or in connection properties object.
 
@@ -272,4 +304,3 @@ try (Connection conn = DriverManager.getConnection("jdbc:duckdb:"); Statement st
 ```sql
 FROM java_table_basic(42, param1='foobar')
 ```
-

src/jni/bindings_table_function_bind.cpp

@@ -132,7 +132,7 @@ JNIEXPORT void JNICALL Java_org_duckdb_DuckDBBindings_duckdb_1bind_1set_1bind_1d
 		return;
 	}
 
-	duckdb_bind_set_bind_data(bi, bind_data_holder.release(), GlobalRefHolder::destroy);
+	duckdb_bind_set_bind_data(bi, bind_data_holder.release(), GlobalRefHolder::destroy_table_function_state);
 }
 
 /*

src/jni/bindings_table_function_init.cpp

@@ -60,7 +60,7 @@ JNIEXPORT void JNICALL Java_org_duckdb_DuckDBBindings_duckdb_1init_1set_1init_1d
 		return;
 	}
 
-	duckdb_init_set_init_data(ii, init_data_holder.release(), GlobalRefHolder::destroy);
+	duckdb_init_set_init_data(ii, init_data_holder.release(), GlobalRefHolder::destroy_table_function_state);
 }
 
 /*

src/jni/holders.cpp

@@ -113,6 +113,57 @@ void GlobalRefHolder::destroy(void *holder_in) noexcept {
 	delete holder;
 }
 
+void GlobalRefHolder::destroy_table_function_state(void *holder_in) noexcept {
+	auto holder = reinterpret_cast<GlobalRefHolder *>(holder_in);
+	if (holder == nullptr) {
+		return;
+	}
+
+	try {
+		AttachedJNIEnv attached = holder->attach_current_thread();
+		if (attached.env != nullptr) {
+			try {
+				attached.env->CallStaticVoidMethod(J_DuckDBTableFunctionWrapper,
+				                                   J_DuckDBTableFunctionWrapper_closeTableFunctionState,
+				                                   holder->global_ref);
+			} catch (...) {
+				// Teardown must not allow a C++ exception to escape a native destructor.
+			}
+			try {
+				if (attached.env->ExceptionCheck()) {
+					attached.env->ExceptionClear();
+				}
+			} catch (...) {
+				// JNI cleanup is best effort; the holder is still deleted below.
+			}
+			// Keep the callback's attachment alive through reference deletion and holder destruction.
+			try {
+				if (holder->global_ref != nullptr) {
+					jobject global_ref = holder->global_ref;
+					holder->global_ref = nullptr;
+					attached.env->DeleteGlobalRef(global_ref);
+				}
+			} catch (...) {
+				// The reference was cleared before the JNI call, so the holder destructor cannot
+				// redundantly attempt another attach if the call fails unexpectedly.
+			}
+			delete holder;
+			return;
+		}
+
+		// attach_current_thread already attempted the only safe cleanup path. Without a valid
+		// JNIEnv, JNI does not permit DeleteGlobalRef; clear the reference so the holder destructor
+		// does not redundantly attempt another attach.
+		holder->global_ref = nullptr;
+	} catch (...) {
+		// A failed attach or unexpected JNI failure must not skip holder deletion or trigger a
+		// second attach from the holder destructor.
+		holder->global_ref = nullptr;
+	}
+
+	delete holder;
+}
+
 LocalRefHolder::LocalRefHolder(JNIEnv *env_in, jobject local_ref_in) : env(env_in), local_ref(local_ref_in) {
 }
 

src/jni/holders.hpp

@@ -80,6 +80,7 @@ struct GlobalRefHolder {
 	void detach_current_thread();
 
 	static void destroy(void *holder_in) noexcept;
+	static void destroy_table_function_state(void *holder_in) noexcept;
 };
 
 struct LocalRefHolder {

src/jni/refs.cpp

@@ -123,6 +123,7 @@ jmethodID J_DuckDBTableFunctionWrapper_executeBind;
 jmethodID J_DuckDBTableFunctionWrapper_executeGlobalInit;
 jmethodID J_DuckDBTableFunctionWrapper_executeLocalInit;
 jmethodID J_DuckDBTableFunctionWrapper_executeFunction;
+jmethodID J_DuckDBTableFunctionWrapper_closeTableFunctionState;
 
 static std::vector<jobject> global_refs;
 
@@ -318,6 +319,8 @@ void create_refs(JNIEnv *env) {
 	    get_method_id(env, J_DuckDBTableFunctionWrapper, "executeLocalInit", "(Ljava/nio/ByteBuffer;)V");
 	J_DuckDBTableFunctionWrapper_executeFunction = get_method_id(env, J_DuckDBTableFunctionWrapper, "executeFunction",
 	                                                             "(Ljava/nio/ByteBuffer;Ljava/nio/ByteBuffer;)V");
+	J_DuckDBTableFunctionWrapper_closeTableFunctionState =
+	    get_static_method_id(env, J_DuckDBTableFunctionWrapper, "closeTableFunctionState", "(Ljava/lang/Object;)V");
 }
 
 void delete_global_refs(JNIEnv *env) noexcept {

src/jni/refs.hpp

@@ -120,6 +120,7 @@ extern jmethodID J_DuckDBTableFunctionWrapper_executeBind;
 extern jmethodID J_DuckDBTableFunctionWrapper_executeGlobalInit;
 extern jmethodID J_DuckDBTableFunctionWrapper_executeLocalInit;
 extern jmethodID J_DuckDBTableFunctionWrapper_executeFunction;
+extern jmethodID J_DuckDBTableFunctionWrapper_closeTableFunctionState;
 
 void create_refs(JNIEnv *env);
 

src/main/java/org/duckdb/DuckDBTableFunctionState.java

@@ -0,0 +1,42 @@
+package org.duckdb;
+
+/**
+ * Opt-in lifecycle contract for state returned by a {@link DuckDBTableFunction}.
+ *
+ * <p>DuckDB takes ownership of an object implementing this interface after the
+ * object has been successfully installed as bind, global-init, or local-init
+ * data. DuckDB then makes a best-effort call to {@link #close()} when the
+ * corresponding native state is destroyed. A {@code null} state and an object
+ * that does not implement this interface are not closed by this lifecycle.
+ * Implementations must not rely on garbage collection for releasing their
+ * resources.</p>
+ *
+ * <p>Teardown may run on a native execution or worker thread rather than the
+ * thread that created the state. Local states can be created and closed
+ * concurrently with one another. There is no global ordering guarantee among
+ * bind, global-init, and local-init states, or among local states.</p>
+ *
+ * <p>A holder is closed at most once, but the same Java object may be installed
+ * in multiple holders and consequently have {@code close()} called once per
+ * holder. Implementations should therefore be idempotent and thread-safe when
+ * they manage shared resources or may be aliased. {@code close()} must not
+ * re-enter the same connection or execute SQL during teardown.</p>
+ *
+ * <p>Exceptions and other {@link Throwable}s from teardown are handled on a
+ * best-effort basis and are not propagated to the query caller. Implementations
+ * that need diagnostics must record them themselves. A close failure must not
+ * prevent other state cleanup or release of the native/JNI reference.</p>
+ */
+@SuppressWarnings("try")
+public interface DuckDBTableFunctionState extends AutoCloseable {
+    /**
+     * Releases resources owned by this table-function state.
+     *
+     * <p>This method may be called from a different thread, and its failure is
+     * not reported through the query.</p>
+     *
+     * @throws Exception if releasing the resource fails; the exception is
+     *                   handled as best effort by the table-function runtime
+     */
+    @Override void close() throws Exception;
+}

src/main/java/org/duckdb/DuckDBTableFunctionWrapper.java

@@ -13,6 +13,17 @@ class DuckDBTableFunctionWrapper {
         this.function = function;
     }
 
+    static void closeTableFunctionState(Object state) {
+        if (!(state instanceof DuckDBTableFunctionState)) {
+            return;
+        }
+        try {
+            ((DuckDBTableFunctionState) state).close();
+        } catch (Throwable ignored) {
+            // State cleanup is best effort because the query may already be finished.
+        }
+    }
+
     public void executeBind(ByteBuffer bindInfoRef) {
         try {
             DuckDBTableFunctionBindInfo bindInfo = new DuckDBTableFunctionBindInfo(bindInfoRef);