From f9e7db16a6333090f35ffd753b6362108e9566b8 Mon Sep 17 00:00:00 2001 From: Shawn Chen Date: Fri, 13 Nov 2026 13:25:54 +1300 Subject: [PATCH 1/2] Close owned Arrow readers on insert validation failure --- chdb-arrow/README.md | 21 ++++++++++ .../java/org/chdb/arrow/ArrowOptions.java | 15 +++++++ .../java/org/chdb/arrow/ArrowSession.java | 41 ++++++++++--------- .../java/org/chdb/arrow/ArrowOptionsTest.java | 19 +++++++++ .../java/org/chdb/arrow/ArrowSessionTest.java | 39 ++++++++++++++++-- 5 files changed, 112 insertions(+), 23 deletions(-) create mode 100644 chdb-arrow/src/test/java/org/chdb/arrow/ArrowOptionsTest.java diff --git a/chdb-arrow/README.md b/chdb-arrow/README.md index 99e9d67..41288b1 100644 --- a/chdb-arrow/README.md +++ b/chdb-arrow/README.md @@ -168,8 +168,29 @@ exports String as UTF-8. Both query forms accept an explicit `ArrowOptions` for String output, dictionary LowCardinality and lossy unsupported-type output. DateTime defaults to Arrow unsigned 32-bit seconds without timezone; use DateTime64 in SQL for timestamp semantics. +Compose named choices from the immutable default, for example +`ArrowOptions.DEFAULT.withLowCardinalityAsDictionary().withStringAsBinary()`. +`withUnsupportedAsBinary()` is deliberately lossy; it does not preserve the +original ClickHouse type. The three-boolean constructor remains available for +existing callers, but the named choices are easier to review at call sites. `registerBatch(root, allocator)` adapts one current `VectorSchemaRoot` batch; the caller retains ownership of the root and must not mutate or close it before unregistering. +`insertOwnedReader` closes an accepted reader even when table/schema validation +or native insertion fails. If the session is closed or a query is active, it +rejects the call before taking ownership. `insertBatch` never owns the root. + +### Current Java interface limits + +There is no export method on an existing JDBC `ResultSet` or `PreparedStatement`. +An Arrow session opens a separate connection, so connection-local state and JDBC +bindings do not carry over. The core has a server-side named-parameter entry +point for *streaming* Arrow queries, but the Java/JNI path does not yet expose +it; the materialized Arrow C entry point has no matching parameter variant. +Do not interpolate untrusted values into SQL as a workaround. Incremental +output also has no caller-selected batch size: `queryArrow` returns core batches +at core's granularity. Adding a Java parameter that promises an exact batch +size would require a core facility or explicit Java-side rechunking and its +associated copy/allocation cost. ## Testing from source diff --git a/chdb-arrow/src/main/java/org/chdb/arrow/ArrowOptions.java b/chdb-arrow/src/main/java/org/chdb/arrow/ArrowOptions.java index 63744f0..6827b91 100644 --- a/chdb-arrow/src/main/java/org/chdb/arrow/ArrowOptions.java +++ b/chdb-arrow/src/main/java/org/chdb/arrow/ArrowOptions.java @@ -19,6 +19,21 @@ public ArrowOptions(boolean unsupportedAsBinary, boolean lowCardinalityAsDiction this.stringAsString = stringAsString; } + /** Opt into lossy binary output for types without a faithful Arrow mapping. */ + public ArrowOptions withUnsupportedAsBinary() { + return new ArrowOptions(true, lowCardinalityAsDictionary, stringAsString); + } + + /** Export LowCardinality values as Arrow dictionaries instead of materializing them. */ + public ArrowOptions withLowCardinalityAsDictionary() { + return new ArrowOptions(unsupportedAsBinary, true, stringAsString); + } + + /** Export ClickHouse String as Arrow binary instead of the default UTF-8. */ + public ArrowOptions withStringAsBinary() { + return new ArrowOptions(unsupportedAsBinary, lowCardinalityAsDictionary, false); + } + int nativeFlags() { return (unsupportedAsBinary ? 1 : 0) | (lowCardinalityAsDictionary ? 2 : 0) diff --git a/chdb-arrow/src/main/java/org/chdb/arrow/ArrowSession.java b/chdb-arrow/src/main/java/org/chdb/arrow/ArrowSession.java index 82267ea..073783e 100644 --- a/chdb-arrow/src/main/java/org/chdb/arrow/ArrowSession.java +++ b/chdb-arrow/src/main/java/org/chdb/arrow/ArrowSession.java @@ -225,8 +225,10 @@ public synchronized ArrowQuery queryArrow( * Inserts a one-shot reader into an existing table without Arrow IPC. Only simple ASCII * table or database.table names are accepted; core receives no arbitrary settings text. * All source columns must be named uniquely, so core cannot silently fall back to - * positional insertion. The reader is owned and closed after schema validation, including - * on query failure. Like ordinary chDB inserts, this is not transactional. + * positional insertion. Once an open session has accepted the call (and has no active + * query), it owns and closes the reader even if validation or the insert fails. A closed + * session or active query leaves ownership with the caller. Like ordinary chDB inserts, + * this is not transactional. * * @return the core-reported rows written */ @@ -240,24 +242,25 @@ public synchronized long insertOwnedReader( throw new IllegalStateException("Arrow session is closed"); } requireNoActiveQuery(); - if (!tableName.matches("[A-Za-z_][A-Za-z0-9_]*(\\.[A-Za-z_][A-Za-z0-9_]*)?")) { - throw new IllegalArgumentException("Expected a simple table or database.table name"); - } - Set columns = new HashSet<>(); - for (org.apache.arrow.vector.types.pojo.Field field : reader.getVectorSchemaRoot() - .getSchema().getFields()) { - if (field.getName() == null || field.getName().isEmpty() - || !columns.add(field.getName())) { - throw new IllegalArgumentException("Arrow insert requires distinct, named columns"); + try (ArrowReader owned = reader) { + if (!tableName.matches("[A-Za-z_][A-Za-z0-9_]*(\\.[A-Za-z_][A-Za-z0-9_]*)?")) { + throw new IllegalArgumentException("Expected a simple table or database.table name"); + } + Set columns = new HashSet<>(); + for (org.apache.arrow.vector.types.pojo.Field field : owned.getVectorSchemaRoot() + .getSchema().getFields()) { + if (field.getName() == null || field.getName().isEmpty() + || !columns.add(field.getName())) { + throw new IllegalArgumentException("Arrow insert requires distinct, named columns"); + } + } + if (columns.isEmpty()) { + throw new IllegalArgumentException("Arrow insert needs at least one column"); + } + try (ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator)) { + Data.exportArrayStream(allocator, owned, stream); + return connection.insertArrowStreamNative(tableName, stream.memoryAddress()); } - } - if (columns.isEmpty()) { - throw new IllegalArgumentException("Arrow insert needs at least one column"); - } - try (ArrowReader owned = reader; - ArrowArrayStream stream = ArrowArrayStream.allocateNew(allocator)) { - Data.exportArrayStream(allocator, owned, stream); - return connection.insertArrowStreamNative(tableName, stream.memoryAddress()); } } diff --git a/chdb-arrow/src/test/java/org/chdb/arrow/ArrowOptionsTest.java b/chdb-arrow/src/test/java/org/chdb/arrow/ArrowOptionsTest.java new file mode 100644 index 0000000..b6d8eba --- /dev/null +++ b/chdb-arrow/src/test/java/org/chdb/arrow/ArrowOptionsTest.java @@ -0,0 +1,19 @@ +package org.chdb.arrow; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import org.junit.jupiter.api.Test; + +class ArrowOptionsTest { + @Test + void namedChoicesComposeWithoutMutatingDefault() { + ArrowOptions options = ArrowOptions.DEFAULT.withLowCardinalityAsDictionary() + .withStringAsBinary().withUnsupportedAsBinary(); + + assertEquals(4, ArrowOptions.DEFAULT.nativeFlags()); + assertEquals(6, ArrowOptions.DEFAULT.withLowCardinalityAsDictionary().nativeFlags()); + assertEquals(0, ArrowOptions.DEFAULT.withStringAsBinary().nativeFlags()); + assertEquals(5, ArrowOptions.DEFAULT.withUnsupportedAsBinary().nativeFlags()); + assertEquals(3, options.nativeFlags()); + } +} diff --git a/chdb-arrow/src/test/java/org/chdb/arrow/ArrowSessionTest.java b/chdb-arrow/src/test/java/org/chdb/arrow/ArrowSessionTest.java index 8846fb9..477e855 100644 --- a/chdb-arrow/src/test/java/org/chdb/arrow/ArrowSessionTest.java +++ b/chdb-arrow/src/test/java/org/chdb/arrow/ArrowSessionTest.java @@ -565,6 +565,37 @@ void failedInsertIntoMissingTableDoesNotConsumeCallerBatch() throws Exception { } } + @Test + void insertOwnedReaderClosesOnValidationFailureButPreservesCallerRoot() throws Exception { + try (RootAllocator allocator = new RootAllocator(); + ArrowSession session = ArrowSession.openInMemory(); + VectorSchemaRoot root = idBatch(allocator, 42)) { + AtomicReference closed = new AtomicReference<>(false); + OneBatchReader reader = new OneBatchReader(allocator, root) { + @Override protected void closeReadSource() { closed.set(true); } + }; + assertThrows(IllegalArgumentException.class, + () -> session.insertOwnedReader("not a table", reader, allocator)); + assertTrue(closed.get(), "the accepted reader must close on invalid table name"); + assertEquals(42L, ((BigIntVector) root.getVector("id")).get(0)); + } + + try (RootAllocator allocator = new RootAllocator(); + ArrowSession session = ArrowSession.openInMemory()) { + BigIntVector first = new BigIntVector("id", allocator); + BigIntVector second = new BigIntVector("id", allocator); + try (VectorSchemaRoot duplicateRoot = VectorSchemaRoot.of(first, second)) { + AtomicReference closed = new AtomicReference<>(false); + OneBatchReader reader = new OneBatchReader(allocator, duplicateRoot) { + @Override protected void closeReadSource() { closed.set(true); } + }; + assertThrows(IllegalArgumentException.class, + () -> session.insertOwnedReader("some_table", reader, allocator)); + assertTrue(closed.get(), "the accepted reader must close on invalid schema"); + } + } + } + @Test @Disabled("core 26.11.0-rc.3 ArrowStream inserts invalid String into Int64 as 0; " + "ordinary INSERT SELECT rejects it with CANNOT_PARSE_TEXT") @@ -627,7 +658,7 @@ void outputTypeOptionsAreExplicitForMaterializedAndStreamedQueries() throws Exce assertTrue(schema.getFields().get(2).getType() instanceof ArrowType.Decimal); assertTrue(defaults.loadNextBatch()); } - ArrowOptions binaryString = new ArrowOptions(false, false, false); + ArrowOptions binaryString = ArrowOptions.DEFAULT.withStringAsBinary(); try (ArrowReader binary = session.queryArrowMaterialized(sql, allocator, binaryString)) { assertTrue(binary.getVectorSchemaRoot().getSchema().getFields().get(0).getType() instanceof ArrowType.Binary); @@ -653,7 +684,7 @@ void dictionaryTimestampAndNestedTypesHaveObservableSchemas() throws Exception { assertTrue(schema.getFields().get(2).getType() instanceof ArrowType.List); } try (ArrowQuery dictionaries = session.queryArrow( - sql, allocator, new ArrowOptions(false, true, true))) { + sql, allocator, ArrowOptions.DEFAULT.withLowCardinalityAsDictionary())) { long id = dictionaries.getVectorSchemaRoot().getSchema().getFields().get(0) .getDictionary().getId(); assertTrue(dictionaries.loadNextBatch()); @@ -671,7 +702,7 @@ void dictionarySchemaIsStableAcrossStreamedBatches() throws Exception { ArrowQuery query = session.queryArrow( "SELECT toLowCardinality(if(number % 2 = 0, 'even', 'odd')) AS lc " + "FROM numbers(200000)", - allocator, new ArrowOptions(false, true, true))) { + allocator, ArrowOptions.DEFAULT.withLowCardinalityAsDictionary())) { Schema initial = query.getVectorSchemaRoot().getSchema(); assertTrue(initial.getFields().get(0).getDictionary() != null); long id = initial.getFields().get(0).getDictionary().getId(); @@ -702,7 +733,7 @@ void exportingIncrementalDictionaryBeforeFirstBatchPreservesValueType() throws E ArrowArrayStream exported = ArrowArrayStream.allocateNew(allocator)) { ArrowQuery query = session.queryArrow( "SELECT CAST('hello', 'LowCardinality(String)') AS lc", - allocator, new ArrowOptions(false, true, true)); + allocator, ArrowOptions.DEFAULT.withLowCardinalityAsDictionary()); // Export transfers reader ownership to the C Stream. getSchema is called // before the stream requests its first batch. Data.exportArrayStream(allocator, query, exported); From d28b9f75d4bd166472043b68f2de4e6749484c7f Mon Sep 17 00:00:00 2001 From: Shawn Chen Date: Fri, 13 Nov 2026 13:32:56 +1300 Subject: [PATCH 2/2] Bind named parameters in streaming Arrow queries --- chdb-arrow/README.md | 28 +++++++++--- .../java/org/chdb/arrow/ArrowSession.java | 23 +++++++++- .../java/org/chdb/arrow/ArrowSessionTest.java | 38 ++++++++++++++++ .../java/org/chdb/internal/ChdbNative.java | 6 ++- .../java/org/chdb/jdbc/ChdbConnection.java | 27 ++++++++++- chdb-jni/src/chdb_jni.cpp | 45 +++++++++++++++++-- chdb-jni/src/chdb_jni_handles.h | 2 +- pom.xml | 2 +- 8 files changed, 156 insertions(+), 15 deletions(-) diff --git a/chdb-arrow/README.md b/chdb-arrow/README.md index 41288b1..33d64f8 100644 --- a/chdb-arrow/README.md +++ b/chdb-arrow/README.md @@ -179,14 +179,32 @@ retains ownership of the root and must not mutate or close it before unregisteri or native insertion fails. If the session is closed or a query is active, it rejects the call before taking ownership. `insertBatch` never owns the root. +Streaming queries can bind core's typed named parameters without SQL interpolation: + +```java +try (RootAllocator allocator = new RootAllocator(); + ArrowSession session = ArrowSession.openInMemory(); + ArrowQuery rows = session.queryArrow("SELECT {id:Int64} AS id", + java.util.Collections.singletonMap("id", "42"), allocator)) { + while (rows.loadNextBatch()) { + System.out.println(rows.getVectorSchemaRoot().getVector("id").getObject(0)); + } +} +``` + +The map's values are strings parsed according to each SQL placeholder's type, +using the JDBC driver's escaped-text encoding. A null value requires a +`Nullable(...)` placeholder. Names must be simple ASCII identifiers. Binding +errors may surface on the first batch fetch; values do not remain on the session. + ### Current Java interface limits There is no export method on an existing JDBC `ResultSet` or `PreparedStatement`. An Arrow session opens a separate connection, so connection-local state and JDBC -bindings do not carry over. The core has a server-side named-parameter entry -point for *streaming* Arrow queries, but the Java/JNI path does not yet expose -it; the materialized Arrow C entry point has no matching parameter variant. -Do not interpolate untrusted values into SQL as a workaround. Incremental +bindings do not carry over. The separate streaming named-parameter overload +does not accept JDBC `?` placeholders or a `PreparedStatement`; the materialized +Arrow C entry point has no matching parameter variant. Do not interpolate +untrusted values into SQL as a workaround. Incremental output also has no caller-selected batch size: `queryArrow` returns core batches at core's granularity. Adding a Java parameter that promises an exact batch size would require a core facility or explicit Java-side rechunking and its @@ -197,7 +215,7 @@ associated copy/allocation cost. The test `ArrowSessionTest` is opt-in (`-Dchdb.arrow.it=true`) and requires a matching engine and JNI shim in the directory given by `-Dchdb.library.path`. The previously published Java `1.0.0-rc.2` platform JAR has JNI ABI 2 and cannot run this work-in-progress -ABI 3 module. Arrow Java 19.0.0 on classpath also requires +ABI 4 module. Arrow Java 19.0.0 on classpath also requires `--add-opens=java.base/java.nio=ALL-UNNAMED` on supported JDKs. The current source pins chDB core `26.11.0-rc.3`. For a macOS arm64 release-asset diff --git a/chdb-arrow/src/main/java/org/chdb/arrow/ArrowSession.java b/chdb-arrow/src/main/java/org/chdb/arrow/ArrowSession.java index 073783e..884af30 100644 --- a/chdb-arrow/src/main/java/org/chdb/arrow/ArrowSession.java +++ b/chdb-arrow/src/main/java/org/chdb/arrow/ArrowSession.java @@ -6,6 +6,7 @@ import java.sql.SQLException; import java.util.ArrayList; import java.util.HashSet; +import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.UUID; @@ -208,15 +209,33 @@ public synchronized ArrowQuery queryArrow(String sql, BufferAllocator allocator) /** Incremental output with explicit core Arrow type mapping choices. */ public synchronized ArrowQuery queryArrow( String sql, BufferAllocator allocator, ArrowOptions options) throws SQLException { + return queryArrow(sql, java.util.Collections.emptyMap(), allocator, options); + } + + /** + * Streams a query with server-side named bindings for SQL {@code {name:Type}} placeholders. + * Values are encoded like JDBC string parameters, not interpolated into SQL. Declare a + * nullable placeholder (e.g. {@code {name:Nullable(String)}}) for a null map value. + * The bindings are scoped to this query; the caller's map is read only during this call. + */ + public synchronized ArrowQuery queryArrow( + String sql, Map parameters, BufferAllocator allocator) throws SQLException { + return queryArrow(sql, parameters, allocator, ArrowOptions.DEFAULT); + } + + /** Named bindings with explicit Arrow output type choices. */ + public synchronized ArrowQuery queryArrow(String sql, Map parameters, + BufferAllocator allocator, ArrowOptions options) throws SQLException { Objects.requireNonNull(sql, "sql"); + Objects.requireNonNull(parameters, "parameters"); Objects.requireNonNull(allocator, "allocator"); Objects.requireNonNull(options, "options"); if (connection == null) { throw new IllegalStateException("Arrow session is closed"); } requireNoActiveQuery(); - ArrowQuery query = connection.arrowStreamOpenNative( - sql, options.nativeFlags(), handle -> new ArrowQuery(connection, allocator, handle)); + ArrowQuery query = connection.arrowStreamOpenNative(sql, parameters, + options.nativeFlags(), handle -> new ArrowQuery(connection, allocator, handle)); queries.add(query); return query; } diff --git a/chdb-arrow/src/test/java/org/chdb/arrow/ArrowSessionTest.java b/chdb-arrow/src/test/java/org/chdb/arrow/ArrowSessionTest.java index 477e855..df66e9f 100644 --- a/chdb-arrow/src/test/java/org/chdb/arrow/ArrowSessionTest.java +++ b/chdb-arrow/src/test/java/org/chdb/arrow/ArrowSessionTest.java @@ -14,6 +14,8 @@ import java.nio.charset.StandardCharsets; import java.util.Arrays; import java.util.List; +import java.util.LinkedHashMap; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -448,6 +450,42 @@ void streamedQueryReadsMultipleBatchesWithoutMaterializingFirst() throws Excepti } } + @Test + void streamedNamedParametersAreBoundAndScopedToOneQuery() throws Exception { + try (RootAllocator allocator = new RootAllocator(); + ArrowSession session = ArrowSession.openInMemory()) { + Map values = new LinkedHashMap<>(); + values.put("id", "42"); + values.put("text", "a\\b\n\t\0' SQL"); + values.put("optional", null); + try (ArrowQuery query = session.queryArrow( + "SELECT {id:Int64} AS id, {text:String} AS text, " + + "{optional:Nullable(String)} AS optional", values, allocator)) { + assertTrue(query.loadNextBatch()); + VectorSchemaRoot root = query.getVectorSchemaRoot(); + assertEquals(42L, ((BigIntVector) root.getVector("id")).get(0)); + assertEquals(values.get("text"), root.getVector("text").getObject(0).toString()); + assertTrue(root.getVector("optional").isNull(0)); + assertFalse(query.loadNextBatch()); + } + try (ArrowQuery next = session.queryArrow("SELECT toInt64(43) AS id", allocator)) { + assertTrue(next.loadNextBatch()); + assertEquals(43L, ((BigIntVector) next.getVectorSchemaRoot().getVector("id")).get(0)); + } + assertThrows(IllegalArgumentException.class, () -> session.queryArrow( + "SELECT 1", java.util.Collections.singletonMap("bad name", "x"), allocator)); + try (ArrowQuery failed = session.queryArrow("SELECT {id:Int64} AS id", + java.util.Collections.singletonMap("id", "not-an-integer"), allocator)) { + assertThrows(IOException.class, failed::loadNextBatch); + } + try (ArrowQuery afterFailure = session.queryArrow("SELECT toInt64(44) AS id", allocator)) { + assertTrue(afterFailure.loadNextBatch()); + assertEquals(44L, ((BigIntVector) afterFailure.getVectorSchemaRoot() + .getVector("id")).get(0)); + } + } + } + @Test void streamedEmptyResultRetainsSchemaAndPartialCloseFreesConnection() throws Exception { try (RootAllocator allocator = new RootAllocator(); diff --git a/chdb-jdbc/src/main/java/org/chdb/internal/ChdbNative.java b/chdb-jdbc/src/main/java/org/chdb/internal/ChdbNative.java index 6605297..1cbc32b 100644 --- a/chdb-jdbc/src/main/java/org/chdb/internal/ChdbNative.java +++ b/chdb-jdbc/src/main/java/org/chdb/internal/ChdbNative.java @@ -28,7 +28,7 @@ public final class ChdbNative { * Must equal {@code chdb_jni::kJniAbiVersion} in the shim and {@code * chdb.jni.abi.version} in the poms. Verified at load time before any other call. */ - public static final int JNI_ABI_VERSION = 3; + public static final int JNI_ABI_VERSION = 4; /** Handle kinds, for {@link #openHandleCount(int)}. */ public static final int KIND_CONNECTION = 1; @@ -141,6 +141,10 @@ public static native void queryArrowMaterialized( /** Starts a lazy Arrow query and returns a registry ID, not a native pointer. */ public static native long arrowStreamOpen(long connection, byte[] sql, int optionFlags); + /** Starts a lazy Arrow query with binary-safe, server-side named parameters. */ + public static native long arrowStreamOpenWithParams(long connection, byte[] sql, + int optionFlags, byte[][] paramNames, byte[][] paramValues); + /** Fills a caller-allocated, initially empty C Stream with one fetched batch or EOF. */ public static native void arrowStreamFetch(long connection, long stream, long outAddress); diff --git a/chdb-jdbc/src/main/java/org/chdb/jdbc/ChdbConnection.java b/chdb-jdbc/src/main/java/org/chdb/jdbc/ChdbConnection.java index ee9ec09..ce948d7 100644 --- a/chdb-jdbc/src/main/java/org/chdb/jdbc/ChdbConnection.java +++ b/chdb-jdbc/src/main/java/org/chdb/jdbc/ChdbConnection.java @@ -20,6 +20,7 @@ import java.util.Collections; import java.util.LinkedHashMap; import java.util.Map; +import java.util.Objects; import java.util.Properties; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -270,6 +271,27 @@ private void arrowRegistrationNative(String name, long address, boolean register /** Opens one streaming query and holds the statement slot until arrowStreamCloseNative. */ public T arrowStreamOpenNative( String sql, int optionFlags, LongFunction ownerFactory) throws SQLException { + return arrowStreamOpenNative(sql, Collections.emptyMap(), optionFlags, ownerFactory); + } + + /** Bind typed {name:Type} placeholders using the same escaped-text encoding as JDBC. */ + public T arrowStreamOpenNative( + String sql, Map parameters, int optionFlags, + LongFunction ownerFactory) throws SQLException { + Objects.requireNonNull(parameters, "parameters"); + byte[][] names = new byte[parameters.size()][]; + byte[][] values = new byte[parameters.size()][]; + int index = 0; + for (Map.Entry entry : parameters.entrySet()) { + String name = Objects.requireNonNull(entry.getKey(), "parameter name"); + if (!name.matches("[A-Za-z_][A-Za-z0-9_]*")) { + throw new IllegalArgumentException("Expected a simple Arrow parameter name: " + name); + } + names[index] = Utf8.encode(name); + values[index] = Utf8.encode(entry.getValue() == null + ? TextEscape.NULL_MARKER : TextEscape.escape(entry.getValue())); + index++; + } statementSlot.acquire(sql); boolean entered = false; boolean handedOff = false; @@ -279,7 +301,10 @@ public T arrowStreamOpenNative( if (!entered || isClosed()) { throw ChdbExceptions.closed("Connection"); } - stream = ChdbNative.arrowStreamOpen(handle, Utf8.encode(sql), optionFlags); + stream = names.length == 0 + ? ChdbNative.arrowStreamOpen(handle, Utf8.encode(sql), optionFlags) + : ChdbNative.arrowStreamOpenWithParams( + handle, Utf8.encode(sql), optionFlags, names, values); T owner = ownerFactory.apply(stream); // Published before the execution gate is released, so close() cannot miss a // stream in the interval between native open and Java owner construction. diff --git a/chdb-jni/src/chdb_jni.cpp b/chdb-jni/src/chdb_jni.cpp index f358734..3b5a80b 100644 --- a/chdb-jni/src/chdb_jni.cpp +++ b/chdb-jni/src/chdb_jni.cpp @@ -616,9 +616,9 @@ std::shared_ptr requireArrowStream(JNIEnv * env, jlong id) } } // namespace -extern "C" JNIEXPORT jlong JNICALL -Java_org_chdb_internal_ChdbNative_arrowStreamOpen( - JNIEnv * env, jclass, jlong connection_id, jbyteArray sql_bytes, jint option_flags) +static jlong arrowStreamOpenImpl(JNIEnv * env, jlong connection_id, + jbyteArray sql_bytes, jint option_flags, jobjectArray param_name_bytes, + jobjectArray param_value_bytes) { try { @@ -628,6 +628,16 @@ Java_org_chdb_internal_ChdbNative_arrowStreamOpen( std::string sql; if (sql_bytes == nullptr || !toBytes(env, sql_bytes, sql)) return 0; + std::vector names; + std::vector values; + if (!toBytesVector(env, param_name_bytes, names) + || !toBytesVector(env, param_value_bytes, values)) + return 0; + if (names.size() != values.size()) + { + throwNative(env, "parameter name and value arrays have different lengths"); + return 0; + } chdb_arrow_options options; if (!arrowOptions(env, option_flags, options)) return 0; @@ -639,7 +649,18 @@ Java_org_chdb_internal_ChdbNative_arrowStreamOpen( throwNative(env, "connection was closed before Arrow stream open"); return 0; } - raw = chdb_stream_query_arrow_n(*connection->conn, sql.data(), sql.size(), &options); + if (names.empty()) + { + raw = chdb_stream_query_arrow_n(*connection->conn, sql.data(), sql.size(), &options); + } + else + { + ParamArrays params = buildParams(names, values); + raw = chdb_stream_query_arrow_with_params_n(*connection->conn, + sql.data(), sql.size(), &options, params.names.data(), + params.name_lengths.data(), params.values.data(), + params.value_lengths.data(), params.names.size()); + } } std::unique_ptr result( raw, &chdb_destroy_query_result); @@ -666,6 +687,22 @@ Java_org_chdb_internal_ChdbNative_arrowStreamOpen( } } +extern "C" JNIEXPORT jlong JNICALL +Java_org_chdb_internal_ChdbNative_arrowStreamOpen( + JNIEnv * env, jclass, jlong connection_id, jbyteArray sql_bytes, jint option_flags) +{ + return arrowStreamOpenImpl(env, connection_id, sql_bytes, option_flags, nullptr, nullptr); +} + +extern "C" JNIEXPORT jlong JNICALL +Java_org_chdb_internal_ChdbNative_arrowStreamOpenWithParams( + JNIEnv * env, jclass, jlong connection_id, jbyteArray sql_bytes, jint option_flags, + jobjectArray param_name_bytes, jobjectArray param_value_bytes) +{ + return arrowStreamOpenImpl(env, connection_id, sql_bytes, option_flags, + param_name_bytes, param_value_bytes); +} + extern "C" JNIEXPORT void JNICALL Java_org_chdb_internal_ChdbNative_arrowStreamFetch( JNIEnv * env, jclass, jlong connection_id, jlong stream_id, jlong out_address) diff --git a/chdb-jni/src/chdb_jni_handles.h b/chdb-jni/src/chdb_jni_handles.h index bca7998..7240f29 100644 --- a/chdb-jni/src/chdb_jni_handles.h +++ b/chdb-jni/src/chdb_jni_handles.h @@ -24,7 +24,7 @@ namespace chdb_jni // Bumped only on an incompatible change to the Java <-> shim contract. Must match // ChdbNative.JNI_ABI_VERSION on the Java side and chdb.jni.abi.version in the poms. -constexpr int32_t kJniAbiVersion = 3; +constexpr int32_t kJniAbiVersion = 4; enum HandleKind : int32_t { diff --git a/pom.xml b/pom.xml index e0123ce..7550e00 100644 --- a/pom.xml +++ b/pom.xml @@ -68,7 +68,7 @@ 26.11.0-rc.3 1 - 3 + 4 1 19.0.0