Improve Arrow Java ownership and add streaming named parameters

chdb-arrow/README.md

@@ -168,15 +168,54 @@ 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.
+
+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 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
+associated copy/allocation cost.
 
 ## Testing from source
 
 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

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)

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<String, String> 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<String, String> 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;
     }
@@ -225,8 +244,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 +261,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<String> 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<String> 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());
         }
     }
 

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());
+    }
+}

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<String, String> 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();
@@ -565,6 +603,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<Boolean> 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<Boolean> 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 +696,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 +722,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 +740,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 +771,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);

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);