custom data schema support

Author: alexanderkirtzel

packages/destinations/node/bigquery/src/__tests__/index.test.ts

@@ -86,4 +86,13 @@ describe('Node Destination BigQuery', () => {
       },
     ]);
   });
+
+  test('data', async () => {
+    const config = await getConfig({ projectId, bigquery: { credentials } });
+    const data = { foo: 'bar' };
+    const mockFn = getMockFn(config);
+
+    await destination.push(event, config, {}, { data });
+    expect(mockFn).toHaveBeenCalledWith('insert', [{ foo: 'bar' }]);
+  });
 });

packages/destinations/node/bigquery/src/index.ts

@@ -19,12 +19,12 @@ export const destinationBigQuery: Destination = {
     return config;
   },
 
-  async push(event, config) {
+  async push(event, config, mapping, options) {
     return await tryCatchAsync(push, (error) => {
       if (config.onLog) config.onLog('Push error');
       // @TODO queue handling
       throwError(error);
-    })(event, config);
+    })(event, config, mapping, options);
   },
 };
 

packages/destinations/node/bigquery/src/push.ts

@@ -1,10 +1,12 @@
 import type { WalkerOS } from '@elbwalker/types';
 import type { PushFn, Row } from './types';
+import { isObject } from '@elbwalker/utils';
 
-export const push: PushFn = async function (event, config) {
+export const push: PushFn = async function (event, config, mapping, options) {
   const { client, datasetId, tableId } = config.custom!;
 
-  const rows = [mapEvent(event)];
+  const row = options?.data ? options.data : mapEvent(event);
+  const rows = [row];
 
   await client.dataset(datasetId).table(tableId).insert(rows);
 
@@ -40,5 +42,5 @@ export const mapEvent = (event: WalkerOS.Event): Row => {
 };
 
 function stringify(obj?: WalkerOS.AnyObject): undefined | string {
-  return obj && Object.keys(obj).length ? JSON.stringify(obj) : undefined;
+  return isObject(obj) ? JSON.stringify(obj) : undefined;
 }