diff --git a/javascript/__test__/bind.spec.ts b/javascript/__test__/bind.spec.ts index 2599b52721..a4467ecd06 100644 --- a/javascript/__test__/bind.spec.ts +++ b/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() }) diff --git a/javascript/__test__/ingest.spec.ts b/javascript/__test__/ingest.spec.ts new file mode 100644 index 0000000000..2cfc8c9072 --- /dev/null +++ b/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) +}) diff --git a/javascript/lib/index.ts b/javascript/lib/index.ts index 0215b104a9..baa84326a3 100644 --- a/javascript/lib/index.ts +++ b/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 { + 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 { const stmt = await this.createStatement() try { @@ -306,23 +328,9 @@ export class AdbcStatement implements AdbcStatementInterface { } } - async bind(data: RecordBatch | Table): Promise { + async bind(data: Table): Promise { 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) diff --git a/javascript/lib/types.ts b/javascript/lib/types.ts index 524d98d73b..59fb7487e2 100644 --- a/javascript/lib/types.ts +++ b/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 } +/** + * 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 + /** + * 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 + /** * 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 + bind(data: Table): Promise /** * Close the statement and release resources. diff --git a/javascript/src/client.rs b/javascript/src/client.rs index 3bc2e8dbb3..b6bc9ac1e4 100644 --- a/javascript/src/client.rs +++ b/javascript/src/client.rs @@ -282,23 +282,9 @@ impl AdbcStatementCore { } pub fn bind(&mut self, c_data: Vec) -> 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(()) } }