From 2ef940bceb5d8da92d0024eedaec72f95601493e Mon Sep 17 00:00:00 2001 From: Alex Kasko Date: Tue, 20 Oct 2026 18:41:33 +0100 Subject: [PATCH] Restore Arrow streams support --- .../linux-tests-host/action.yml | 7 +- .gitignore | 9 +- CMakeLists.txt | 1 + CMakeLists.txt.in | 1 + data/arrow/arrow_jars.txt | 14 + duckdb_java.def | 3 + duckdb_java.exp | 3 + duckdb_java.map | 3 + src/jni/arrow.cpp | 184 ++++ src/jni/arrow.hpp | 75 ++ src/jni/duckdb_java.cpp | 107 ++- src/jni/functions.cpp | 16 +- src/jni/functions.hpp | 8 +- src/jni/holders.cpp | 28 + src/jni/holders.hpp | 25 + .../java/org/duckdb/DuckDBConnection.java | 17 +- src/main/java/org/duckdb/DuckDBNative.java | 9 +- src/main/java/org/duckdb/DuckDBResultSet.java | 48 +- src/test/external/TestArrow.java | 788 ++++++++++++++++++ src/test/external/test_arrow.sh | 30 + src/test/test.iml | 11 - 21 files changed, 1285 insertions(+), 102 deletions(-) create mode 100644 data/arrow/arrow_jars.txt create mode 100644 src/jni/arrow.cpp create mode 100644 src/jni/arrow.hpp create mode 100644 src/test/external/TestArrow.java create mode 100755 src/test/external/test_arrow.sh delete mode 100644 src/test/test.iml diff --git a/.github/composite-actions/linux-tests-host/action.yml b/.github/composite-actions/linux-tests-host/action.yml index 6414e0483..16a67439e 100644 --- a/.github/composite-actions/linux-tests-host/action.yml +++ b/.github/composite-actions/linux-tests-host/action.yml @@ -4,7 +4,7 @@ description: 'Linux test run in host OS' runs: using: "composite" steps: - - name: Setup JDK 26 + - name: Setup JDK uses: actions/setup-java@v6 with: distribution: 'corretto' @@ -18,3 +18,8 @@ runs: java \ -cp ./build/release/duckdb_jdbc_tests.jar:./build/release/duckdb_jdbc.jar \ org.duckdb.TestDuckDBJDBC + + - name: Arrow Tests Host + shell: bash + run: | + ./src/test/external/test_arrow.sh diff --git a/.gitignore b/.gitignore index 427bdb67f..efc1d5e02 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,6 @@ -build -.idea -cmake-build-debug +/build +/.idea +/.local +/.arrow +*.iml +test1.db diff --git a/CMakeLists.txt b/CMakeLists.txt index 090360c33..300bfa70c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -651,6 +651,7 @@ if(JEMALLOC_ENABLED) endif() add_library(duckdb_java SHARED + src/jni/arrow.cpp src/jni/bindings_appender.cpp src/jni/bindings_common.cpp src/jni/bindings_data_chunk.cpp diff --git a/CMakeLists.txt.in b/CMakeLists.txt.in index 73ceaf49f..a798d6ac5 100644 --- a/CMakeLists.txt.in +++ b/CMakeLists.txt.in @@ -131,6 +131,7 @@ if(JEMALLOC_ENABLED) endif() add_library(duckdb_java SHARED + src/jni/arrow.cpp src/jni/bindings_appender.cpp src/jni/bindings_common.cpp src/jni/bindings_data_chunk.cpp diff --git a/data/arrow/arrow_jars.txt b/data/arrow/arrow_jars.txt new file mode 100644 index 000000000..08c86b4bf --- /dev/null +++ b/data/arrow/arrow_jars.txt @@ -0,0 +1,14 @@ +https://repo1.maven.org/maven2/org/apache/arrow/arrow-c-data/17.0.0/arrow-c-data-17.0.0.jar +https://repo1.maven.org/maven2/org/apache/arrow/arrow-format/17.0.0/arrow-format-17.0.0.jar +https://repo1.maven.org/maven2/org/apache/arrow/arrow-memory-core/17.0.0/arrow-memory-core-17.0.0.jar +https://repo1.maven.org/maven2/org/apache/arrow/arrow-memory-unsafe/17.0.0/arrow-memory-unsafe-17.0.0.jar +https://repo1.maven.org/maven2/org/apache/arrow/arrow-vector/17.0.0/arrow-vector-17.0.0.jar +https://repo1.maven.org/maven2/com/fasterxml/jackson/core/jackson-annotations/2.17.1/jackson-annotations-2.17.1.jar +https://repo1.maven.org/maven2/com/fasterxml/jackson/core/jackson-core/2.17.1/jackson-core-2.17.1.jar +https://repo1.maven.org/maven2/com/fasterxml/jackson/core/jackson-databind/2.17.1/jackson-databind-2.17.1.jar +https://repo1.maven.org/maven2/com/fasterxml/jackson/datatype/jackson-datatype-jsr310/2.17.1/jackson-datatype-jsr310-2.17.1.jar +https://repo1.maven.org/maven2/com/google/code/findbugs/jsr305/3.0.2/jsr305-3.0.2.jar +https://repo1.maven.org/maven2/com/google/flatbuffers/flatbuffers-java/24.3.25/flatbuffers-java-24.3.25.jar +https://repo1.maven.org/maven2/commons-codec/commons-codec/1.17.0/commons-codec-1.17.0.jar +https://repo1.maven.org/maven2/org/slf4j/slf4j-api/2.0.13/slf4j-api-2.0.13.jar +https://repo1.maven.org/maven2/org/checkerframework/checker-qual/3.44.0/checker-qual-3.44.0.jar diff --git a/duckdb_java.def b/duckdb_java.def index da3cca726..51b580df6 100644 --- a/duckdb_java.def +++ b/duckdb_java.def @@ -20,6 +20,9 @@ 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_1close_1streams +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 233681ecd..35e6ce1db 100644 --- a/duckdb_java.exp +++ b/duckdb_java.exp @@ -17,6 +17,9 @@ _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_1close_1streams +_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 55b47937a..51e1aa703 100644 --- a/duckdb_java.map +++ b/duckdb_java.map @@ -19,6 +19,9 @@ 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_1close_1streams; + 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/arrow.cpp b/src/jni/arrow.cpp new file mode 100644 index 000000000..ce74570b3 --- /dev/null +++ b/src/jni/arrow.cpp @@ -0,0 +1,184 @@ +#include "arrow.hpp" + +#include "duckdb/common/arrow/arrow_appender.hpp" +#include "duckdb/common/arrow/arrow_converter.hpp" +#include "duckdb/function/table/arrow/arrow_duck_schema.hpp" + +using namespace duckdb; + +JavaResultArrowStream::JavaResultArrowStream(duckdb::unique_ptr res_p, + duckdb::unique_ptr> stream_p, idx_t batch_size_p) + : res(std::move(res_p)), res_stream(std::move(stream_p)), batch_size(batch_size_p) { + if (res) { + types = res->GetTypes(); + names = IdentifiersToStrings(res->GetNames()); + properties = res->client_properties; + } else { + types = res_stream->GetTypes(); + names = IdentifiersToStrings(res_stream->GetNames()); + properties = res_stream->GetClientProperties(); + } + if (properties.client_context) { + extension_types = ArrowTypeExtensionData::GetExtensionTypes(*properties.client_context, types); + } +} + +void JavaResultArrowStream::Close() { + std::lock_guard guard(lock); + closed = true; + chunk = nullptr; + res = nullptr; + res_stream = nullptr; +} + +ArrowArrayStream *JavaResultArrowStream::Export(duckdb::shared_ptr state) { + auto exported = new Exported(); + exported->state = std::move(state); + auto &stream = exported->stream; + stream.private_data = exported; + stream.get_schema = GetSchema; + stream.get_next = GetNext; + stream.get_last_error = GetLastError; + stream.release = Release; + return &stream; +} + +JavaResultArrowStream &JavaResultArrowStream::GetState(ArrowArrayStream *stream) { + return *reinterpret_cast(stream->private_data)->state; +} + +int JavaResultArrowStream::GetSchema(ArrowArrayStream *stream, ArrowSchema *out) { + if (!stream->release) { + return -1; + } + auto &self = GetState(stream); + std::lock_guard guard(self.lock); + out->release = nullptr; + try { + ArrowConverter::ToArrowSchema(out, self.types, self.names, self.properties); + } catch (std::exception &e) { + self.last_error = ErrorData(e); + return -1; + } + return 0; +} + +int JavaResultArrowStream::GetNext(ArrowArrayStream *stream, ArrowArray *out) { + if (!stream->release) { + return -1; + } + auto &self = GetState(stream); + std::lock_guard guard(self.lock); + out->release = nullptr; + if (self.closed) { + self.last_error = ErrorData(ConnectionException("Connection was closed")); + return -1; + } + try { + ArrowAppender appender(self.types, self.batch_size, self.properties, self.extension_types); + idx_t count = 0; + while (count < self.batch_size) { + if (!self.chunk || self.chunk_offset >= self.chunk->size()) { + if (!self.FetchChunk()) { + return -1; + } + if (!self.chunk) { + break; + } + continue; + } + auto &chunk = *self.chunk; + auto to_append = MinValue(self.batch_size - count, chunk.size() - self.chunk_offset); + appender.Append(chunk, self.chunk_offset, self.chunk_offset + to_append, chunk.size()); + self.chunk_offset += to_append; + count += to_append; + } + if (count > 0) { + *out = appender.Finalize(); + } + } catch (std::exception &e) { + self.last_error = ErrorData(e); + return -1; + } + return 0; +} + +const char *JavaResultArrowStream::GetLastError(ArrowArrayStream *stream) { + if (!stream->release) { + return "stream was released"; + } + auto &self = GetState(stream); + std::lock_guard guard(self.lock); + return self.last_error.HasError() ? self.last_error.Message().c_str() : nullptr; +} + +void JavaResultArrowStream::Release(ArrowArrayStream *stream) { + if (!stream || !stream->release) { + return; + } + stream->release = nullptr; + delete reinterpret_cast(stream->private_data); +} + +bool JavaResultArrowStream::FetchChunk() { + chunk_offset = 0; + if (finished) { + chunk = nullptr; + return true; + } + if (res) { + chunk = res->Fetch(); + if (res->HasError()) { + last_error = res->GetErrorObject(); + return false; + } + } else { + chunk = res_stream->Fetch(); + if (res_stream->HasError()) { + last_error = res_stream->GetErrorObject(); + return false; + } + } + if (!chunk) { + finished = true; + } + return true; +} + +JavaRegisteredArrowStream::JavaRegisteredArrowStream(ArrowArrayStream *source) : stream(*source) { + source->release = nullptr; +} + +JavaRegisteredArrowStream::~JavaRegisteredArrowStream() { + Close(); +} + +void JavaRegisteredArrowStream::Close() { + std::lock_guard guard(lock); + if (stream.release) { + stream.release(&stream); + stream.release = nullptr; + } +} + +void JavaRegisteredArrowStream::GetSchema(ArrowSchema &schema) { + std::lock_guard guard(lock); + if (!stream.release) { + throw InvalidInputException("This stream has been released"); + } + if (stream.get_schema(&stream, &schema) != 0) { + auto error = stream.get_last_error(&stream); + throw InvalidInputException(error ? error : "Unknown error getting Arrow stream schema"); + } +} + +duckdb::unique_ptr JavaRegisteredArrowStream::Take() { + std::lock_guard guard(lock); + if (!stream.release) { + throw InvalidInputException("This stream has been released"); + } + auto res = make_uniq(); + res->arrow_array_stream = stream; + stream.release = nullptr; + return res; +} diff --git a/src/jni/arrow.hpp b/src/jni/arrow.hpp new file mode 100644 index 000000000..4bc1699a3 --- /dev/null +++ b/src/jni/arrow.hpp @@ -0,0 +1,75 @@ +#pragma once + +#include "duckdb.hpp" +#include "duckdb/common/arrow/arrow_wrapper.hpp" +#include "duckdb/main/query_result_stream.hpp" +#include "holders.hpp" + +#include + +namespace duckdb { +class ArrowTypeExtensionData; +} + +//! Exports the chunks of a JDBC result as an Arrow stream of record batches of at most batch_size rows. +//! The result is moved out of the ResultHolder into this state, that is shared between the exported stream +//! and the connection. Connection closes the state (releasing the result) if it is closed before the stream. +class JavaResultArrowStream : public ArrowStreamState { +public: + JavaResultArrowStream(duckdb::unique_ptr res_p, + duckdb::unique_ptr> stream_p, duckdb::idx_t batch_size_p); + + void Close() override; + + //! Creates a stream that shares the state, the stream is deleted when it is released + static ArrowArrayStream *Export(duckdb::shared_ptr state); + +private: + struct Exported { + ArrowArrayStream stream; + duckdb::shared_ptr state; + }; + + static JavaResultArrowStream &GetState(ArrowArrayStream *stream); + static int GetSchema(ArrowArrayStream *stream, ArrowSchema *out); + static int GetNext(ArrowArrayStream *stream, ArrowArray *out); + static const char *GetLastError(ArrowArrayStream *stream); + static void Release(ArrowArrayStream *stream); + + //! Loads the next chunk, chunk is null at the end of the result + bool FetchChunk(); + + std::mutex lock; + duckdb::unique_ptr res; + duckdb::unique_ptr> res_stream; + duckdb::idx_t batch_size; + duckdb::vector types; + duckdb::vector names; + duckdb::ClientProperties properties; + duckdb::unordered_map> extension_types; + duckdb::unique_ptr chunk; + duckdb::idx_t chunk_offset = 0; + bool finished = false; + bool closed = false; + duckdb::ErrorData last_error; +}; + +//! Arrow stream registered with a connection. The stream is moved out of the Java-owned struct on registration, +//! it is released when it is consumed by a scan, when the view is dropped or when the connection is closed. +class JavaRegisteredArrowStream : public ArrowStreamState { +public: + explicit JavaRegisteredArrowStream(ArrowArrayStream *source); + + ~JavaRegisteredArrowStream() override; + + void Close() override; + + void GetSchema(ArrowSchema &schema); + + //! A stream can only be consumed by a single scan + duckdb::unique_ptr Take(); + +private: + std::mutex lock; + ArrowArrayStream stream; +}; diff --git a/src/jni/duckdb_java.cpp b/src/jni/duckdb_java.cpp index 901468915..659f6878a 100644 --- a/src/jni/duckdb_java.cpp +++ b/src/jni/duckdb_java.cpp @@ -1,10 +1,10 @@ extern "C" { #include "duckdb.h" } +#include "arrow.hpp" #include "config.hpp" #include "duckdb.hpp" #include "duckdb/catalog/catalog_search_path.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" @@ -1047,71 +1047,66 @@ 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; +jlong _duckdb_jdbc_arrow_stream(JNIEnv *env, jclass, jobject res_ref_buf, jobject conn_ref_buf, jlong batch_size) { + if (!res_ref_buf) { + throw InvalidInputException("Invalid result set"); + } + auto res_ref = reinterpret_cast(env->GetDirectBufferAddress(res_ref_buf)); + 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"); + } + if (batch_size <= 0) { + throw InvalidInputException("Arrow batch size must be greater than 0"); + } + auto conn_holder = get_connection_ref(env, conn_ref_buf); + + auto state = make_shared_ptr(std::move(res_ref->res), std::move(res_ref->stream), + static_cast(batch_size)); + res_ref->chunk = nullptr; + conn_holder->track_arrow_stream(state); + return reinterpret_cast(JavaResultArrowStream::Export(std::move(state))); } -class JavaArrowTabularStreamFactory { +class JavaArrowScanFactory : public ArrowScanFactory { 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; + explicit JavaArrowScanFactory(duckdb::shared_ptr state_p) : state(std::move(state_p)) { + } + + void GetSchema(ArrowSchema &schema) override { + state->GetSchema(schema); + } + + duckdb::unique_ptr ProduceStream(ArrowStreamParameters &) override { + return state->Take(); + } + +private: + duckdb::shared_ptr state; }; 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 conn_holder = get_connection_ref(env, conn_ref_buf); + auto name = jbyteArray_to_string(env, name_j); + auto arrow_array_stream = reinterpret_cast(arrow_array_stream_pointer); + if (!arrow_array_stream || !arrow_array_stream->release) { + throw InvalidInputException("This stream has been released"); + } - 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 state = make_shared_ptr(arrow_array_stream); + conn_holder->track_arrow_stream(state); + auto factory = make_shared_ptr(std::move(state)); + conn->TableFunction("arrow_scan_dumb", {}, {}, std::move(factory))->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); +void _duckdb_jdbc_arrow_close_streams(JNIEnv *env, jclass, jobject conn_ref_buf) { + get_connection_ref(env, conn_ref_buf)->close_arrow_streams(); } -*/ 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 e2f24a247..564f6c252 100644 --- a/src/jni/functions.cpp +++ b/src/jni/functions.cpp @@ -271,10 +271,9 @@ 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) { +JNIEXPORT jlong JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1stream(JNIEnv * env, jclass param0, jobject param1, jobject param2, jlong param3) { try { - return _duckdb_jdbc_arrow_stream(env, param0, param1, param2); + return _duckdb_jdbc_arrow_stream(env, param0, param1, param2, param3); } catch (const std::exception &e) { duckdb::ErrorData error(e); ThrowJNI(env, error.Message().c_str()); @@ -292,7 +291,16 @@ JNIEXPORT void JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1registe } } -*/ + +JNIEXPORT void JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1close_1streams(JNIEnv * env, jclass param0, jobject param1) { + try { + return _duckdb_jdbc_arrow_close_streams(env, param0, param1); + } catch (const std::exception &e) { + duckdb::ErrorData error(e); + ThrowJNI(env, error.Message().c_str()); + + } +} 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/functions.hpp b/src/jni/functions.hpp index c6cadd826..c7f72177b 100644 --- a/src/jni/functions.hpp +++ b/src/jni/functions.hpp @@ -109,14 +109,18 @@ jint _duckdb_jdbc_fetch_size(JNIEnv * env, jclass param0); JNIEXPORT jint JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1fetch_1size(JNIEnv * env, jclass param0); -jlong _duckdb_jdbc_arrow_stream(JNIEnv * env, jclass param0, jobject param1, jlong param2); +jlong _duckdb_jdbc_arrow_stream(JNIEnv * env, jclass param0, jobject param1, jobject param2, jlong param3); -JNIEXPORT jlong JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1stream(JNIEnv * env, jclass param0, jobject param1, jlong param2); +JNIEXPORT jlong JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1stream(JNIEnv * env, jclass param0, jobject param1, jobject param2, jlong param3); void _duckdb_jdbc_arrow_register(JNIEnv * env, jclass param0, jobject param1, jlong param2, jbyteArray param3); JNIEXPORT void JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1register(JNIEnv * env, jclass param0, jobject param1, jlong param2, jbyteArray param3); +void _duckdb_jdbc_arrow_close_streams(JNIEnv * env, jclass param0, jobject param1); + +JNIEXPORT void JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1arrow_1close_1streams(JNIEnv * env, jclass param0, jobject param1); + jobject _duckdb_jdbc_create_appender(JNIEnv * env, jclass param0, jobject param1, jbyteArray param2, jbyteArray param3); JNIEXPORT jobject JNICALL Java_org_duckdb_DuckDBNative_duckdb_1jdbc_1create_1appender(JNIEnv * env, jclass param0, jobject param1, jbyteArray param2, jbyteArray param3); diff --git a/src/jni/holders.cpp b/src/jni/holders.cpp index ae32b9900..4c13a2a05 100644 --- a/src/jni/holders.cpp +++ b/src/jni/holders.cpp @@ -1,5 +1,6 @@ #include "holders.hpp" +#include #include ConnectionHolder *get_connection_ref(JNIEnv *env, jobject conn_ref_buf) { @@ -13,6 +14,33 @@ ConnectionHolder *get_connection_ref(JNIEnv *env, jobject conn_ref_buf) { return conn_holder; } +void ConnectionHolder::track_arrow_stream(duckdb::shared_ptr state) { + std::lock_guard guard(arrow_streams_lock); + auto expired = [](const duckdb::weak_ptr &weak) { + return weak.expired(); + }; + arrow_streams.erase(std::remove_if(arrow_streams.begin(), arrow_streams.end(), expired), arrow_streams.end()); + arrow_streams.emplace_back(std::move(state)); +} + +void ConnectionHolder::close_arrow_streams() { + duckdb::vector> live; + { + std::lock_guard guard(arrow_streams_lock); + // Last stream created is first closed + for (auto it = arrow_streams.rbegin(); it != arrow_streams.rend(); ++it) { + auto state = it->lock(); + if (state) { + live.emplace_back(std::move(state)); + } + } + arrow_streams.clear(); + } + for (auto &state : live) { + state->Close(); + } +} + /** * Throws a SQLException and returns nullptr if a valid Connection can't be retrieved from the buffer. */ diff --git a/src/jni/holders.hpp b/src/jni/holders.hpp index 9e07c4e54..bd5eced77 100644 --- a/src/jni/holders.hpp +++ b/src/jni/holders.hpp @@ -7,6 +7,7 @@ extern "C" { #include "refs.hpp" #include +#include /** * Holds a copy of a shared_ptr to an existing DB instance. @@ -23,6 +24,17 @@ struct DBHolder { DBHolder &operator=(const DBHolder &) = delete; }; +/** + * State of an Arrow stream exported from or registered with a connection. Arrow streams are owned by their consumers + * and can outlive the connection, connection closes the state of all its live streams before it is disconnected. + */ +struct ArrowStreamState { + virtual ~ArrowStreamState() = default; + + //! Releases the resources held by the stream, must be idempotent + virtual void Close() = 0; +}; + /** * 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 @@ -37,9 +49,22 @@ struct ConnectionHolder { : db(_db), connection(duckdb::make_uniq(*_db)) { } + ~ConnectionHolder() { + close_arrow_streams(); + } + + void track_arrow_stream(duckdb::shared_ptr state); + + void close_arrow_streams(); + DBHolder *create_db_ref() { return new DBHolder(db); } + +private: + std::mutex arrow_streams_lock; + //! Streams are untracked implicitly when they are released by their consumers + duckdb::vector> arrow_streams; }; struct StatementHolder { diff --git a/src/main/java/org/duckdb/DuckDBConnection.java b/src/main/java/org/duckdb/DuckDBConnection.java index 2f7c6dc7c..2934c609f 100644 --- a/src/main/java/org/duckdb/DuckDBConnection.java +++ b/src/main/java/org/duckdb/DuckDBConnection.java @@ -215,6 +215,10 @@ public void close() throws SQLException { } appenders.clear(); + // Arrow streams are owned by their consumers and are not tracked + // on Java side, last stream created is first closed on native side + DuckDBNative.duckdb_jdbc_arrow_close_streams(connRef); + DuckDBNative.duckdb_jdbc_disconnect(connRef); connRef = null; notifyMonitor = (monitorName != null); @@ -561,9 +565,17 @@ private static long getArrowStreamAddress(Object arrow_array_stream) { } } + /** + * Register an Arrow stream as a temporary view in this connection + * + *

Ownership of the stream is taken by the connection, the passed instance is marked as released. The stream + * can be consumed by a single query only. It is released when it is consumed, when the view is replaced or + * when the connection is closed. + * + * @param name name of the view + * @param arrow_array_stream an instance of {@link org.apache.arrow.c.ArrowArrayStream} + */ 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); @@ -577,7 +589,6 @@ 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 f5d3724f4..6c9c94d95 100644 --- a/src/main/java/org/duckdb/DuckDBNative.java +++ b/src/main/java/org/duckdb/DuckDBNative.java @@ -196,12 +196,11 @@ 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, ByteBuffer conn_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 void duckdb_jdbc_arrow_close_streams(ByteBuffer conn_ref); 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 6f97bda30..0fabd7c81 100644 --- a/src/main/java/org/duckdb/DuckDBResultSet.java +++ b/src/main/java/org/duckdb/DuckDBResultSet.java @@ -126,6 +126,9 @@ private void check(int columnIndex) throws SQLException { /** * Export the result set as an ArrowReader * + *

The rows of the result are moved into the returned reader, the reader must be closed by the caller. + * When the connection is closed before the reader, the reader fails to load next batches. + * * @param arrow_buffer_allocator an instance of {@link org.apache.arrow.memory.BufferAllocator} * @param arrow_batch_size batch size of arrow vectors to return * @return an instance of {@link org.apache.arrow.vector.ipc.ArrowReader} @@ -134,26 +137,37 @@ public synchronized Object arrowExportStream(Object arrow_buffer_allocator, long throws SQLException { checkOpen(); - 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; + conn.connRefLock.lock(); + try { + conn.checkOpen(); + resultRefLock.lock(); 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); + checkOpen(); + stream_pointer = DuckDBNative.duckdb_jdbc_arrow_stream(resultRef, conn.connRef, arrow_batch_size); + } finally { + resultRefLock.unlock(); + } + } finally { + conn.connRefLock.unlock(); + } + 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 { diff --git a/src/test/external/TestArrow.java b/src/test/external/TestArrow.java new file mode 100644 index 000000000..70e62e02e --- /dev/null +++ b/src/test/external/TestArrow.java @@ -0,0 +1,788 @@ +import static java.nio.charset.StandardCharsets.UTF_8; +import static java.util.Arrays.asList; +import static org.duckdb.DuckDBDriver.JDBC_STREAM_RESULTS; +import static org.duckdb.TestDuckDBJDBC.JDBC_URL; +import static org.duckdb.test.Assertions.*; +import static org.duckdb.test.Runner.runTests; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.math.BigDecimal; +import java.sql.*; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import org.apache.arrow.c.ArrowArrayStream; +import org.apache.arrow.c.Data; +import org.apache.arrow.memory.BufferAllocator; +import org.apache.arrow.memory.RootAllocator; +import org.apache.arrow.vector.BigIntVector; +import org.apache.arrow.vector.BitVector; +import org.apache.arrow.vector.DateDayVector; +import org.apache.arrow.vector.DecimalVector; +import org.apache.arrow.vector.FieldVector; +import org.apache.arrow.vector.Float8Vector; +import org.apache.arrow.vector.IntVector; +import org.apache.arrow.vector.TimeStampMicroVector; +import org.apache.arrow.vector.VarCharVector; +import org.apache.arrow.vector.VectorSchemaRoot; +import org.apache.arrow.vector.complex.ListVector; +import org.apache.arrow.vector.complex.StructVector; +import org.apache.arrow.vector.ipc.ArrowReader; +import org.apache.arrow.vector.ipc.ArrowStreamReader; +import org.apache.arrow.vector.ipc.ArrowStreamWriter; +import org.apache.arrow.vector.types.FloatingPointPrecision; +import org.apache.arrow.vector.types.pojo.ArrowType; +import org.apache.arrow.vector.types.pojo.Field; +import org.apache.arrow.vector.types.pojo.Schema; +import org.duckdb.DuckDBConnection; +import org.duckdb.DuckDBResultSet; + +/** + * Tests for Arrow integration: {@link DuckDBConnection#registerArrowStream} and + * {@link DuckDBResultSet#arrowExportStream}. + * + * Launched directly from source with test_arrow.sh, requires duckdb_jdbc.jar, duckdb_jdbc_tests.jar + * and Arrow JARs (listed in data/arrow/arrow_jars.txt) on classpath. + */ +public class TestArrow { + + private static final Schema PEOPLE_SCHEMA = new Schema( + asList(Field.nullable("id", new ArrowType.Int(32, true)), Field.nullable("name", ArrowType.Utf8.INSTANCE), + Field.nullable("score", new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE)), + Field.nullable("active", ArrowType.Bool.INSTANCE))); + + private static final Schema NUMBERS_SCHEMA = new Schema(asList(Field.nullable("v", new ArrowType.Int(64, true)))); + + // registerArrowStream + + public static void test_arrow_register_count() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + byte[] ipc = + peopleIpc(allocator, person(1, "a", 1.0, true), person(2, "b", 2.0, false), person(3, "c", 3.0, true)); + register(conn, allocator, stream, "people", ipc); + + try (ResultSet rs = stmt.executeQuery("SELECT count(*) FROM people")) { + assertTrue(rs.next()); + assertEquals(rs.getLong(1), 3L); + assertFalse(rs.next()); + } + } + } + + public static void test_arrow_register_values() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + byte[] ipc = peopleIpc(allocator, person(1, "alice", 1.5, true), person(2, "bob", 2.5, false)); + register(conn, allocator, stream, "people", ipc); + + try (ResultSet rs = stmt.executeQuery("SELECT id, name, score, active FROM people ORDER BY id")) { + assertTrue(rs.next()); + assertEquals(rs.getInt("id"), 1); + assertEquals(rs.getString("name"), "alice"); + assertEquals(rs.getDouble("score"), 1.5d); + assertTrue(rs.getBoolean("active")); + + assertTrue(rs.next()); + assertEquals(rs.getInt("id"), 2); + assertEquals(rs.getString("name"), "bob"); + assertEquals(rs.getDouble("score"), 2.5d); + assertFalse(rs.getBoolean("active")); + + assertFalse(rs.next()); + } + } + } + + public static void test_arrow_register_column_types() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + register(conn, allocator, stream, "people", peopleIpc(allocator, person(1, "a", 1.0, true))); + + List columns = new ArrayList<>(); + try (ResultSet rs = stmt.executeQuery("DESCRIBE SELECT * FROM people")) { + while (rs.next()) { + columns.add(rs.getString("column_name") + ":" + rs.getString("column_type")); + } + } + assertListsEqual(columns, asList("id:INTEGER", "name:VARCHAR", "score:DOUBLE", "active:BOOLEAN")); + } + } + + public static void test_arrow_register_nulls() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + byte[] ipc = peopleIpc(allocator, person(1, null, null, null), person(2, "x", 4.0, true)); + register(conn, allocator, stream, "people", ipc); + + try (ResultSet rs = stmt.executeQuery("SELECT name, score, active FROM people ORDER BY id")) { + assertTrue(rs.next()); + assertNull(rs.getString(1)); + rs.getDouble(2); + assertTrue(rs.wasNull()); + rs.getBoolean(3); + assertTrue(rs.wasNull()); + + assertTrue(rs.next()); + assertEquals(rs.getString(1), "x"); + assertFalse(rs.next()); + } + } + } + + public static void test_arrow_register_multiple_batches() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + // 5 batches of 1000 rows, values 0..4999 + register(conn, allocator, stream, "numbers", numbersIpc(allocator, 5, 1000)); + + try (ResultSet rs = stmt.executeQuery("SELECT count(*), sum(v), min(v), max(v) FROM numbers")) { + assertTrue(rs.next()); + assertEquals(rs.getLong(1), 5000L); + assertEquals(rs.getLong(2), 4999L * 5000L / 2); + assertEquals(rs.getLong(3), 0L); + assertEquals(rs.getLong(4), 4999L); + assertFalse(rs.next()); + } + } + } + + public static void test_arrow_register_large_batches() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + // batches much larger than DuckDB vector size + int batchCount = 10; + int batchSize = 100_000; + register(conn, allocator, stream, "numbers", numbersIpc(allocator, batchCount, batchSize)); + + try (ResultSet rs = stmt.executeQuery("SELECT count(*), sum(v) FROM numbers")) { + assertTrue(rs.next()); + long total = (long) batchCount * batchSize; + assertEquals(rs.getLong(1), total); + assertEquals(rs.getLong(2), (total - 1) * total / 2); + assertFalse(rs.next()); + } + } + } + + public static void test_arrow_register_empty_stream() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + register(conn, allocator, stream, "people", peopleIpc(allocator)); + + try (ResultSet rs = stmt.executeQuery("SELECT * FROM people")) { + ResultSetMetaData meta = rs.getMetaData(); + assertEquals(meta.getColumnCount(), 4); + assertEquals(meta.getColumnName(1), "id"); + assertEquals(meta.getColumnName(4), "active"); + assertFalse(rs.next()); + } + } + } + + public static void test_arrow_register_aggregation() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + byte[] ipc = peopleIpc(allocator, person(1, "a", 10.0, true), person(2, "b", 20.0, false), + person(3, "c", 30.0, true), person(4, "d", 40.0, true)); + register(conn, allocator, stream, "people", ipc); + + try (ResultSet rs = stmt.executeQuery("SELECT active, count(*) AS n, avg(score) AS avg_score " + + "FROM people WHERE id > 1 GROUP BY active ORDER BY active")) { + assertTrue(rs.next()); + assertFalse(rs.getBoolean("active")); + assertEquals(rs.getLong("n"), 1L); + assertEquals(rs.getDouble("avg_score"), 20.0d); + + assertTrue(rs.next()); + assertTrue(rs.getBoolean("active")); + assertEquals(rs.getLong("n"), 2L); + assertEquals(rs.getDouble("avg_score"), 35.0d); + + assertFalse(rs.next()); + } + } + } + + public static void test_arrow_register_join_two_streams() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream peopleStream = ArrowArrayStream.allocateNew(allocator); + ArrowArrayStream numbersStream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + byte[] peopleIpc = peopleIpc(allocator, person(0, "alice", 0.0, true), person(1, "bob", 0.0, true), + person(2, "carol", 0.0, true)); + register(conn, allocator, peopleStream, "people", peopleIpc); + // values 0..9 + register(conn, allocator, numbersStream, "numbers", numbersIpc(allocator, 2, 5)); + + try (ResultSet rs = stmt.executeQuery("SELECT p.name, count(*) FROM people p " + + "JOIN numbers n ON p.id = n.v % 3 " + + "GROUP BY p.name ORDER BY p.name")) { + assertTrue(rs.next()); + assertEquals(rs.getString(1), "alice"); + assertEquals(rs.getLong(2), 4L); + assertTrue(rs.next()); + assertEquals(rs.getString(1), "bob"); + assertEquals(rs.getLong(2), 3L); + assertTrue(rs.next()); + assertEquals(rs.getString(1), "carol"); + assertEquals(rs.getLong(2), 3L); + assertFalse(rs.next()); + } + } + } + + public static void test_arrow_register_join_table() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE bonus (id INTEGER, bonus DOUBLE)"); + stmt.execute("INSERT INTO bonus VALUES (1, 0.5), (3, 1.5)"); + byte[] ipc = + peopleIpc(allocator, person(1, "a", 1.0, true), person(2, "b", 2.0, true), person(3, "c", 3.0, true)); + register(conn, allocator, stream, "people", ipc); + + try (ResultSet rs = + stmt.executeQuery("SELECT sum(p.score + b.bonus) FROM people p JOIN bonus b USING (id)")) { + assertTrue(rs.next()); + assertEquals(rs.getDouble(1), 1.0d + 0.5d + 3.0d + 1.5d); + assertFalse(rs.next()); + } + } + } + + public static void test_arrow_register_create_table_as() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + byte[] ipc = peopleIpc(allocator, person(1, "a", 1.0, true), person(2, "b", 2.0, false)); + register(conn, allocator, stream, "people", ipc); + + stmt.execute("CREATE TABLE people_copy AS SELECT * FROM people"); + + // copy is a regular table and can be queried multiple times + for (int i = 0; i < 2; i++) { + try (ResultSet rs = stmt.executeQuery("SELECT count(*) FILTER (active), count(*) FROM people_copy")) { + assertTrue(rs.next()); + assertEquals(rs.getLong(1), 1L); + assertEquals(rs.getLong(2), 2L); + assertFalse(rs.next()); + } + } + } + } + + public static void test_arrow_register_stream_consumed_once() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + byte[] ipc = peopleIpc(allocator, person(1, "a", 1.0, true), person(2, "b", 2.0, true)); + register(conn, allocator, stream, "people", ipc); + + try (ResultSet rs = stmt.executeQuery("SELECT count(*) FROM people")) { + assertTrue(rs.next()); + assertEquals(rs.getLong(1), 2L); + } + + // Arrow stream can only be read once, it is released after the first scan + String msg = assertThrows(() -> { stmt.executeQuery("SELECT count(*) FROM people"); }, SQLException.class); + assertTrue(msg.contains("This stream has been released"), msg); + } + } + + public static void test_arrow_register_connection_scope() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + register(conn, allocator, stream, "people", peopleIpc(allocator, person(1, "a", 1.0, true))); + + try (Connection dup = conn.duplicate(); Statement dupStmt = dup.createStatement()) { + String msg = + assertThrows(() -> { dupStmt.executeQuery("SELECT count(*) FROM people"); }, SQLException.class); + assertTrue(msg.contains("Table with name people does not exist"), msg); + } + + try (ResultSet rs = stmt.executeQuery("SELECT count(*) FROM people")) { + assertTrue(rs.next()); + assertEquals(rs.getLong(1), 1L); + } + } + } + + public static void test_arrow_register_replace() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream1 = ArrowArrayStream.allocateNew(allocator); + ArrowArrayStream stream2 = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + register(conn, allocator, stream1, "tab1", peopleIpc(allocator, person(1, "a", 1.0, true))); + try (ResultSet rs = stmt.executeQuery("SELECT count(*) FROM tab1")) { + assertTrue(rs.next()); + assertEquals(rs.getLong(1), 1L); + } + + byte[] ipc = + peopleIpc(allocator, person(1, "a", 1.0, true), person(2, "b", 2.0, true), person(3, "c", 3.0, true)); + register(conn, allocator, stream2, "tab1", ipc); + try (ResultSet rs = stmt.executeQuery("SELECT count(*) FROM tab1")) { + assertTrue(rs.next()); + assertEquals(rs.getLong(1), 3L); + } + } + } + + public static void test_arrow_register_stream_moved() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement()) { + TrackingReader reader = new TrackingReader(allocator, peopleIpc(allocator, person(1, "a", 1.0, true))); + // stream is moved into the connection on registration, the passed struct can be closed right away + try (ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator)) { + Data.exportArrayStream(allocator, reader, stream); + conn.registerArrowStream("people", stream); + } + assertFalse(reader.closed); + + try (ResultSet rs = stmt.executeQuery("SELECT count(*) FROM people")) { + assertTrue(rs.next()); + assertEquals(rs.getLong(1), 1L); + } + assertTrue(reader.closed); + } + } + + public static void test_arrow_register_released_on_connection_close() throws Exception { + try (BufferAllocator allocator = new RootAllocator()) { + TrackingReader reader1 = new TrackingReader(allocator, peopleIpc(allocator, person(1, "a", 1.0, true))); + TrackingReader reader2 = new TrackingReader(allocator, numbersIpc(allocator, 1, 10)); + try (ArrowArrayStream stream1 = ArrowArrayStream.allocateNew(allocator); + ArrowArrayStream stream2 = ArrowArrayStream.allocateNew(allocator)) { + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Data.exportArrayStream(allocator, reader1, stream1); + conn.registerArrowStream("people", stream1); + Data.exportArrayStream(allocator, reader2, stream2); + conn.registerArrowStream("numbers", stream2); + + // streams that were never consumed are released when connection is closed + assertFalse(reader1.closed); + assertFalse(reader2.closed); + conn.close(); + assertTrue(reader1.closed); + assertTrue(reader2.closed); + + assertThrows(() -> { conn.registerArrowStream("people", stream1); }, RuntimeException.class); + } + } + } + + public static void test_arrow_register_invalid_argument() throws Exception { + try (DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class)) { + String msg = + assertThrows(() -> { conn.registerArrowStream("tab1", "not a stream"); }, RuntimeException.class); + assertTrue(msg.contains("Need to pass an ArrowArrayStream"), msg); + } + } + + public static void test_arrow_register_closed_connection() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator)) { + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + conn.close(); + assertThrows(() -> { conn.registerArrowStream("tab1", stream); }, RuntimeException.class); + } + } + + // arrowExportStream + + public static void test_arrow_export_values() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); Connection conn = DriverManager.getConnection(JDBC_URL); + Statement stmt = conn.createStatement(); + DuckDBResultSet rs = stmt.executeQuery("SELECT i::INTEGER AS id, 'name_' || i AS name, " + + "i / 2 AS half, i % 2 = 0 AS even FROM range(0, 3) t(i)") + .unwrap(DuckDBResultSet.class); + ArrowReader reader = (ArrowReader) rs.arrowExportStream(allocator, 1024)) { + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + assertTrue(reader.loadNextBatch()); + assertEquals(root.getRowCount(), 3); + + IntVector id = (IntVector) root.getVector("id"); + VarCharVector name = (VarCharVector) root.getVector("name"); + Float8Vector half = (Float8Vector) root.getVector("half"); + BitVector even = (BitVector) root.getVector("even"); + for (int i = 0; i < 3; i++) { + assertEquals(id.get(i), i); + assertEquals(new String(name.get(i), UTF_8), "name_" + i); + assertEquals(half.get(i), i / 2.0d); + assertEquals(even.get(i), i % 2 == 0 ? 1 : 0); + } + + assertFalse(reader.loadNextBatch()); + } + } + + public static void test_arrow_export_schema() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); Connection conn = DriverManager.getConnection(JDBC_URL); + Statement stmt = conn.createStatement(); + DuckDBResultSet rs = stmt.executeQuery("SELECT 42::INTEGER AS c_int, 42::BIGINT AS c_bigint, " + + "'foo' AS c_varchar, 1.5::DOUBLE AS c_double, " + + "true AS c_bool, DATE '2020-01-02' AS c_date") + .unwrap(DuckDBResultSet.class); + ArrowReader reader = (ArrowReader) rs.arrowExportStream(allocator, 1024)) { + Schema schema = reader.getVectorSchemaRoot().getSchema(); + List names = new ArrayList<>(); + for (Field field : schema.getFields()) { + names.add(field.getName()); + } + assertListsEqual(names, asList("c_int", "c_bigint", "c_varchar", "c_double", "c_bool", "c_date")); + assertEquals(schema.findField("c_int").getType(), new ArrowType.Int(32, true)); + assertEquals(schema.findField("c_bigint").getType(), new ArrowType.Int(64, true)); + assertEquals(schema.findField("c_varchar").getType(), ArrowType.Utf8.INSTANCE); + assertEquals(schema.findField("c_double").getType(), + new ArrowType.FloatingPoint(FloatingPointPrecision.DOUBLE)); + assertEquals(schema.findField("c_bool").getType(), ArrowType.Bool.INSTANCE); + assertTrue(schema.findField("c_date").getType() instanceof ArrowType.Date); + } + } + + public static void test_arrow_export_temporal_and_decimal() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); Connection conn = DriverManager.getConnection(JDBC_URL); + Statement stmt = conn.createStatement(); + DuckDBResultSet rs = stmt.executeQuery("SELECT DATE '1970-01-11' AS c_date, " + + "TIMESTAMP '2020-01-02 03:04:05.123456' AS c_ts, " + + "123.456::DECIMAL(18,3) AS c_dec") + .unwrap(DuckDBResultSet.class); + ArrowReader reader = (ArrowReader) rs.arrowExportStream(allocator, 1024)) { + assertTrue(reader.loadNextBatch()); + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + assertEquals(root.getRowCount(), 1); + + DateDayVector date = (DateDayVector) root.getVector("c_date"); + assertEquals(date.get(0), 10); + + TimeStampMicroVector ts = (TimeStampMicroVector) root.getVector("c_ts"); + assertEquals(ts.getObject(0), LocalDateTime.of(2020, 1, 2, 3, 4, 5, 123456000)); + + DecimalVector dec = (DecimalVector) root.getVector("c_dec"); + assertEquals(dec.getObject(0), new BigDecimal("123.456")); + } + } + + public static void test_arrow_export_nested() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); Connection conn = DriverManager.getConnection(JDBC_URL); + Statement stmt = conn.createStatement(); + DuckDBResultSet rs = stmt.executeQuery("SELECT [1, 2, NULL, 4] AS c_list, " + + "{'a': 42, 'b': 'foo'} AS c_struct") + .unwrap(DuckDBResultSet.class); + ArrowReader reader = (ArrowReader) rs.arrowExportStream(allocator, 1024)) { + assertTrue(reader.loadNextBatch()); + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + + ListVector list = (ListVector) root.getVector("c_list"); + assertEquals(list.getObject(0), asList(1, 2, null, 4)); + + StructVector struct = (StructVector) root.getVector("c_struct"); + IntVector a = (IntVector) struct.getChild("a"); + VarCharVector b = (VarCharVector) struct.getChild("b"); + assertEquals(a.get(0), 42); + assertEquals(new String(b.get(0), UTF_8), "foo"); + } + } + + public static void test_arrow_export_nulls() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); Connection conn = DriverManager.getConnection(JDBC_URL); + Statement stmt = conn.createStatement(); + DuckDBResultSet rs = stmt.executeQuery("SELECT CASE WHEN i % 2 = 0 THEN NULL ELSE i END AS v, " + + "CASE WHEN i % 2 = 0 THEN NULL ELSE 'foo' END AS s " + + "FROM range(0, 4) t(i)") + .unwrap(DuckDBResultSet.class); + ArrowReader reader = (ArrowReader) rs.arrowExportStream(allocator, 1024)) { + assertTrue(reader.loadNextBatch()); + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + FieldVector v = root.getVector("v"); + FieldVector s = root.getVector("s"); + assertEquals(root.getRowCount(), 4); + assertEquals(v.getNullCount(), 2); + assertEquals(s.getNullCount(), 2); + for (int i = 0; i < 4; i++) { + assertEquals(v.isNull(i), i % 2 == 0); + assertEquals(s.isNull(i), i % 2 == 0); + } + } + } + + public static void test_arrow_export_batch_size() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); Connection conn = DriverManager.getConnection(JDBC_URL); + Statement stmt = conn.createStatement(); + DuckDBResultSet rs = stmt.executeQuery("SELECT i FROM range(0, 10000) t(i)").unwrap(DuckDBResultSet.class); + ArrowReader reader = (ArrowReader) rs.arrowExportStream(allocator, 4096)) { + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + List batchSizes = new ArrayList<>(); + long sum = 0; + while (reader.loadNextBatch()) { + batchSizes.add(root.getRowCount()); + BigIntVector vec = (BigIntVector) root.getVector("i"); + for (int i = 0; i < root.getRowCount(); i++) { + sum += vec.get(i); + } + } + assertListsEqual(batchSizes, asList(4096, 4096, 1808)); + assertEquals(sum, 9999L * 10000L / 2); + } + } + + public static void test_arrow_export_streaming_result() throws Exception { + Properties config = new Properties(); + config.put(JDBC_STREAM_RESULTS, true); + try (BufferAllocator allocator = new RootAllocator(); + Connection conn = DriverManager.getConnection(JDBC_URL, config); Statement stmt = conn.createStatement(); + DuckDBResultSet rs = stmt.executeQuery("SELECT i, 'name_' || i AS name FROM range(0, 100000) t(i)") + .unwrap(DuckDBResultSet.class); + ArrowReader reader = (ArrowReader) rs.arrowExportStream(allocator, 10000)) { + VectorSchemaRoot root = reader.getVectorSchemaRoot(); + long count = 0; + long sum = 0; + int batches = 0; + while (reader.loadNextBatch()) { + assertTrue(root.getRowCount() <= 10000); + BigIntVector vec = (BigIntVector) root.getVector("i"); + VarCharVector name = (VarCharVector) root.getVector("name"); + for (int i = 0; i < root.getRowCount(); i++) { + assertEquals(new String(name.get(i), UTF_8), "name_" + vec.get(i)); + sum += vec.get(i); + } + count += root.getRowCount(); + batches++; + } + assertEquals(count, 100000L); + assertEquals(sum, 99999L * 100000L / 2); + assertEquals(batches, 10); + } + } + + public static void test_arrow_export_empty_result() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); Connection conn = DriverManager.getConnection(JDBC_URL); + Statement stmt = conn.createStatement(); + DuckDBResultSet rs = stmt.executeQuery("SELECT i, i::VARCHAR AS s FROM range(0, 10) t(i) WHERE i < 0") + .unwrap(DuckDBResultSet.class); + ArrowReader reader = (ArrowReader) rs.arrowExportStream(allocator, 1024)) { + assertEquals(reader.getVectorSchemaRoot().getSchema().getFields().size(), 2); + assertFalse(reader.loadNextBatch()); + } + } + + public static void test_arrow_export_invalid_argument() throws Exception { + try (Connection conn = DriverManager.getConnection(JDBC_URL); Statement stmt = conn.createStatement(); + DuckDBResultSet rs = stmt.executeQuery("SELECT 42").unwrap(DuckDBResultSet.class)) { + String msg = + assertThrows(() -> { rs.arrowExportStream("not an allocator", 1024); }, RuntimeException.class); + assertTrue(msg.contains("Need to pass an Arrow BufferAllocator"), msg); + } + } + + public static void test_arrow_export_closed_result_set() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); Connection conn = DriverManager.getConnection(JDBC_URL); + Statement stmt = conn.createStatement()) { + DuckDBResultSet rs = stmt.executeQuery("SELECT 42").unwrap(DuckDBResultSet.class); + rs.close(); + assertThrows(() -> { rs.arrowExportStream(allocator, 1024); }, SQLException.class); + } + } + + public static void test_arrow_export_connection_closed() throws Exception { + try (BufferAllocator allocator = new RootAllocator()) { + Connection conn = DriverManager.getConnection(JDBC_URL); + Statement stmt = conn.createStatement(); + DuckDBResultSet rs = stmt.executeQuery("SELECT i FROM range(0, 10000) t(i)").unwrap(DuckDBResultSet.class); + try (ArrowReader reader = (ArrowReader) rs.arrowExportStream(allocator, 1024)) { + assertTrue(reader.loadNextBatch()); + assertEquals(reader.getVectorSchemaRoot().getRowCount(), 1024); + + // result is released when connection is closed, reader stays usable only for closing + conn.close(); + String msg = assertThrows(reader::loadNextBatch, Exception.class); + assertTrue(msg.contains("Connection was closed"), msg); + } + assertThrows(() -> { rs.arrowExportStream(allocator, 1024); }, SQLException.class); + } + } + + public static void test_arrow_export_reader_closed_before_connection() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); Connection conn = DriverManager.getConnection(JDBC_URL); + Statement stmt = conn.createStatement()) { + for (int i = 0; i < 10; i++) { + try (DuckDBResultSet rs = + stmt.executeQuery("SELECT i FROM range(0, 10000) t(i)").unwrap(DuckDBResultSet.class); + ArrowReader reader = (ArrowReader) rs.arrowExportStream(allocator, 1024)) { + assertTrue(reader.loadNextBatch()); + } + } + // connection stays usable after the exported readers are closed + try (ResultSet rs = stmt.executeQuery("SELECT 42")) { + assertTrue(rs.next()); + assertEquals(rs.getInt(1), 42); + } + } + } + + // arrowExportStream -> registerArrowStream + + public static void test_arrow_export_register_round_trip() throws Exception { + try (BufferAllocator allocator = new RootAllocator(); + ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator); + DuckDBConnection conn = DriverManager.getConnection(JDBC_URL).unwrap(DuckDBConnection.class); + Statement stmt = conn.createStatement(); + DuckDBResultSet rs = stmt.executeQuery("SELECT i AS id, 'name_' || i AS name FROM range(0, 10000) t(i)") + .unwrap(DuckDBResultSet.class)) { + ArrowReader reader = (ArrowReader) rs.arrowExportStream(allocator, 1024); + // reader is owned by the exported stream and is closed by DuckDB when the stream is released + Data.exportArrayStream(allocator, reader, stream); + conn.registerArrowStream("exported", stream); + + try (Statement stmt2 = conn.createStatement(); + ResultSet rs2 = stmt2.executeQuery("SELECT count(*), sum(id), count(DISTINCT name) FROM exported")) { + assertTrue(rs2.next()); + assertEquals(rs2.getLong(1), 10000L); + assertEquals(rs2.getLong(2), 9999L * 10000L / 2); + assertEquals(rs2.getLong(3), 10000L); + assertFalse(rs2.next()); + } + } + } + + // helpers + + private static class Person { + final Integer id; + final String name; + final Double score; + final Boolean active; + + Person(Integer id, String name, Double score, Boolean active) { + this.id = id; + this.name = name; + this.score = score; + this.active = active; + } + } + + private static Person person(Integer id, String name, Double score, Boolean active) { + return new Person(id, name, score, active); + } + + // records whether the reader was closed by its consumer + private static class TrackingReader extends ArrowStreamReader { + boolean closed = false; + + TrackingReader(BufferAllocator allocator, byte[] ipc) { + super(new ByteArrayInputStream(ipc), allocator); + } + + @Override + protected void closeReadSource() throws IOException { + closed = true; + super.closeReadSource(); + } + } + + private interface BatchFiller { + void fill(VectorSchemaRoot root, int batchIndex); + } + + private static byte[] writeIpc(BufferAllocator allocator, Schema schema, int batchCount, BatchFiller filler) + throws Exception { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + try (VectorSchemaRoot root = VectorSchemaRoot.create(schema, allocator); + ArrowStreamWriter writer = new ArrowStreamWriter(root, null, out)) { + writer.start(); + for (int b = 0; b < batchCount; b++) { + filler.fill(root, b); + writer.writeBatch(); + } + writer.end(); + } + return out.toByteArray(); + } + + // single batch, or schema-only stream when no rows are passed + private static byte[] peopleIpc(BufferAllocator allocator, Person... rows) throws Exception { + return writeIpc(allocator, PEOPLE_SCHEMA, rows.length == 0 ? 0 : 1, (root, b) -> { + IntVector id = (IntVector) root.getVector("id"); + VarCharVector name = (VarCharVector) root.getVector("name"); + Float8Vector score = (Float8Vector) root.getVector("score"); + BitVector active = (BitVector) root.getVector("active"); + root.allocateNew(); + for (int i = 0; i < rows.length; i++) { + Person p = rows[i]; + if (p.id == null) { + id.setNull(i); + } else { + id.setSafe(i, p.id); + } + if (p.name == null) { + name.setNull(i); + } else { + name.setSafe(i, p.name.getBytes(UTF_8)); + } + if (p.score == null) { + score.setNull(i); + } else { + score.setSafe(i, p.score); + } + if (p.active == null) { + active.setNull(i); + } else { + active.setSafe(i, p.active ? 1 : 0); + } + } + root.setRowCount(rows.length); + }); + } + + // values are 0 .. batchCount * batchSize - 1 + private static byte[] numbersIpc(BufferAllocator allocator, int batchCount, int batchSize) throws Exception { + return writeIpc(allocator, NUMBERS_SCHEMA, batchCount, (root, b) -> { + BigIntVector vec = (BigIntVector) root.getVector("v"); + vec.allocateNew(batchSize); + for (int i = 0; i < batchSize; i++) { + vec.set(i, (long) b * batchSize + i); + } + root.setRowCount(batchSize); + }); + } + + // reader is owned by the stream after export, it is closed when DuckDB releases the stream + private static void register(DuckDBConnection conn, BufferAllocator allocator, ArrowArrayStream stream, String name, + byte[] ipc) { + ArrowStreamReader reader = new ArrowStreamReader(new ByteArrayInputStream(ipc), allocator); + Data.exportArrayStream(allocator, reader, stream); + conn.registerArrowStream(name, stream); + } + + public static void main(String[] args) throws Exception { + System.exit(runTests(args, TestArrow.class)); + } +} diff --git a/src/test/external/test_arrow.sh b/src/test/external/test_arrow.sh new file mode 100755 index 000000000..783f15761 --- /dev/null +++ b/src/test/external/test_arrow.sh @@ -0,0 +1,30 @@ +#!/usr/bin/env bash +# Runs Arrow tests (src/test/external/TestArrow.java) directly from source, requires Java 17+. +# Arrow JARs listed in data/arrow/arrow_jars.txt are downloaded into .arrow if missing. +# Usage: src/test/external/test_arrow.sh [test_name_filter] + +set -euo pipefail + +# paths are relative to the project root +cd "$(dirname "$0")/../../.." + +BUILD_DIR=build/release +ARROW_DIR=".arrow" + +if [ -n "${JAVA_HOME:-}" ]; then + JAVA="${JAVA_HOME}/bin/java" +else + JAVA=java +fi + +if [ ! -d "${ARROW_DIR}" ] || [ -z "$(ls -A "${ARROW_DIR}")" ]; then + mkdir -p "${ARROW_DIR}" + xargs curl -fsSL --retry 3 --output-dir "${ARROW_DIR}" --remote-name-all < data/arrow/arrow_jars.txt +fi + +"${JAVA}" -version + +exec "${JAVA}" \ + --add-opens=java.base/java.nio=ALL-UNNAMED \ + -cp "${BUILD_DIR}/duckdb_jdbc.jar:${BUILD_DIR}/duckdb_jdbc_tests.jar:${ARROW_DIR}/*" \ + src/test/external/TestArrow.java "$@" diff --git a/src/test/test.iml b/src/test/test.iml deleted file mode 100644 index a0e49a3ba..000000000 --- a/src/test/test.iml +++ /dev/null @@ -1,11 +0,0 @@ - - - - - - - - - - - \ No newline at end of file