feat: add convenience API for bulk ingest

javascript/__test__/bind.spec.ts

@@ -70,17 +70,40 @@ test('statement: bind and query data', async () => {
   assert.strictEqual(rowCount, 1)
 })
 
-test('statement: bind multi-batch table throws descriptive error', async () => {
-  // Both batches must share the same schema instance for Table to accept them
+test('statement: bind empty table inserts 0 rows', async () => {
+  const empty = tableFromArrays({ id: [] as number[], name: [] as string[] })
+  assert.strictEqual(empty.numRows, 0)
+
+  const stmt2 = await conn.createStatement()
+  await stmt2.setSqlQuery('INSERT INTO bind_test (id, name) VALUES (?, ?)')
+  await stmt2.bind(empty)
+  const affected = await stmt2.executeUpdate()
+  assert.strictEqual(affected, 0)
+  await stmt2.close()
+})
+
+test('statement: bind single-batch table', async () => {
+  const data = tableFromArrays({ id: [100], name: ['single'] })
+  assert.strictEqual(data.batches.length, 1)
+
+  const stmt2 = await conn.createStatement()
+  await stmt2.setSqlQuery('INSERT INTO bind_test (id, name) VALUES (?, ?)')
+  await stmt2.bind(data)
+  const affected = await stmt2.executeUpdate()
+  assert.strictEqual(affected, 1)
+  await stmt2.close()
+})
+
+test('statement: bind multi-batch table', async () => {
   const base = tableFromArrays({ id: [10], name: ['first'] })
   const batch1 = base.batches[0]
-  const batch2 = base.batches[0] // same schema, reused to construct a multi-batch Table
-  const multiTable = new Table([batch1, batch2])
+  const multiTable = new Table([batch1, batch1])
   assert.strictEqual(multiTable.batches.length, 2)
 
   const stmt2 = await conn.createStatement()
-  const error = await stmt2.bind(multiTable).catch((e) => e)
-  assert.ok(error instanceof Error)
-  assert.match(error.message, /bind\(\).*batches|batches.*bind\(\)/i)
+  await stmt2.setSqlQuery('INSERT INTO bind_test (id, name) VALUES (?, ?)')
+  await stmt2.bind(multiTable)
+  const affected = await stmt2.executeUpdate()
+  assert.strictEqual(affected, 2)
   await stmt2.close()
 })

javascript/__test__/ingest.spec.ts

@@ -0,0 +1,101 @@
+// 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.
+
+import { test, before, after } from 'node:test'
+import assert from 'node:assert/strict'
+import { createSqliteDatabase } from './test_utils'
+import { AdbcDatabase, AdbcConnection, IngestMode } from '../lib/index.js'
+import { tableFromArrays, Table } from 'apache-arrow'
+
+let db: AdbcDatabase
+let conn: AdbcConnection
+
+before(async () => {
+  db = await createSqliteDatabase()
+  conn = await db.connect()
+})
+
+after(async () => {
+  try {
+    await conn?.close()
+    await db?.close()
+  } catch {
+    // ignore
+  }
+})
+
+test('ingest: create mode inserts data into a new table', async () => {
+  const data = tableFromArrays({ id: [1, 2, 3], name: ['alice', 'bob', 'carol'] })
+  const rowCount = await conn.ingest('ingest_create', data)
+  assert.strictEqual(rowCount, 3)
+
+  const result = await conn.query('SELECT id, name FROM ingest_create ORDER BY id')
+  assert.strictEqual(result.numRows, 3)
+  assert.strictEqual(result.getChildAt(0)?.get(0), 1)
+  assert.strictEqual(result.getChildAt(1)?.get(0), 'alice')
+})
+
+test('ingest: create mode fails if table already exists', async () => {
+  const data = tableFromArrays({ id: [1] })
+  await conn.ingest('ingest_create_dup', data)
+  await assert.rejects(() => conn.ingest('ingest_create_dup', data))
+})
+
+test('ingest: append mode adds rows to an existing table', async () => {
+  const initial = tableFromArrays({ id: [1], name: ['alice'] })
+  await conn.ingest('ingest_append', initial)
+
+  const more = tableFromArrays({ id: [2], name: ['bob'] })
+  const rowCount = await conn.ingest('ingest_append', more, { mode: IngestMode.Append })
+  assert.strictEqual(rowCount, 1)
+
+  const result = await conn.query('SELECT id FROM ingest_append ORDER BY id')
+  assert.strictEqual(result.numRows, 2)
+})
+
+test('ingest: replace mode drops and recreates the table', async () => {
+  const initial = tableFromArrays({ id: [1, 2, 3] })
+  await conn.ingest('ingest_replace', initial)
+
+  const replacement = tableFromArrays({ id: [99] })
+  await conn.ingest('ingest_replace', replacement, { mode: IngestMode.Replace })
+
+  const result = await conn.query('SELECT id FROM ingest_replace')
+  assert.strictEqual(result.numRows, 1)
+  assert.strictEqual(result.getChildAt(0)?.get(0), 99)
+})
+
+test('ingest: multi-batch table inserts all batches', async () => {
+  const batch = tableFromArrays({ id: [1], name: ['alice'] }).batches[0]
+  const data = new Table([batch, batch])
+  assert.strictEqual(data.batches.length, 2)
+
+  const rowCount = await conn.ingest('ingest_multi_batch', data)
+  assert.strictEqual(rowCount, 2)
+
+  const result = await conn.query('SELECT id FROM ingest_multi_batch')
+  assert.strictEqual(result.numRows, 2)
+})
+
+test('ingest: create_append mode creates table if not exists then appends', async () => {
+  const data = tableFromArrays({ id: [1] })
+  await conn.ingest('ingest_create_append', data, { mode: IngestMode.CreateAppend })
+  await conn.ingest('ingest_create_append', data, { mode: IngestMode.CreateAppend })
+
+  const result = await conn.query('SELECT id FROM ingest_create_append')
+  assert.strictEqual(result.numRows, 2)
+})

javascript/lib/index.ts

@@ -23,10 +23,11 @@ import type {
   AdbcStatement as AdbcStatementInterface,
   ConnectOptions,
   GetObjectsOptions,
+  IngestOptions,
 } from './types.js'
-import { LoadFlags, ObjectDepth, InfoCode } from './types.js'
+import { LoadFlags, ObjectDepth, InfoCode, IngestMode } from './types.js'
 
-import { RecordBatchReader, RecordBatch, Table, tableToIPC, Schema } from 'apache-arrow'
+import { RecordBatch, RecordBatchReader, Table, tableToIPC, Schema } from 'apache-arrow'
 import { AdbcError } from './error.js'
 
 // Safely define Symbol.asyncDispose for compatibility with Node.js environments older than v21.
@@ -75,8 +76,8 @@ async function iteratorToReader(iterator: NativeIterator): Promise<RecordBatchRe
 }
 
 // Export Options types, constants, and Error class
-export type { ConnectOptions, GetObjectsOptions }
-export { AdbcError, LoadFlags, ObjectDepth, InfoCode }
+export type { ConnectOptions, GetObjectsOptions, IngestOptions }
+export { AdbcError, LoadFlags, ObjectDepth, InfoCode, IngestMode }
 
 /**
  * Represents an ADBC Database.
@@ -214,6 +215,27 @@ export class AdbcConnection implements AdbcConnectionInterface {
     }
   }
 
+  async ingest(tableName: string, data: Table, options?: IngestOptions): Promise<number> {
+    const stmt = await this.createStatement()
+    try {
+      stmt.setOption('adbc.ingest.target_table', tableName)
+      stmt.setOption('adbc.ingest.mode', options?.mode ?? IngestMode.Create)
+      if (options?.catalog !== undefined) {
+        stmt.setOption('adbc.ingest.target_catalog', options.catalog)
+      }
+      if (options?.dbSchema !== undefined) {
+        stmt.setOption('adbc.ingest.target_db_schema', options.dbSchema)
+      }
+      if (options?.temporary === true) {
+        stmt.setOption('adbc.ingest.temporary', 'true')
+      }
+      await stmt.bind(data)
+      return await stmt.executeUpdate()
+    } finally {
+      await stmt.close()
+    }
+  }
+
   async execute(sql: string, params?: Table): Promise<number> {
     const stmt = await this.createStatement()
     try {
@@ -306,23 +328,9 @@ export class AdbcStatement implements AdbcStatementInterface {
     }
   }
 
-  async bind(data: RecordBatch | Table): Promise<void> {
+  async bind(data: Table): Promise<void> {
     try {
-      let table: Table
-      if (data instanceof Table) {
-        table = data
-      } else {
-        table = new Table(data)
-      }
-
-      if (table.batches.length > 1) {
-        throw new Error(
-          `bind() requires a single-batch Table or RecordBatch, but received ${table.batches.length} batches. ` +
-            `Concatenate the table into one batch first (e.g. tableFromArrays(...)).`,
-        )
-      }
-
-      const ipcBytes = tableToIPC(table, 'stream')
+      const ipcBytes = tableToIPC(data, 'stream')
       await this._inner.bind(Buffer.from(ipcBytes))
     } catch (e) {
       throw AdbcError.fromError(e)

javascript/lib/types.ts

@@ -15,7 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
-import { RecordBatch, RecordBatchReader, Table, Schema } from 'apache-arrow'
+import { RecordBatchReader, Table, Schema } from 'apache-arrow'
 
 /**
  * Bitmask flags controlling how the driver manager resolves a driver name.
@@ -144,6 +144,41 @@ export interface ConnectOptions {
   databaseOptions?: Record<string, string>
 }
 
+/**
+ * Ingestion modes for the `ingest` convenience method.
+ *
+ * These correspond to the `adbc.ingest.mode.*` option values in the ADBC spec.
+ *
+ * @example
+ * await conn.ingest('my_table', data, { mode: IngestMode.Append })
+ */
+export const IngestMode = {
+  /** Append to an existing table. Fails if the table does not exist. */
+  Append: 'adbc.ingest.mode.append',
+  /** Create a new table and insert. Fails if the table already exists. */
+  Create: 'adbc.ingest.mode.create',
+  /** Create the table if it does not exist, then append. */
+  CreateAppend: 'adbc.ingest.mode.create_append',
+  /** Drop the existing table (if any) and recreate it, then insert. */
+  Replace: 'adbc.ingest.mode.replace',
+} as const
+export type IngestMode = (typeof IngestMode)[keyof typeof IngestMode]
+
+/** Options for the `ingest` convenience method. */
+export interface IngestOptions {
+  /**
+   * How to handle an existing table.
+   * Defaults to {@link IngestMode.Create}.
+   */
+  mode?: IngestMode
+  /** The catalog to create/locate the target table in (optional). */
+  catalog?: string
+  /** The database schema to create/locate the target table in (optional). */
+  dbSchema?: string
+  /** Whether to ingest into a temporary table (optional). */
+  temporary?: boolean
+}
+
 /** Options for getObjects metadata call. */
 export interface GetObjectsOptions {
   /**
@@ -282,6 +317,20 @@ export interface AdbcConnection {
    */
   queryStream(sql: string, params?: Table): Promise<RecordBatchReader>
 
+  /**
+   * Ingest Arrow data into a database table.
+   *
+   * Convenience method that sets the ingestion options, binds the data, and
+   * calls executeUpdate. Depending on the driver, this can avoid per-row
+   * overhead compared to a prepare-bind-insert loop.
+   *
+   * @param tableName The target table name.
+   * @param data Arrow Table to ingest.
+   * @param options Ingestion options (mode, catalog, dbSchema, temporary).
+   * @returns A Promise resolving to the number of rows ingested, or -1 if unknown.
+   */
+  ingest(tableName: string, data: Table, options?: IngestOptions): Promise<number>
+
   /**
    * Execute a SQL statement (INSERT, UPDATE, DELETE, DDL) and return the row count.
    *
@@ -352,12 +401,11 @@ export interface AdbcStatement {
   /**
    * Bind parameters or data for ingestion.
    *
-   * This binds an Arrow RecordBatch or Table to the statement.
    * This is used for bulk ingestion or parameterized queries.
    *
-   * @param data Arrow RecordBatch or Table containing the data to bind.
+   * @param data Arrow Table containing the data to bind.
    */
-  bind(data: RecordBatch | Table): Promise<void>
+  bind(data: Table): Promise<void>
 
   /**
    * Close the statement and release resources.

javascript/src/client.rs

@@ -282,23 +282,9 @@ impl AdbcStatementCore {
   }
 
   pub fn bind(&mut self, c_data: Vec<u8>) -> Result<()> {
-    let mut reader =
+    let reader =
       StreamReader::try_new(std::io::Cursor::new(c_data), None).map_err(ClientError::Arrow)?;
-    let batch = match reader.next() {
-      Some(Ok(b)) => b,
-      Some(Err(e)) => return Err(ClientError::Arrow(e)),
-      None => {
-        return Err(ClientError::Other(
-          "bind() received an empty record batch stream".to_string(),
-        ))
-      }
-    };
-    if reader.next().is_some() {
-      return Err(ClientError::Other(
-        "bind() received multiple record batches; concatenate into one batch first".to_string(),
-      ));
-    }
-    self.inner.bind(batch)?;
+    self.inner.bind_stream(Box::new(reader))?;
     Ok(())
   }
 }