This is an automated email from the ASF dual-hosted git repository.
lidavidm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-adbc.git
The following commit(s) were added to refs/heads/main by this push:
new 9be029859 feat(java/core): add fluent ingest API (#4466)
9be029859 is described below
commit 9be02985954c29b19ca14be9564f3ad92e63b5d9
Author: David Li <[email protected]>
AuthorDate: Tue Jul 14 00:55:47 2026 -0700
feat(java/core): add fluent ingest API (#4466)
Add a more natural API for bulk ingest that (by default) hides the
statement within, making for a clearer API.
---
java/.checker-framework/java.util.Objects.astub | 25 ++
java/core/pom.xml | 5 +
.../org/apache/arrow/adbc/core/AdbcConnection.java | 10 +-
.../apache/arrow/adbc/core/BulkIngestBuilder.java | 107 +++++++
.../arrow/adbc/core/BulkIngestBuilderImpl.java | 187 ++++++++++++
.../arrow/adbc/core/BulkIngestBuilderTest.java | 326 +++++++++++++++++++++
.../adbc/driver/jni/SqlServerIntegrationTest.java | 76 +++++
7 files changed, 733 insertions(+), 3 deletions(-)
diff --git a/java/.checker-framework/java.util.Objects.astub
b/java/.checker-framework/java.util.Objects.astub
new file mode 100644
index 000000000..f26848295
--- /dev/null
+++ b/java/.checker-framework/java.util.Objects.astub
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package java.util;
+
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+public class Objects
+{
+ public static <T> T requireNonNull(@Nullable T obj);
+}
diff --git a/java/core/pom.xml b/java/core/pom.xml
index 3e30f93f0..a1727382d 100644
--- a/java/core/pom.xml
+++ b/java/core/pom.xml
@@ -36,6 +36,11 @@
<groupId>org.apache.arrow</groupId>
<artifactId>arrow-memory-core</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.arrow</groupId>
+ <artifactId>arrow-memory-netty</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.arrow</groupId>
<artifactId>arrow-vector</artifactId>
diff --git
a/java/core/src/main/java/org/apache/arrow/adbc/core/AdbcConnection.java
b/java/core/src/main/java/org/apache/arrow/adbc/core/AdbcConnection.java
index fef9bfa99..9e260473a 100644
--- a/java/core/src/main/java/org/apache/arrow/adbc/core/AdbcConnection.java
+++ b/java/core/src/main/java/org/apache/arrow/adbc/core/AdbcConnection.java
@@ -17,7 +17,6 @@
package org.apache.arrow.adbc.core;
import java.nio.ByteBuffer;
-import org.apache.arrow.vector.VectorSchemaRoot;
import org.apache.arrow.vector.ipc.ArrowReader;
import org.apache.arrow.vector.types.pojo.Schema;
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -52,7 +51,7 @@ public interface AdbcConnection extends AutoCloseable,
AdbcOptions {
AdbcStatement createStatement() throws AdbcException;
/**
- * Create a new statement to bulk insert a {@link VectorSchemaRoot} into a
table.
+ * Create a new statement to bulk ingest Arrow data into a table.
*
* <p>Bind data to the statement, then call {@link
AdbcStatement#executeUpdate()}. See {@link
* BulkIngestMode} for description of behavior around creating tables.
@@ -64,7 +63,7 @@ public interface AdbcConnection extends AutoCloseable,
AdbcOptions {
}
/**
- * Create a new statement to bulk insert a {@link VectorSchemaRoot} into a
table.
+ * Create a new statement to bulk ingest Arrow data into a table.
*
* <p>Bind data to the statement, then call {@link
AdbcStatement#executeUpdate()}. See {@link
* BulkIngestMode} for description of behavior around creating tables.
@@ -75,6 +74,11 @@ public interface AdbcConnection extends AutoCloseable,
AdbcOptions {
"Connection does not support bulkIngest(String, BulkIngestMode,
IngestOption...)");
}
+ /** Bulk ingest Arrow data into a table, using a fluent builder-style API. */
+ default BulkIngestBuilder bulkIngest() {
+ return new BulkIngestBuilderImpl(this);
+ }
+
/**
* Create a result set from a serialized PartitionDescriptor.
*
diff --git
a/java/core/src/main/java/org/apache/arrow/adbc/core/BulkIngestBuilder.java
b/java/core/src/main/java/org/apache/arrow/adbc/core/BulkIngestBuilder.java
new file mode 100644
index 000000000..4905ac16f
--- /dev/null
+++ b/java/core/src/main/java/org/apache/arrow/adbc/core/BulkIngestBuilder.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.arrow.adbc.core;
+
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
+
+/** A fluent builder-style interface for configuring and executing a bulk
ingest. */
+public interface BulkIngestBuilder extends AutoCloseable {
+ /** Create the target table; fail if already exists. */
+ BulkIngestBuilder createMode() throws AdbcException;
+
+ /** Append to the target table; fail if not found. */
+ BulkIngestBuilder appendMode() throws AdbcException;
+
+ /** Drop (if exists) and create the target table. */
+ BulkIngestBuilder replaceMode() throws AdbcException;
+
+ /** Create the target table if necessary, append if already exists. */
+ BulkIngestBuilder createAppendMode() throws AdbcException;
+
+ /**
+ * How to handle an already-existing/non-existent target table. Default is
{@link
+ * BulkIngestMode#CREATE}.
+ *
+ * <p>Also see the convenience methods {@link #createMode()}, {@link
#appendMode()}, {@link
+ * #replaceMode()}, and {@link #createAppendMode()}.
+ */
+ BulkIngestBuilder mode(BulkIngestMode mode) throws AdbcException;
+
+ /**
+ * Ingest data from the given root.
+ *
+ * <p>Like {@link AdbcStatement#bind(VectorSchemaRoot)}, this ingest will
NOT close the given
+ * root. If a different root or stream is bound, this root will simply be
forgotten.
+ */
+ BulkIngestBuilder bind(VectorSchemaRoot root) throws AdbcException;
+
+ /**
+ * Ingest data from the given reader.
+ *
+ * <p>Like {@link AdbcStatement#bind(ArrowReader)}, this ingest WILL close
the given stream. If a
+ * different root or stream is bound, this stream will be closed.
+ */
+ BulkIngestBuilder bind(ArrowReader stream) throws AdbcException;
+
+ /** Ingest into a table with the given name (required). */
+ BulkIngestBuilder targetTable(String table) throws AdbcException;
+
+ /** Ingest into a table in the given schema. */
+ BulkIngestBuilder targetSchema(String schema) throws AdbcException;
+
+ /** Ingest into a table in the given catalog. */
+ BulkIngestBuilder targetCatalog(String catalog) throws AdbcException;
+
+ /** Ingest into a temporary table. */
+ default BulkIngestBuilder temporary() throws AdbcException {
+ return this.temporary(true);
+ }
+
+ /** Whether to ingest into a temporary table or not. */
+ BulkIngestBuilder temporary(boolean isTemporary) throws AdbcException;
+
+ /** Set an arbitrary ingest option. */
+ BulkIngestBuilder option(IngestOption option) throws AdbcException;
+
+ /** Set a statement option before ingest. */
+ <T> BulkIngestBuilder option(TypedKey<T> key, T value) throws AdbcException;
+
+ /**
+ * Convert this builder to an {@link AdbcStatement} (low-level API for
further control) without
+ * executing.
+ *
+ * <p>The caller is responsible for closing the returned statement.
+ */
+ AdbcStatement toStatement() throws AdbcException;
+
+ /**
+ * Execute the bulk ingest.
+ *
+ * <p>This will consume the bound stream, if any. To ingest again, bind a
new stream or root or
+ * update data in the already-bound root.
+ */
+ default AdbcStatement.UpdateResult ingest() throws AdbcException {
+ try (var statement = toStatement()) {
+ return statement.executeUpdate();
+ }
+ }
+
+ /** Clean up any state in this bulk ingest. */
+ @Override
+ void close() throws AdbcException;
+}
diff --git
a/java/core/src/main/java/org/apache/arrow/adbc/core/BulkIngestBuilderImpl.java
b/java/core/src/main/java/org/apache/arrow/adbc/core/BulkIngestBuilderImpl.java
new file mode 100644
index 000000000..91daaec10
--- /dev/null
+++
b/java/core/src/main/java/org/apache/arrow/adbc/core/BulkIngestBuilderImpl.java
@@ -0,0 +1,187 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.arrow.adbc.core;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+class BulkIngestBuilderImpl implements BulkIngestBuilder {
+ private final AdbcConnection connection;
+ BulkIngestMode mode = BulkIngestMode.CREATE;
+ @Nullable VectorSchemaRoot root;
+ @Nullable ArrowReader reader;
+ @Nullable String targetTable, targetSchema, targetCatalog;
+ boolean temporary = false;
+ List<IngestOption> ingestOptions = new ArrayList<>();
+ List<Map.Entry<TypedKey<?>, Object>> options = new ArrayList<>();
+
+ BulkIngestBuilderImpl(AdbcConnection connection) {
+ this.connection = Objects.requireNonNull(connection, "connection must not
be null");
+ }
+
+ @Override
+ public BulkIngestBuilder createMode() {
+ mode = BulkIngestMode.CREATE;
+ return this;
+ }
+
+ @Override
+ public BulkIngestBuilder appendMode() {
+ mode = BulkIngestMode.APPEND;
+ return this;
+ }
+
+ @Override
+ public BulkIngestBuilder replaceMode() {
+ mode = BulkIngestMode.REPLACE;
+ return this;
+ }
+
+ @Override
+ public BulkIngestBuilder createAppendMode() {
+ mode = BulkIngestMode.CREATE_APPEND;
+ return this;
+ }
+
+ @Override
+ public BulkIngestBuilder mode(BulkIngestMode mode) {
+ this.mode = mode;
+ return this;
+ }
+
+ @Override
+ public BulkIngestBuilder bind(VectorSchemaRoot root) throws AdbcException {
+ this.root = root;
+ if (this.reader != null) {
+ try {
+ this.reader.close();
+ } catch (IOException e) {
+ throw AdbcException.io("failed to close existing
ArrowReader").withCause(e);
+ }
+ }
+ this.reader = null;
+ return this;
+ }
+
+ @Override
+ public BulkIngestBuilder bind(ArrowReader stream) throws AdbcException {
+ root = null;
+ reader = stream;
+ return this;
+ }
+
+ @Override
+ public BulkIngestBuilder targetTable(String table) throws AdbcException {
+ targetTable = table;
+ return this;
+ }
+
+ @Override
+ public BulkIngestBuilder targetSchema(String schema) throws AdbcException {
+ targetSchema = schema;
+ return this;
+ }
+
+ @Override
+ public BulkIngestBuilder targetCatalog(String catalog) throws AdbcException {
+ targetCatalog = catalog;
+ return this;
+ }
+
+ @Override
+ public BulkIngestBuilder temporary(boolean isTemporary) throws AdbcException
{
+ temporary = isTemporary;
+ return this;
+ }
+
+ @Override
+ public BulkIngestBuilder option(IngestOption option) throws AdbcException {
+ ingestOptions.add(option);
+ return this;
+ }
+
+ @Override
+ public <T> BulkIngestBuilder option(TypedKey<T> key, T value) throws
AdbcException {
+ // redundant cast - make ErrorProne happy
+ options.add(Map.entry(key, Objects.requireNonNull((Object) value)));
+ return this;
+ }
+
+ @Override
+ public AdbcStatement toStatement() throws AdbcException {
+ AdbcStatement statement = null;
+ try {
+ if (targetTable == null) {
+ throw AdbcException.invalidArgument("must set targetTable");
+ }
+
+ List<IngestOption> ingestOptions = new ArrayList<>(this.ingestOptions);
+ if (temporary) {
+ ingestOptions.add(IngestOption.TEMPORARY);
+ }
+ if (targetCatalog != null) {
+ ingestOptions.add(IngestOption.targetCatalog(targetCatalog));
+ }
+ if (targetSchema != null) {
+ ingestOptions.add(IngestOption.targetDbSchema(targetSchema));
+ }
+ IngestOption[] varargs = ingestOptions.toArray(new IngestOption[0]);
+ statement = connection.bulkIngest(Objects.requireNonNull(targetTable),
mode, varargs);
+
+ if (root != null) {
+ statement.bind(root);
+ root = null;
+ } else if (reader != null) {
+ statement.bind(reader);
+ // Prevent reader from being closed below; it's now owned by the
statement
+ reader = null;
+ } else {
+ throw AdbcException.invalidArgument(
+ "must bind either a VectorSchemaRoot or an ArrowReader");
+ }
+
+ for (Map.Entry<TypedKey<?>, Object> option : options) {
+ //noinspection unchecked
+ statement.setOption((TypedKey<Object>) option.getKey(),
option.getValue());
+ }
+ } catch (Exception e) {
+ if (statement != null) {
+ statement.close();
+ }
+ throw e;
+ }
+ return statement;
+ }
+
+ @Override
+ public void close() throws AdbcException {
+ // Do not close root
+ if (reader != null) {
+ try {
+ reader.close();
+ } catch (IOException e) {
+ throw AdbcException.io("failed to close ArrowReader").withCause(e);
+ }
+ }
+ }
+}
diff --git
a/java/core/src/test/java/org/apache/arrow/adbc/core/BulkIngestBuilderTest.java
b/java/core/src/test/java/org/apache/arrow/adbc/core/BulkIngestBuilderTest.java
new file mode 100644
index 000000000..cbefafe0a
--- /dev/null
+++
b/java/core/src/test/java/org/apache/arrow/adbc/core/BulkIngestBuilderTest.java
@@ -0,0 +1,326 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.arrow.adbc.core;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.arrow.vector.types.Types;
+import org.apache.arrow.vector.types.pojo.Field;
+import org.apache.arrow.vector.types.pojo.Schema;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class BulkIngestBuilderTest {
+ Schema schema = new Schema(List.of(Field.nullable("foo",
Types.MinorType.INT.getType())));
+ BufferAllocator allocator;
+ TestConnection connection;
+ VectorSchemaRoot root;
+
+ @BeforeEach
+ void beforeEach() {
+ allocator = new RootAllocator();
+ connection = new TestConnection();
+ root = VectorSchemaRoot.create(schema, allocator);
+ }
+
+ @AfterEach
+ void afterEach() {
+ root.close();
+ connection.close();
+ allocator.close();
+ }
+
+ @Test
+ void noTable() throws Exception {
+ try (var ingest = connection.bulkIngest()) {
+ assertThatThrownBy(ingest::ingest)
+ .hasMessageContaining("must set targetTable")
+ .isInstanceOf(AdbcException.class);
+ }
+ }
+
+ @Test
+ void noRoot() throws Exception {
+ try (var ingest = connection.bulkIngest()) {
+ ingest.targetTable("foobar");
+ assertThatThrownBy(ingest::ingest)
+ .hasMessageContaining("must bind")
+ .isInstanceOf(AdbcException.class);
+ }
+ }
+
+ @Test
+ void standardOptions() throws Exception {
+ try (var ingest = connection.bulkIngest()) {
+ ingest
+ .targetTable("foobar")
+ .bind(root)
+ .temporary(true)
+ .targetCatalog("catalog")
+ .targetSchema("schema")
+ .createAppendMode()
+ .ingest();
+ }
+ assertThat(connection.targetTable).isEqualTo("foobar");
+ assertThat(connection.mode).isEqualTo(BulkIngestMode.CREATE_APPEND);
+ assertThat(connection.options)
+ .anyMatch(
+ option ->
+ (option instanceof IngestOption.TargetNamespaceIngestOption)
+ && Objects.equals(
+ ((IngestOption.TargetNamespaceIngestOption)
option).getTargetDbSchema(),
+ "schema"));
+ assertThat(connection.options)
+ .anyMatch(
+ option ->
+ (option instanceof IngestOption.TargetNamespaceIngestOption)
+ && Objects.equals(
+ ((IngestOption.TargetNamespaceIngestOption)
option).getTargetCatalog(),
+ "catalog"));
+ assertThat(connection.options)
+ .anyMatch(
+ option ->
+ (option instanceof IngestOption.TemporaryIngestOption)
+ && ((IngestOption.TemporaryIngestOption)
option).isTemporary());
+ assertThat(connection.customOptions).isEmpty();
+ }
+
+ @Test
+ void customOptions() throws Exception {
+ var key = new TypedKey<>("foobar", Integer.class);
+ try (var ingest = connection.bulkIngest()) {
+ ingest
+ .targetTable("foobar")
+ .bind(root)
+ .option(new CustomIngestOption())
+ .option(key, 64)
+ .ingest();
+ }
+ assertThat(connection.targetTable).isEqualTo("foobar");
+ assertThat(connection.mode).isEqualTo(BulkIngestMode.CREATE);
+
+ assertThat(connection.options).anyMatch(option -> option instanceof
CustomIngestOption);
+ assertThat(connection.customOptions).containsExactly(Map.entry(key, 64));
+ }
+
+ @Test
+ void closesReader() throws Exception {
+ assertThatThrownBy(
+ () -> {
+ try (var ingest = connection.bulkIngest()) {
+ ingest.bind(new TestArrowReader(allocator));
+ }
+ })
+ .isInstanceOf(AdbcException.class)
+ .hasMessageContaining("failed to close ArrowReader")
+ .hasCauseInstanceOf(IOException.class)
+ .cause()
+ .hasMessageContaining("expected");
+
+ assertThatThrownBy(
+ () -> {
+ try (var ingest = connection.bulkIngest()) {
+ ingest.bind(new TestArrowReader(allocator));
+ ingest.bind(root);
+ }
+ })
+ .isInstanceOf(AdbcException.class)
+ .hasMessageContaining("failed to close existing ArrowReader")
+ .hasCauseInstanceOf(IOException.class)
+ .cause()
+ .hasMessageContaining("expected");
+ }
+
+ @Test
+ void noCloseRoot() throws Exception {
+ try (var ingest = connection.bulkIngest()) {
+ ingest.bind(root);
+ }
+ root.setRowCount(5);
+ assertThat(root.contentToTSVString()).isNotEmpty();
+ }
+
+ @Test
+ void exclusiveRootReader() throws Exception {
+ try (var ingest = connection.bulkIngest()) {
+ ingest.targetTable("foobar");
+ ingest.bind(root);
+ ingest.bind(new NoOpArrowReader(allocator));
+ ingest.ingest();
+ }
+ assertThat(connection.reader).isNotNull();
+ assertThat(connection.root).isNull();
+ }
+
+ @Test
+ void exclusiveReaderRoot() throws Exception {
+ try (var ingest = connection.bulkIngest()) {
+ ingest.targetTable("foobar");
+ ingest.bind(new NoOpArrowReader(allocator));
+ ingest.bind(root);
+ ingest.ingest();
+ }
+ assertThat(connection.reader).isNull();
+ assertThat(connection.root).isNotNull();
+ }
+
+ static class TestConnection implements AdbcConnection {
+ String targetTable;
+ BulkIngestMode mode;
+ IngestOption[] options;
+ List<Map.Entry<TypedKey<?>, Object>> customOptions = new ArrayList<>();
+ VectorSchemaRoot root;
+ ArrowReader reader;
+
+ @Override
+ public AdbcStatement bulkIngest(
+ String targetTableName, BulkIngestMode mode, IngestOption... options) {
+ var stmt = new TestStatement(this);
+ this.targetTable = targetTableName;
+ this.mode = mode;
+ this.options = options;
+ return stmt;
+ }
+
+ @Override
+ public AdbcStatement createStatement() throws AdbcException {
+ return new TestStatement(this);
+ }
+
+ @Override
+ public ArrowReader getInfo(int @Nullable [] infoCodes) throws
AdbcException {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void close() {
+ if (reader != null) {
+ try {
+ reader.close();
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+ }
+
+ static class TestStatement implements AdbcStatement {
+ private final TestConnection connection;
+
+ public TestStatement(TestConnection testConnection) {
+ this.connection = testConnection;
+ }
+
+ @Override
+ public void bind(ArrowReader reader) throws AdbcException {
+ connection.reader = reader;
+ }
+
+ @Override
+ public void bind(VectorSchemaRoot root) throws AdbcException {
+ connection.root = root;
+ }
+
+ @Override
+ public QueryResult executeQuery() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public UpdateResult executeUpdate() {
+ return new UpdateResult(-1);
+ }
+
+ @Override
+ public void prepare() {}
+
+ @Override
+ public void close() {}
+
+ @Override
+ public <T> void setOption(TypedKey<T> key, T value) throws AdbcException {
+ connection.customOptions.add(Map.entry(key, value));
+ }
+ }
+
+ static class CustomIngestOption implements IngestOption {}
+
+ static class TestArrowReader extends ArrowReader {
+ TestArrowReader(BufferAllocator allocator) {
+ super(allocator);
+ }
+
+ @Override
+ public boolean loadNextBatch() throws IOException {
+ return false;
+ }
+
+ @Override
+ public long bytesRead() {
+ return 0;
+ }
+
+ @Override
+ protected void closeReadSource() throws IOException {}
+
+ @Override
+ protected Schema readSchema() throws IOException {
+ return null;
+ }
+
+ @Override
+ public void close() throws IOException {
+ throw new IOException("expected");
+ }
+ }
+
+ static class NoOpArrowReader extends ArrowReader {
+ NoOpArrowReader(BufferAllocator allocator) {
+ super(allocator);
+ }
+
+ @Override
+ public boolean loadNextBatch() throws IOException {
+ return false;
+ }
+
+ @Override
+ public long bytesRead() {
+ return 0;
+ }
+
+ @Override
+ protected void closeReadSource() throws IOException {}
+
+ @Override
+ protected Schema readSchema() throws IOException {
+ return null;
+ }
+ }
+}
diff --git
a/java/driver/jni-validation/src/test/java/org/apache/arrow/adbc/driver/jni/SqlServerIntegrationTest.java
b/java/driver/jni-validation/src/test/java/org/apache/arrow/adbc/driver/jni/SqlServerIntegrationTest.java
index e895411a3..e059f92fa 100644
---
a/java/driver/jni-validation/src/test/java/org/apache/arrow/adbc/driver/jni/SqlServerIntegrationTest.java
+++
b/java/driver/jni-validation/src/test/java/org/apache/arrow/adbc/driver/jni/SqlServerIntegrationTest.java
@@ -322,6 +322,7 @@ class SqlServerIntegrationTest {
void bulkIngestTarget() throws Exception {
runSetup(
"DROP TABLE IF EXISTS secondary.foobar",
+ "DROP TABLE IF EXISTS secondary.spam",
"DROP SCHEMA IF EXISTS secondary",
"CREATE SCHEMA secondary");
@@ -393,6 +394,81 @@ class SqlServerIntegrationTest {
}
}
+ @Test
+ void bulkIngestFluent() throws Exception {
+ final Schema schema =
+ new Schema(
+ List.of(
+ Field.nullable("ndx", Types.MinorType.INT.getType()),
+ Field.nullable("value", Types.MinorType.VARCHAR.getType())));
+ try (VectorSchemaRoot vsr = VectorSchemaRoot.create(schema, allocator)) {
+ IntVector iv = (IntVector) vsr.getVector(0);
+ VarCharVector vv = (VarCharVector) vsr.getVector(1);
+
+ try (var ingest = conn.bulkIngest().targetTable("spam").temporary()) {
+ iv.setSafe(0, 1);
+ iv.setSafe(1, 2);
+ vv.setNull(0);
+ vv.setSafe(1, "foobar".getBytes(StandardCharsets.UTF_8));
+ vsr.setRowCount(2);
+
+ ingest.bind(vsr);
+ assertThat(ingest.ingest().getAffectedRows()).isEqualTo(2);
+ }
+ }
+
+ try (AdbcStatement stmt = conn.createStatement()) {
+ stmt.setSqlQuery("SELECT value FROM #spam ORDER BY ndx");
+ try (var result = stmt.executeQuery()) {
+ var values = ArrowToJava.toStrings(result.getReader(), "value");
+ assertThat(values).containsExactly(null, "foobar");
+ }
+ }
+ }
+
+ @Test
+ void bulkIngestFluentLowLevel() throws Exception {
+ runSetup(
+ "DROP TABLE IF EXISTS secondary.foobar",
+ "DROP TABLE IF EXISTS secondary.spam",
+ "DROP SCHEMA IF EXISTS secondary",
+ "CREATE SCHEMA secondary");
+
+ final Schema schema =
+ new Schema(
+ List.of(
+ Field.nullable("ndx", Types.MinorType.INT.getType()),
+ Field.nullable("value", Types.MinorType.VARCHAR.getType())));
+ try (VectorSchemaRoot vsr = VectorSchemaRoot.create(schema, allocator)) {
+ IntVector iv = (IntVector) vsr.getVector(0);
+ VarCharVector vv = (VarCharVector) vsr.getVector(1);
+
+ var key = new TypedKey<>("adbc.ingest.target_db_schema", String.class);
+ try (var ingest =
+ conn.bulkIngest()
+ .targetTable("spam")
+ .option(key, "secondary")
+ .option(IngestOption.NOT_TEMPORARY)) {
+ iv.setSafe(0, 1);
+ iv.setSafe(1, 2);
+ vv.setNull(0);
+ vv.setSafe(1, "foobar".getBytes(StandardCharsets.UTF_8));
+ vsr.setRowCount(2);
+
+ ingest.bind(vsr);
+ assertThat(ingest.ingest().getAffectedRows()).isEqualTo(2);
+ }
+ }
+
+ try (AdbcStatement stmt = conn.createStatement()) {
+ stmt.setSqlQuery("SELECT value FROM secondary.spam ORDER BY ndx");
+ try (var result = stmt.executeQuery()) {
+ var values = ArrowToJava.toStrings(result.getReader(), "value");
+ assertThat(values).containsExactly(null, "foobar");
+ }
+ }
+ }
+
@Test
void currentCatalogSchema() throws Exception {
runSetup(
