From 1428bca521ffeb6c9827f6acbfa8d0e6e4bfbc9c Mon Sep 17 00:00:00 2001 From: ohmayr Date: Tue, 20 Oct 2026 22:36:45 +0000 Subject: [PATCH 1/2] feat(bigquery): allow WRITE_APPEND in createQueryArrowStream Remove the guard that rejected WRITE_APPEND. With explicit destination tables, the stream reads the destination through a read session, so a WRITE_APPEND query returns the existing rows plus the appended ones. This matches query(), which also reads back the whole destination table. Replace the rejection tests with a unit test that WRITE_APPEND reaches the jobs.insert configuration, and a system test that WRITE_APPEND returns existing and new rows while WRITE_EMPTY fails on a non-empty table and leaves it unchanged. --- handwritten/bigquery/src/arrow.ts | 7 --- handwritten/bigquery/system-test/arrow.ts | 59 +++++++++++++-------- handwritten/bigquery/test/arrow.ts | 64 +++++++++++++++++------ 3 files changed, 85 insertions(+), 45 deletions(-) diff --git a/handwritten/bigquery/src/arrow.ts b/handwritten/bigquery/src/arrow.ts index 780451a686f..b1eb7d7af86 100644 --- a/handwritten/bigquery/src/arrow.ts +++ b/handwritten/bigquery/src/arrow.ts @@ -909,13 +909,6 @@ export function createQueryArrowStream_( throw new Error('maxResults must be a non-negative integer.'); } - if (queryObj.writeDisposition === 'WRITE_APPEND') { - throw new Error( - 'WRITE_APPEND is not supported with Arrow query results because reading ' + - 'the destination table via the Storage Read API would include pre-existing rows.', - ); - } - const ctx: ArrowStreamContext = { bq, arrow, diff --git a/handwritten/bigquery/system-test/arrow.ts b/handwritten/bigquery/system-test/arrow.ts index 7bc5fad2341..1cbcb81f675 100644 --- a/handwritten/bigquery/system-test/arrow.ts +++ b/handwritten/bigquery/system-test/arrow.ts @@ -138,28 +138,6 @@ describe('Arrow Query Results', () => { assert.deepStrictEqual(rows, [{x: 2}]); }); - it('should reject WRITE_APPEND before running the query and leave the destination table unchanged', async () => { - // Reading a WRITE_APPEND destination would also return the rows already - // in the table, so the stream rejects it before sending the query. - const tableRef = `\`${dataset.id}.append_target\``; - await bigquery.query(`CREATE TABLE ${tableRef} AS SELECT 1 AS x`); - - await assert.rejects( - bigquery - .createQueryArrowStream({ - query: 'SELECT 2 AS x', - destination: dataset.table('append_target'), - writeDisposition: 'WRITE_APPEND', - }) - .toArray(), - /WRITE_APPEND is not supported with Arrow query results/, - ); - - // Verify the query never ran: the table still holds only its original row. - const [rows] = await bigquery.query(`SELECT x FROM ${tableRef}`); - assert.deepStrictEqual(rows, [{x: 1}]); - }); - it('should raise the same error as query() when a query fails during execution', async () => { const sql = 'SELECT 1 / 0'; let structErr: (Error & {code?: number}) | undefined; @@ -809,4 +787,41 @@ describe('Arrow Query Results', () => { .toArray(); assert.strictEqual(new arrow.Table(allBatches).numRows, totalRows); }); + + it('should return existing and new rows for WRITE_APPEND and fail WRITE_EMPTY on a non-empty destination table', async () => { + const tableRef = `\`${dataset.id}.append_target\``; + const destTable = dataset.table('append_target'); + await bigquery.query(`CREATE TABLE ${tableRef} AS SELECT 1 AS x`); + + // 1. WRITE_APPEND adds the new row and, like query(), reads back the whole + // destination table, including the row that was already there. + const appendBatches: arrow.RecordBatch[] = await bigquery + .createQueryArrowStream({ + query: 'SELECT 2 AS x', + destination: destTable, + writeDisposition: 'WRITE_APPEND', + }) + .toArray(); + const appendTable = new arrow.Table(appendBatches); + assert.deepStrictEqual( + Array.from(appendTable.getChild('x')!.toArray()).sort(), + [1n, 2n], + ); + + // 2. WRITE_EMPTY fails because the destination table already has rows. + await assert.rejects( + bigquery + .createQueryArrowStream({ + query: 'SELECT 3 AS x', + destination: destTable, + writeDisposition: 'WRITE_EMPTY', + }) + .toArray(), + /Already Exists/, + ); + + // Verify the failed WRITE_EMPTY job left the table unchanged. + const [rows] = await bigquery.query(`SELECT x FROM ${tableRef} ORDER BY x`); + assert.deepStrictEqual(rows, [{x: 1}, {x: 2}]); + }); }); diff --git a/handwritten/bigquery/test/arrow.ts b/handwritten/bigquery/test/arrow.ts index 91e62f21947..18e25122139 100644 --- a/handwritten/bigquery/test/arrow.ts +++ b/handwritten/bigquery/test/arrow.ts @@ -523,22 +523,6 @@ describe('Arrow Support', () => { assert.strictEqual(batches[0].numCols, 0); }); - it('should fail before sending the query when writeDisposition is WRITE_APPEND', async () => { - const getCapturedReq = stubJobsQuery(bq, {}); - - // Verify WRITE_APPEND is rejected before executing the query on BigQuery. - await assert.rejects( - bq - .createQueryArrowStream({ - query: 'SELECT 1', - writeDisposition: 'WRITE_APPEND', - }) - .toArray(), - /WRITE_APPEND is not supported with Arrow query results/, - ); - assert.strictEqual(getCapturedReq(), undefined); - }); - it('should fail when a SQL query string is not provided', async () => { // Verify missing query string emits an error on the stream. await assert.rejects( @@ -1839,6 +1823,54 @@ describe('Arrow Support', () => { assert.strictEqual(batches[0].numRows, 2); }); + it('should send WRITE_APPEND through createQueryJob and read the whole destination table through a read session', async () => { + const {destination, getCapturedJobOpts} = stubDestinationJob( + 'append_tbl', + 'dest-append-1', + ); + const sessionReqs: CreateReadSessionRequest[] = []; + stubStorageReadClient(bq, { + createReadSession: async req => { + sessionReqs.push(req); + return [ + { + arrowSchema: SAMPLE_READ_ROWS_RESPONSE.arrowSchema, + streams: [ + { + name: 'projects/test-project/locations/US/sessions/sess-2/streams/stream-0', + }, + ], + }, + ]; + }, + readRows: () => + fakeReadRowsStream([ + {arrowRecordBatch: SAMPLE_READ_ROWS_RESPONSE.arrowRecordBatch}, + ]), + }); + + const batches: arrow.RecordBatch[] = await bq + .createQueryArrowStream({ + query: 'SELECT id, name FROM tbl', + destination, + writeDisposition: 'WRITE_APPEND', + }) + .toArray(); + + // Verify WRITE_APPEND is sent to jobs.insert instead of being rejected. + const queryConfig = getCapturedJobOpts()?.configuration?.query; + assert.strictEqual(queryConfig?.writeDisposition, 'WRITE_APPEND'); + + // Verify the stream reads the destination table, which holds the + // existing rows plus the appended ones. + assert.deepStrictEqual( + sessionReqs.map(req => req.readSession.table), + ['projects/test-project/datasets/dest_ds/tables/append_tbl'], + ); + assert.strictEqual(batches.length, 1); + assert.strictEqual(batches[0].numRows, 2); + }); + it('should emit a 0-row RecordBatch preserving session.arrowSchema when createReadSession returns no streams', async () => { const {destination} = stubDestinationJob('empty_tbl', 'dest-empty-1'); From 2dbd77b5f6e79206d3b07c3ffcaa162df8157373 Mon Sep 17 00:00:00 2001 From: ohmayr Date: Wed, 21 Oct 2026 01:17:00 +0000 Subject: [PATCH 2/2] docs(bigquery): note WRITE_APPEND reads back existing Arrow rows With WRITE_APPEND the destination table also holds rows from before the query. Say so in the createQueryArrowStream JSDoc, as query() behaves the same, and limit the read-session comment about row order to a table the query just wrote. --- handwritten/bigquery/src/arrow.ts | 3 ++- handwritten/bigquery/src/bigquery.ts | 4 +++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/handwritten/bigquery/src/arrow.ts b/handwritten/bigquery/src/arrow.ts index 44e61c1bcb2..8701835099e 100644 --- a/handwritten/bigquery/src/arrow.ts +++ b/handwritten/bigquery/src/arrow.ts @@ -484,7 +484,8 @@ async function createArrowReadSession( dataFormat: 'ARROW', readOptions: {arrowSerializationOptions: ctx.arrowSerializationOptions}, }, - // One stream keeps rows in the query's order (e.g., ORDER BY). + // One stream keeps rows in the query's order (e.g., ORDER BY) for a table + // the query just wrote. maxStreamCount: 1, }); return session ?? undefined; diff --git a/handwritten/bigquery/src/bigquery.ts b/handwritten/bigquery/src/bigquery.ts index 3d0e5e86e76..ced33b8cc1a 100644 --- a/handwritten/bigquery/src/bigquery.ts +++ b/handwritten/bigquery/src/bigquery.ts @@ -403,7 +403,9 @@ export class BigQuery extends Service { * `for await` does this for you. * * With an explicit `destination` table, the stream reads that whole table - * through a Storage Read API session once the query finishes. + * through a Storage Read API session once the query finishes. With + * `writeDisposition: 'WRITE_APPEND'`, that includes rows the table already + * had, as with {@link BigQuery#query}. * * @param {string|ArrowQuery} [options] A string SQL query or configuration * object. See {@link BigQuery#query} for a complete list of options.