Close pending statements on connection close

This is a version 3 of this PR that attempts to safely close pending
statements when the connection is closed. In it all tracking and
locking logic is moved from C++ to Java:

 - `Connection`, `Statement` and `ResultSet` instances use their own
 locks; the check that corresponding reference is still alive is
 performed before every native call after obtaining the lock.
 - `Statement` lock is held during the query execution, note, this lock
 is NOT requied to call `statement#cancel()` because this operation is
 implemented on a `Connection` level.
 - When a `Connection` is being closed, pending query is cancelled
 first, and then all active statements are closed in a reverse creation
 order.

Note, thread safety for Appender and Arrow interfaces is going to be
addressed in subsequent PRs.

Testing: new tests added for various sequential and concurrent closure
scenarios.

Fixes: #101

Author: staticlibs

CMakeLists.txt

@@ -439,7 +439,7 @@ set(DUCKDB_SRC_FILES
   src/duckdb/extension/json/json_serializer.cpp
   src/duckdb/ub_extension_json_json_functions.cpp)
 
-set(JEMACLLOC_SRC_FILES 
+set(JEMALLOC_SRC_FILES 
   src/duckdb/extension/jemalloc/jemalloc_extension.cpp
   src/duckdb/extension/jemalloc/jemalloc/src/jemalloc.c
   src/duckdb/extension/jemalloc/jemalloc/src/arena.c
@@ -553,7 +553,7 @@ add_jar(duckdb_jdbc_tests ${JAVA_TEST_FILES} INCLUDE_JARS duckdb_jdbc)
 if(MSVC)
   list(APPEND DUCKDB_SRC_FILES duckdb_java.def)
 else()
-  list(APPEND DUCKDB_SRC_FILES ${JEMACLLOC_SRC_FILES})
+  list(APPEND DUCKDB_SRC_FILES ${JEMALLOC_SRC_FILES})
 endif()
 
 add_library(duckdb_java SHARED

CMakeLists.txt.in

@@ -46,7 +46,7 @@ set(DUCKDB_DEFINITIONS
 set(DUCKDB_SRC_FILES 
   ${SOURCES})
 
-set(JEMACLLOC_SRC_FILES 
+set(JEMALLOC_SRC_FILES 
   ${JEMALLOC_SOURCES})
 
 
@@ -95,7 +95,7 @@ add_jar(duckdb_jdbc_tests ${JAVA_TEST_FILES} INCLUDE_JARS duckdb_jdbc)
 if(MSVC)
   list(APPEND DUCKDB_SRC_FILES duckdb_java.def)
 else()
-  list(APPEND DUCKDB_SRC_FILES ${JEMACLLOC_SRC_FILES})
+  list(APPEND DUCKDB_SRC_FILES ${JEMALLOC_SRC_FILES})
 endif()
 
 add_library(duckdb_java SHARED

src/jni/duckdb_java.cpp

@@ -13,6 +13,7 @@
 #include "duckdb/main/extension_util.hpp"
 #include "duckdb/parser/parsed_data/create_type_info.hpp"
 #include "functions.hpp"
+#include "holders.hpp"
 #include "refs.hpp"
 #include "types.hpp"
 #include "util.hpp"
@@ -59,40 +60,6 @@ JNIEXPORT void JNICALL JNI_OnUnload(JavaVM *vm, void *reserved) {
 	delete_global_refs(env);
 }
 
-/**
- * Associates a duckdb::Connection with a duckdb::DuckDB. The DB may be shared amongst many ConnectionHolders, but the
- * Connection is unique to this holder. Every Java DuckDBConnection has exactly 1 of these holders, and they are never
- * shared. The holder is freed when the DuckDBConnection is closed. When the last holder sharing a DuckDB is freed, the
- * DuckDB is released as well.
- */
-struct ConnectionHolder {
-	const duckdb::shared_ptr<duckdb::DuckDB> db;
-	const duckdb::unique_ptr<duckdb::Connection> connection;
-
-	ConnectionHolder(duckdb::shared_ptr<duckdb::DuckDB> _db)
-	    : db(_db), connection(make_uniq<duckdb::Connection>(*_db)) {
-	}
-};
-
-/**
- * Throws a SQLException and returns nullptr if a valid Connection can't be retrieved from the buffer.
- */
-static Connection *get_connection(JNIEnv *env, jobject conn_ref_buf) {
-	if (!conn_ref_buf) {
-		throw ConnectionException("Invalid connection");
-	}
-	auto conn_holder = (ConnectionHolder *)env->GetDirectBufferAddress(conn_ref_buf);
-	if (!conn_holder) {
-		throw ConnectionException("Invalid connection");
-	}
-	auto conn_ref = conn_holder->connection.get();
-	if (!conn_ref || !conn_ref->context) {
-		throw ConnectionException("Invalid connection");
-	}
-
-	return conn_ref;
-}
-
 //! The database instance cache, used so that multiple connections to the same file point to the same database object
 duckdb::DBInstanceCache instance_cache;
 
@@ -189,10 +156,6 @@ void _duckdb_jdbc_disconnect(JNIEnv *env, jclass, jobject conn_ref_buf) {
 	}
 }
 
-struct StatementHolder {
-	duckdb::unique_ptr<PreparedStatement> stmt;
-};
-
 #include "utf8proc_wrapper.hpp"
 
 jobject _duckdb_jdbc_prepare(JNIEnv *env, jclass, jobject conn_ref_buf, jbyteArray query_j) {
@@ -233,11 +196,6 @@ jobject _duckdb_jdbc_prepare(JNIEnv *env, jclass, jobject conn_ref_buf, jbyteArr
 	return env->NewDirectByteBuffer(stmt_ref, 0);
 }
 
-struct ResultHolder {
-	duckdb::unique_ptr<QueryResult> res;
-	duckdb::unique_ptr<DataChunk> chunk;
-};
-
 Value ToValue(JNIEnv *env, jobject param, duckdb::shared_ptr<ClientContext> context) {
 	param = env->CallStaticObjectMethod(J_Timestamp, J_Timestamp_valueOf, param);
 
@@ -930,6 +888,7 @@ static ProfilerPrintFormat GetProfilerPrintFormat(JNIEnv *env, jobject format) {
 	if (env->IsSameObject(format, J_ProfilerPrintFormat_GRAPHVIZ)) {
 		return ProfilerPrintFormat::GRAPHVIZ;
 	}
+	throw InvalidInputException("Invalid profiling format");
 }
 
 jstring _duckdb_jdbc_get_profiling_information(JNIEnv *env, jclass, jobject conn_ref_buf, jobject j_format) {

src/jni/functions.cpp

@@ -404,5 +404,6 @@ JNIEXPORT jstring JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1get_1profil
 		duckdb::ErrorData error(e);
 		ThrowJNI(env, error.Message().c_str());
 
+		return nullptr;
 	}
 }

src/jni/holders.hpp

@@ -0,0 +1,48 @@
+#pragma once
+
+#include "duckdb.hpp"
+
+#include <jni.h>
+
+/**
+ * Associates a duckdb::Connection with a duckdb::DuckDB. The DB may be shared amongst many ConnectionHolders, but the
+ * Connection is unique to this holder. Every Java DuckDBConnection has exactly 1 of these holders, and they are never
+ * shared. The holder is freed when the DuckDBConnection is closed. When the last holder sharing a DuckDB is freed, the
+ * DuckDB is released as well.
+ */
+struct ConnectionHolder {
+	const duckdb::shared_ptr<duckdb::DuckDB> db;
+	const duckdb::unique_ptr<duckdb::Connection> connection;
+
+	ConnectionHolder(duckdb::shared_ptr<duckdb::DuckDB> _db)
+	    : db(_db), connection(duckdb::make_uniq<duckdb::Connection>(*_db)) {
+	}
+};
+
+struct StatementHolder {
+	duckdb::unique_ptr<duckdb::PreparedStatement> stmt;
+};
+
+struct ResultHolder {
+	duckdb::unique_ptr<duckdb::QueryResult> res;
+	duckdb::unique_ptr<duckdb::DataChunk> chunk;
+};
+
+/**
+ * Throws a SQLException and returns nullptr if a valid Connection can't be retrieved from the buffer.
+ */
+inline duckdb::Connection *get_connection(JNIEnv *env, jobject conn_ref_buf) {
+	if (!conn_ref_buf) {
+		throw duckdb::ConnectionException("Invalid connection");
+	}
+	auto conn_holder = (ConnectionHolder *)env->GetDirectBufferAddress(conn_ref_buf);
+	if (!conn_holder) {
+		throw duckdb::ConnectionException("Invalid connection");
+	}
+	auto conn_ref = conn_holder->connection.get();
+	if (!conn_ref || !conn_ref->context) {
+		throw duckdb::ConnectionException("Invalid connection");
+	}
+
+	return conn_ref;
+}

src/main/java/org/duckdb/DuckDBAppender.java

@@ -16,7 +16,7 @@ public DuckDBAppender(DuckDBConnection con, String schemaName, String tableName)
             throw new SQLException("Invalid connection");
         }
         appender_ref = DuckDBNative.duckdb_jdbc_create_appender(
-            con.conn_ref, schemaName.getBytes(StandardCharsets.UTF_8), tableName.getBytes(StandardCharsets.UTF_8));
+            con.connRef, schemaName.getBytes(StandardCharsets.UTF_8), tableName.getBytes(StandardCharsets.UTF_8));
     }
 
     public void beginRow() throws SQLException {