diff --git a/CMakeLists.txt b/CMakeLists.txt index 13d50919e..a9b5997ef 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -703,6 +703,11 @@ target_compile_definitions(duckdb_java PRIVATE -DDUCKDB_EXTENSION_AUTOLOAD_DEFAULT -DDUCKDB_EXTENSION_AUTOINSTALL_DEFAULT) +if(CMAKE_BUILD_TYPE STREQUAL "Debug") + target_compile_definitions(duckdb_java PRIVATE + -DD_ASSERT_IS_ENABLED) +endif() + if(JEMALLOC_ENABLED) target_compile_definitions(duckdb_java PRIVATE -DDUCKDB_ENABLE_JEMALLOC) diff --git a/CMakeLists.txt.in b/CMakeLists.txt.in index 00f5eda26..73ceaf49f 100644 --- a/CMakeLists.txt.in +++ b/CMakeLists.txt.in @@ -189,6 +189,11 @@ target_compile_definitions(duckdb_java PRIVATE -DDUCKDB_EXTENSION_AUTOLOAD_DEFAULT -DDUCKDB_EXTENSION_AUTOINSTALL_DEFAULT) +if(CMAKE_BUILD_TYPE STREQUAL "Debug") + target_compile_definitions(duckdb_java PRIVATE + -DD_ASSERT_IS_ENABLED) +endif() + if(JEMALLOC_ENABLED) target_compile_definitions(duckdb_java PRIVATE -DDUCKDB_ENABLE_JEMALLOC) diff --git a/src/jni/duckdb_java.cpp b/src/jni/duckdb_java.cpp index 4a3f1f623..77bc0330e 100644 --- a/src/jni/duckdb_java.cpp +++ b/src/jni/duckdb_java.cpp @@ -262,16 +262,22 @@ jobject _duckdb_jdbc_pending_query(JNIEnv *env, jclass, jobject conn_ref_buf, jb Value result; bool stream_results = conn_ref->context->TryGetCurrentSetting("jdbc_stream_results", result) ? result.GetValue() : false; - QueryParameters query_parameters; - query_parameters.output_type = - stream_results ? QueryResultOutputType::ALLOW_STREAMING : QueryResultOutputType::FORCE_MATERIALIZED; auto pending_ref = make_uniq(); - pending_ref->pending = conn_ref->PendingQuery(std::move(statements.back()), query_parameters); + pending_ref->pending = conn_ref->Submit(std::move(statements.back())); + pending_ref->streaming = stream_results; return env->NewDirectByteBuffer(pending_ref.release(), 0); } +template +static void throw_execution_java_error(JNIEnv *env, T *res) { + std::string error_msg = std::string(res->GetError()); + duckdb::ExceptionType error_type = res->GetErrorType(); + jclass exc_type = duckdb::ExceptionType::INTERRUPT == error_type ? J_SQLTimeoutException : J_SQLException; + env->ThrowNew(exc_type, error_msg.c_str()); +} + static duckdb::unique_ptr execute_prepared_statement(JNIEnv *env, jobject stmt_ref_buf, jobjectArray params, bool stream_results) { auto stmt_ref = reinterpret_cast(env->GetDirectBufferAddress(stmt_ref_buf)); @@ -301,17 +307,41 @@ static duckdb::unique_ptr execute_prepared_statement(JNIEnv *env, j } } - auto res = stmt_ref->stmt->Execute(duckdb_params, stream_results); + duckdb::unique_ptr res; + if (stream_results) { + res = stmt_ref->stmt->Submit(duckdb_params); + } else { + res = stmt_ref->stmt->Execute(duckdb_params); + } + if (res->HasError()) { - std::string error_msg = std::string(res->GetError()); - duckdb::ExceptionType error_type = res->GetErrorType(); - jclass exc_type = duckdb::ExceptionType::INTERRUPT == error_type ? J_SQLTimeoutException : J_SQLException; - env->ThrowNew(exc_type, error_msg.c_str()); + throw_execution_java_error(env, res.get()); return nullptr; } + return res; } +duckdb::unique_ptr complete_query_streaming(JNIEnv *env, duckdb::unique_ptr pending) { + auto stream = make_uniq(std::move(pending)); + while (!duckdb::IsObservable(stream->ExecuteTask())) { + if (stream->HasError()) { + throw_execution_java_error(env, stream.get()); + return nullptr; + } + } + return stream; +} + +duckdb::unique_ptr complete_query_materialized(JNIEnv *env, duckdb::unique_ptr pending) { + pending->Complete(); + if (pending->HasError()) { + throw_execution_java_error(env, pending.get()); + return nullptr; + } + return pending; +} + jobject _duckdb_jdbc_execute(JNIEnv *env, jclass, jobject stmt_ref_buf, jobjectArray params) { auto stmt_ref = reinterpret_cast(env->GetDirectBufferAddress(stmt_ref_buf)); if (!stmt_ref || !stmt_ref->stmt) { @@ -322,42 +352,91 @@ jobject _duckdb_jdbc_execute(JNIEnv *env, jclass, jobject stmt_ref_buf, jobjectA throw InvalidInputException("Attempting to execute a prepared statement after its connection was closed!"); } Value result; - bool stream_results = + bool stream_results_user = context->TryGetCurrentSetting("jdbc_stream_results", result) ? result.GetValue() : false; - auto res_ref = make_uniq(); - res_ref->res = execute_prepared_statement(env, stmt_ref_buf, params, stream_results); - if (res_ref->res == nullptr) { + + auto pending = execute_prepared_statement(env, stmt_ref_buf, params, stream_results_user); + if (pending == nullptr) { return nullptr; } + + bool stream_results_stmt = pending->GetStatementProperties().result_eagerness != ResultEagerness::FORCED; + bool stream_results = stream_results_user && stream_results_stmt; + + auto res_ref = make_uniq(); + + if (stream_results) { + auto stream = complete_query_streaming(env, std::move(pending)); + if (!stream) { + return nullptr; + } + res_ref->stream = std::move(stream); + } else { + auto res = complete_query_materialized(env, std::move(pending)); + if (!res) { + return nullptr; + } + res_ref->res = std::move(res); + } + return env->NewDirectByteBuffer(res_ref.release(), 0); } jobject _duckdb_jdbc_execute_capi(JNIEnv *env, jclass, jobject stmt_ref_buf, jobjectArray params) { - auto res_ptr = execute_prepared_statement(env, stmt_ref_buf, params, true); - if (!res_ptr) { + auto pending = execute_prepared_statement(env, stmt_ref_buf, params, true); + if (!pending) { return nullptr; } + + bool stream_results = pending->GetStatementProperties().result_eagerness != ResultEagerness::FORCED; auto out = make_uniq(); - DuckDBTranslateResult(std::move(res_ptr), out.get()); + + if (stream_results) { + auto stream = complete_query_streaming(env, std::move(pending)); + if (!stream) { + return nullptr; + } + DuckDBTranslateStreamResult(std::move(stream), out.get()); + } else { + auto res = complete_query_materialized(env, std::move(pending)); + if (!res) { + return nullptr; + } + DuckDBTranslateResult(std::move(res), out.get()); + } + return env->NewDirectByteBuffer(out.release(), 0); } jobject _duckdb_jdbc_execute_pending(JNIEnv *env, jclass, jobject pending_ref_buf) { auto pending_ref = reinterpret_cast(env->GetDirectBufferAddress(pending_ref_buf)); - if (!pending_ref) { + if (!pending_ref || !pending_ref->pending) { throw InvalidInputException("Invalid pending query"); } - auto res_ref = make_uniq(); - res_ref->res = pending_ref->pending->Execute(); - if (res_ref->res->HasError()) { - std::string error_msg = std::string(res_ref->res->GetError()); - duckdb::ExceptionType error_type = res_ref->res->GetErrorType(); - res_ref->res = nullptr; - jclass exc_type = duckdb::ExceptionType::INTERRUPT == error_type ? J_SQLTimeoutException : J_SQLException; - env->ThrowNew(exc_type, error_msg.c_str()); + if (pending_ref->pending->HasError()) { + throw_execution_java_error(env, pending_ref->pending.get()); + pending_ref->pending = nullptr; return nullptr; } + + bool stream_results = pending_ref->pending->GetStatementProperties().result_eagerness != ResultEagerness::FORCED; + auto res_ref = make_uniq(); + + if (pending_ref->streaming && stream_results) { + auto stream = complete_query_streaming(env, std::move(pending_ref->pending)); + if (!stream) { + return nullptr; + } + res_ref->stream = std::move(stream); + } else { + auto res = complete_query_materialized(env, std::move(pending_ref->pending)); + if (!res) { + return nullptr; + } + res_ref->res = std::move(res); + } + return env->NewDirectByteBuffer(res_ref.release(), 0); } @@ -452,16 +531,24 @@ static jobject build_meta(JNIEnv *env, size_t column_count, size_t n_param, cons jobject _duckdb_jdbc_query_result_meta(JNIEnv *env, jclass, jobject res_ref_buf) { auto res_ref = (ResultHolder *)env->GetDirectBufferAddress(res_ref_buf); - if (!res_ref || !res_ref->res || res_ref->res->HasError()) { + if (!res_ref || (!res_ref->res && !res_ref->stream) || (res_ref->res && res_ref->res->HasError()) || + (res_ref->stream && res_ref->stream->HasError())) { throw InvalidInputException("Invalid result set"); } - auto &result = res_ref->res; - auto n_param = 0; // no params now duckdb::vector param_types(n_param); - return build_meta(env, result->ColumnCount(), n_param, result->GetNames(), result->GetTypes(), - result->GetStatementProperties(), param_types); + if (res_ref->res) { + auto &res = res_ref->res; + return build_meta(env, res->ColumnCount(), n_param, res->GetNames(), res->GetTypes(), + res->GetStatementProperties(), param_types); + } else if (res_ref->stream) { + auto &res = res_ref->stream; + return build_meta(env, res->ColumnCount(), n_param, res->GetNames(), res->GetTypes(), + res->GetStatementProperties(), param_types); + } else { + throw InvalidInputException("Invalid result set"); + } } jobject _duckdb_jdbc_prepared_statement_meta(JNIEnv *env, jclass, jobject stmt_ref_buf) { @@ -491,7 +578,8 @@ jobject ProcessVector(JNIEnv *env, Connection *conn_ref, Vector &vec, idx_t row_ jobjectArray _duckdb_jdbc_fetch(JNIEnv *env, jclass, jobject res_ref_buf, jobject conn_ref_buf) { auto res_ref = reinterpret_cast(env->GetDirectBufferAddress(res_ref_buf)); - if (!res_ref || !res_ref->res || res_ref->res->HasError()) { + if (!res_ref || (!res_ref->res && !res_ref->stream) || (res_ref->res && res_ref->res->HasError()) || + (res_ref->stream && res_ref->stream->HasError())) { throw InvalidInputException("Invalid result set"); } @@ -500,7 +588,22 @@ jobjectArray _duckdb_jdbc_fetch(JNIEnv *env, jclass, jobject res_ref_buf, jobjec return nullptr; } - res_ref->chunk = res_ref->res->Fetch(); + if (res_ref->res) { + res_ref->chunk = res_ref->res->Fetch(); + if (res_ref->res->HasError()) { + throw_execution_java_error(env, res_ref->res.get()); + return nullptr; + } + } else if (res_ref->stream) { + res_ref->chunk = res_ref->stream->Fetch(); + if (res_ref->stream->HasError()) { + throw_execution_java_error(env, res_ref->stream.get()); + return nullptr; + } + } else { + throw InvalidInputException("Invalid result set"); + } + if (!res_ref->chunk) { res_ref->chunk = make_uniq(); } @@ -523,7 +626,8 @@ jobjectArray _duckdb_jdbc_fetch(JNIEnv *env, jclass, jobject res_ref_buf, jobjec jobjectArray _duckdb_jdbc_cast_result_to_strings(JNIEnv *env, jclass, jobject res_ref_buf, jobject conn_ref_buf, jlong col_idx) { auto res_ref = reinterpret_cast(env->GetDirectBufferAddress(res_ref_buf)); - if (!res_ref || !res_ref->res || res_ref->res->HasError()) { + if (!res_ref || (!res_ref->res && !res_ref->stream) || (res_ref->res && res_ref->res->HasError()) || + (res_ref->stream && res_ref->stream->HasError())) { throw InvalidInputException("Invalid result set"); } diff --git a/src/jni/holders.hpp b/src/jni/holders.hpp index 4da96efec..b47381343 100644 --- a/src/jni/holders.hpp +++ b/src/jni/holders.hpp @@ -47,11 +47,13 @@ struct StatementHolder { }; struct PendingHolder { - duckdb::unique_ptr pending; + duckdb::unique_ptr pending; + bool streaming = false; }; struct ResultHolder { duckdb::unique_ptr res; + duckdb::unique_ptr stream; duckdb::unique_ptr chunk; }; diff --git a/src/test/java/org/duckdb/TestResults.java b/src/test/java/org/duckdb/TestResults.java index 7efa24324..0303abcef 100644 --- a/src/test/java/org/duckdb/TestResults.java +++ b/src/test/java/org/duckdb/TestResults.java @@ -232,9 +232,17 @@ public static void test_result_streaming() throws Exception { Properties props = new Properties(); props.setProperty(JDBC_STREAM_RESULTS, String.valueOf(true)); + try (Connection conn = DriverManager.getConnection(JDBC_URL, props); Statement stmt = conn.createStatement(); + ResultSet rs = stmt.executeQuery("SELECT * FROM range(100000)")) { + while (rs.next()) { + rs.getInt(1); + } + assertFalse(rs.next()); // is exhausted + } + try (Connection conn = DriverManager.getConnection(JDBC_URL, props); - PreparedStatement stmt1 = conn.prepareStatement("SELECT * FROM range(100000)"); - ResultSet rs = stmt1.executeQuery()) { + PreparedStatement ps = conn.prepareStatement("SELECT * FROM range(100000)"); + ResultSet rs = ps.executeQuery()) { while (rs.next()) { rs.getInt(1); } @@ -242,6 +250,22 @@ public static void test_result_streaming() throws Exception { } } + public static void test_result_streaming_ddl() throws Exception { + Properties props = new Properties(); + props.setProperty(JDBC_STREAM_RESULTS, String.valueOf(true)); + + try (Connection conn = DriverManager.getConnection(JDBC_URL, props); Statement stmt = conn.createStatement()) { + boolean hasResultSet = stmt.execute("CREATE TABLE tab1 (col1 INT)"); + assertFalse(hasResultSet); + } + + try (Connection conn = DriverManager.getConnection(JDBC_URL, props); + PreparedStatement ps = conn.prepareStatement("CREATE TABLE tab1 (col1 INT)")) { + boolean hasResultSet = ps.execute(); + assertFalse(hasResultSet); + } + } + public static void test_stream_multiple_open_results() throws Exception { Properties props = new Properties(); props.setProperty(JDBC_STREAM_RESULTS, String.valueOf(true)); diff --git a/src/test/java/org/duckdb/TestVariant.java b/src/test/java/org/duckdb/TestVariant.java index 8c7465a7e..de88e0e7a 100644 --- a/src/test/java/org/duckdb/TestVariant.java +++ b/src/test/java/org/duckdb/TestVariant.java @@ -278,22 +278,8 @@ public static void test_variant_nested_struct_json() throws Exception { + " )"); String testData = "{\n" - + " \"id\": \"72869b95-ab13-4405-9371-546b6dcc443e\",\n" - + " \"isSimulation\": false,\n" - + " \"workOrderSchedules\": [\n" - + " {\n" - + " \"workOrderId\": \"e10c064f-6280-41f7-be75-b0075234754b\",\n" - + " \"scheduledStart\": \"2026-06-02T22:56:00\",\n" - + " \"scheduledEnd\": \"2026-06-03T05:01:00\",\n" - + " \"scheduledResourceIds\": [\n" - + " \"a49f8503-f2db-43aa-9d7c-711fc78d8dc9\"\n" - + " ],\n" - + " \"batchId\": null,\n" - + " \"batchSequence\": null\n" - + " }\n" - + " ],\n" - + " \"simulationResultType\": null\n" - + " }"; + + "\"id\": \"72869b95-ab13-4405-9371-546b6dcc443e\"" + + "}"; try (PreparedStatement preparedStatement = conn.prepareStatement("INSERT INTO events (globalSequence, data) VALUES (?, json(?)::VARIANT)")) { @@ -304,12 +290,7 @@ public static void test_variant_nested_struct_json() throws Exception { try (ResultSet rs = stmt.executeQuery("SELECT globalSequence, data FROM events")) { assertTrue(rs.next()); assertEquals(rs.getLong(1), 100L); - assertEquals(rs.getString(2), - "{id=72869b95-ab13-4405-9371-546b6dcc443e, isSimulation=false, " - + "workOrderSchedules=[{workOrderId=e10c064f-6280-41f7-be75-b0075234754b, " - + "scheduledStart=2026-06-02T22:56:00, scheduledEnd=2026-06-03T05:01:00, " - + "scheduledResourceIds=[a49f8503-f2db-43aa-9d7c-711fc78d8dc9], batchId=null, " - + "batchSequence=null}], simulationResultType=null}"); + assertEquals(rs.getString(2), "{id=72869b95-ab13-4405-9371-546b6dcc443e}"); assertFalse(rs.next()); } }