diff --git a/duckdb_java.def b/duckdb_java.def index 76ab859a1..da3cca726 100644 --- a/duckdb_java.def +++ b/duckdb_java.def @@ -20,8 +20,6 @@ Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1appender_1begin_1row Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1appender_1close Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1appender_1end_1row Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1appender_1flush -Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1register -Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1stream Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1connect Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1create_1appender Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1create_1db_1ref diff --git a/duckdb_java.exp b/duckdb_java.exp index 97f85ce7a..233681ecd 100644 --- a/duckdb_java.exp +++ b/duckdb_java.exp @@ -17,8 +17,6 @@ _Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1appender_1begin_1row _Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1appender_1close _Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1appender_1end_1row _Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1appender_1flush -_Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1register -_Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1stream _Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1connect _Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1create_1appender _Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1create_1db_1ref diff --git a/duckdb_java.map b/duckdb_java.map index df7bf67f6..55b47937a 100644 --- a/duckdb_java.map +++ b/duckdb_java.map @@ -19,8 +19,6 @@ DUCKDB_JAVA { Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1appender_1close; Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1appender_1end_1row; Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1appender_1flush; - Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1register; - Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1stream; Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1connect; Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1create_1appender; Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1create_1db_1ref; diff --git a/src/jni/duckdb_java.cpp b/src/jni/duckdb_java.cpp index 77bc0330e..2c2765006 100644 --- a/src/jni/duckdb_java.cpp +++ b/src/jni/duckdb_java.cpp @@ -4,7 +4,7 @@ extern "C" { #include "config.hpp" #include "duckdb.hpp" #include "duckdb/catalog/catalog_search_path.hpp" -#include "duckdb/common/arrow/result_arrow_wrapper.hpp" +#include "duckdb/common/arrow/arrow_format.hpp" #include "duckdb/common/operator/cast_operators.hpp" #include "duckdb/common/shared_ptr.hpp" #include "duckdb/common/vector/array_vector.hpp" @@ -322,8 +322,8 @@ static duckdb::unique_ptr execute_prepared_statement(JNIEnv *env, j return res; } -duckdb::unique_ptr complete_query_streaming(JNIEnv *env, duckdb::unique_ptr pending) { - auto stream = make_uniq(std::move(pending)); +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()); @@ -1041,69 +1041,71 @@ void _duckdb_jdbc_appender_append_null(JNIEnv *env, jclass, jobject appender_ref get_appender(env, appender_ref_buf)->Append(nullptr); } +/* TODO: ArrowFormat jlong _duckdb_jdbc_arrow_stream(JNIEnv *env, jclass, jobject res_ref_buf, jlong batch_size) { - if (!res_ref_buf) { - throw InvalidInputException("Invalid result set"); - } - auto res_ref = (ResultHolder *)env->GetDirectBufferAddress(res_ref_buf); - if (!res_ref || !res_ref->res || res_ref->res->HasError()) { - throw InvalidInputException("Invalid result set"); - } - - auto wrapper = new ResultArrowArrayStreamWrapper(std::move(res_ref->res), batch_size); - return (jlong)&wrapper->stream; + if (!res_ref_buf) { + throw InvalidInputException("Invalid result set"); + } + auto res_ref = (ResultHolder *)env->GetDirectBufferAddress(res_ref_buf); + if (!res_ref || !res_ref->res || res_ref->res->HasError()) { + throw InvalidInputException("Invalid result set"); + } + + auto wrapper = new ResultArrowArrayStreamWrapper(std::move(res_ref->res), batch_size); + return (jlong)&wrapper->stream; } class JavaArrowTabularStreamFactory { public: - JavaArrowTabularStreamFactory(ArrowArrayStream *stream_ptr_p) : stream_ptr(stream_ptr_p) {}; - - static duckdb::unique_ptr Produce(uintptr_t factory_p, ArrowStreamParameters ¶meters) { - - auto factory = (JavaArrowTabularStreamFactory *)factory_p; - if (!factory->stream_ptr->release) { - throw InvalidInputException("This stream has been released"); - } - auto res = make_uniq(); - res->arrow_array_stream = *factory->stream_ptr; - factory->stream_ptr->release = nullptr; - return res; - } - - static void GetSchema(uintptr_t factory_p, ArrowSchemaWrapper &schema) { - auto factory = (JavaArrowTabularStreamFactory *)factory_p; - auto stream_ptr = factory->stream_ptr; - if (!stream_ptr->release) { - throw InvalidInputException("This stream has been released"); - } - stream_ptr->get_schema(stream_ptr, &schema.arrow_schema); - auto error = stream_ptr->get_last_error(stream_ptr); - if (error != nullptr) { - throw InvalidInputException(error); - } - } - - ArrowArrayStream *stream_ptr; + JavaArrowTabularStreamFactory(ArrowArrayStream *stream_ptr_p) : stream_ptr(stream_ptr_p) {}; + + static duckdb::unique_ptr Produce(uintptr_t factory_p, ArrowStreamParameters ¶meters) { + + auto factory = (JavaArrowTabularStreamFactory *)factory_p; + if (!factory->stream_ptr->release) { + throw InvalidInputException("This stream has been released"); + } + auto res = make_uniq(); + res->arrow_array_stream = *factory->stream_ptr; + factory->stream_ptr->release = nullptr; + return res; + } + + static void GetSchema(uintptr_t factory_p, ArrowSchemaWrapper &schema) { + auto factory = (JavaArrowTabularStreamFactory *)factory_p; + auto stream_ptr = factory->stream_ptr; + if (!stream_ptr->release) { + throw InvalidInputException("This stream has been released"); + } + stream_ptr->get_schema(stream_ptr, &schema.arrow_schema); + auto error = stream_ptr->get_last_error(stream_ptr); + if (error != nullptr) { + throw InvalidInputException(error); + } + } + + ArrowArrayStream *stream_ptr; }; void _duckdb_jdbc_arrow_register(JNIEnv *env, jclass, jobject conn_ref_buf, jlong arrow_array_stream_pointer, jbyteArray name_j) { - auto conn = get_connection(env, conn_ref_buf); - if (conn == nullptr) { - return; - } - auto name = jbyteArray_to_string(env, name_j); + auto conn = get_connection(env, conn_ref_buf); + if (conn == nullptr) { + return; + } + auto name = jbyteArray_to_string(env, name_j); - auto arrow_array_stream = (ArrowArrayStream *)(uintptr_t)arrow_array_stream_pointer; + auto arrow_array_stream = (ArrowArrayStream *)(uintptr_t)arrow_array_stream_pointer; - auto factory = new JavaArrowTabularStreamFactory(arrow_array_stream); - duckdb::vector parameters; - parameters.push_back(Value::POINTER((uintptr_t)factory)); - parameters.push_back(Value::POINTER((uintptr_t)JavaArrowTabularStreamFactory::Produce)); - parameters.push_back(Value::POINTER((uintptr_t)JavaArrowTabularStreamFactory::GetSchema)); - conn->TableFunction("arrow_scan_dumb", parameters)->CreateView(Identifier(name), true, true); + auto factory = new JavaArrowTabularStreamFactory(arrow_array_stream); + duckdb::vector parameters; + parameters.push_back(Value::POINTER((uintptr_t)factory)); + parameters.push_back(Value::POINTER((uintptr_t)JavaArrowTabularStreamFactory::Produce)); + parameters.push_back(Value::POINTER((uintptr_t)JavaArrowTabularStreamFactory::GetSchema)); + conn->TableFunction("arrow_scan_dumb", parameters)->CreateView(Identifier(name), true, true); } +*/ static ProfilerPrintFormat GetProfilerPrintFormat(JNIEnv *env, jobject format) { jobject jname = env->CallObjectMethod(format, J_ProfilerPrintFormat_getName); diff --git a/src/jni/functions.cpp b/src/jni/functions.cpp index ab2b080fc..e2f24a247 100644 --- a/src/jni/functions.cpp +++ b/src/jni/functions.cpp @@ -271,6 +271,7 @@ JNIEXPORT jint JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1fetch_1size(JN } } +/* JNIEXPORT jlong JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1stream(JNIEnv * env, jclass param0, jobject param1, jlong param2) { try { return _duckdb_jdbc_arrow_stream(env, param0, param1, param2); @@ -291,6 +292,7 @@ JNIEXPORT void JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1registe } } +*/ JNIEXPORT jobject JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1create_1appender(JNIEnv * env, jclass param0, jobject param1, jbyteArray param2, jbyteArray param3) { try { diff --git a/src/jni/holders.hpp b/src/jni/holders.hpp index b47381343..9e07c4e54 100644 --- a/src/jni/holders.hpp +++ b/src/jni/holders.hpp @@ -53,7 +53,7 @@ struct PendingHolder { struct ResultHolder { duckdb::unique_ptr res; - duckdb::unique_ptr stream; + duckdb::unique_ptr> stream; duckdb::unique_ptr chunk; }; diff --git a/src/main/java/org/duckdb/DuckDBConnection.java b/src/main/java/org/duckdb/DuckDBConnection.java index f83cf0796..d2f4c1060 100644 --- a/src/main/java/org/duckdb/DuckDBConnection.java +++ b/src/main/java/org/duckdb/DuckDBConnection.java @@ -559,6 +559,8 @@ private static long getArrowStreamAddress(Object arrow_array_stream) { } public void registerArrowStream(String name, Object arrow_array_stream) { + throw new RuntimeException(new SQLFeatureNotSupportedException("registerArrowStream")); + /* try { checkOpen(); long array_stream_address = getArrowStreamAddress(arrow_array_stream); @@ -572,6 +574,7 @@ public void registerArrowStream(String name, Object arrow_array_stream) { } catch (SQLException e) { throw new RuntimeException(e); } + */ } public String getProfilingInformation(ProfilerPrintFormat format) throws SQLException { diff --git a/src/main/java/org/duckdb/DuckDBNative.java b/src/main/java/org/duckdb/DuckDBNative.java index 8600d7792..f5d3724f4 100644 --- a/src/main/java/org/duckdb/DuckDBNative.java +++ b/src/main/java/org/duckdb/DuckDBNative.java @@ -196,9 +196,12 @@ static native String[] duckdb_jdbc_cast_result_to_strings(ByteBuffer res_ref, By static native int duckdb_jdbc_fetch_size(); - static native long duckdb_jdbc_arrow_stream(ByteBuffer res_ref, long batch_size); + /* + static native long duckdb_jdbc_arrow_stream(ByteBuffer res_ref, long batch_size); - static native void duckdb_jdbc_arrow_register(ByteBuffer conn_ref, long arrow_array_stream_pointer, byte[] name); + static native void duckdb_jdbc_arrow_register(ByteBuffer conn_ref, long arrow_array_stream_pointer, byte[] + name); + */ static native ByteBuffer duckdb_jdbc_create_appender(ByteBuffer conn_ref, byte[] schema_name, byte[] table_name) throws SQLException; diff --git a/src/main/java/org/duckdb/DuckDBResultSet.java b/src/main/java/org/duckdb/DuckDBResultSet.java index 4e4698f65..4566e50df 100644 --- a/src/main/java/org/duckdb/DuckDBResultSet.java +++ b/src/main/java/org/duckdb/DuckDBResultSet.java @@ -134,24 +134,26 @@ public synchronized Object arrowExportStream(Object arrow_buffer_allocator, long throws SQLException { checkOpen(); - try { - Class buffer_allocator_class = Class.forName("org.apache.arrow.memory.BufferAllocator"); - if (!buffer_allocator_class.isInstance(arrow_buffer_allocator)) { - throw new RuntimeException("Need to pass an Arrow BufferAllocator"); - } - Long stream_pointer = DuckDBNative.duckdb_jdbc_arrow_stream(resultRef, arrow_batch_size); - Class arrow_array_stream_class = Class.forName("org.apache.arrow.c.ArrowArrayStream"); - Object arrow_array_stream = - arrow_array_stream_class.getMethod("wrap", long.class).invoke(null, stream_pointer); + throw new SQLFeatureNotSupportedException("arrowExportStream"); + /* + try { + Class buffer_allocator_class = Class.forName("org.apache.arrow.memory.BufferAllocator"); + if (!buffer_allocator_class.isInstance(arrow_buffer_allocator)) { + throw new RuntimeException("Need to pass an Arrow BufferAllocator"); + } + Long stream_pointer = DuckDBNative.duckdb_jdbc_arrow_stream(resultRef, arrow_batch_size); + Class arrow_array_stream_class = Class.forName("org.apache.arrow.c.ArrowArrayStream"); + Object arrow_array_stream = + arrow_array_stream_class.getMethod("wrap", long.class).invoke(null, stream_pointer); - Class c_data_class = Class.forName("org.apache.arrow.c.Data"); + Class c_data_class = Class.forName("org.apache.arrow.c.Data"); - return c_data_class.getMethod("importArrayStream", buffer_allocator_class, arrow_array_stream_class) - .invoke(null, arrow_buffer_allocator, arrow_array_stream); - } catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException | SecurityException | - ClassNotFoundException e) { - throw new RuntimeException(e); - } + return c_data_class.getMethod("importArrayStream", buffer_allocator_class, arrow_array_stream_class) + .invoke(null, arrow_buffer_allocator, arrow_array_stream); + } catch (IllegalAccessException | InvocationTargetException | NoSuchMethodException | SecurityException + | ClassNotFoundException e) { throw new RuntimeException(e); + } + */ } public Object getObject(int columnIndex) throws SQLException {