From ced084ded5c2f1f33720252e9f82db6459a26fb0 Mon Sep 17 00:00:00 2001 From: Kishan Parmar <135701940+kishanprmr@users.noreply.github.com> Date: Wed, 30 Sep 2026 10:41:54 +0530 Subject: [PATCH 1/3] feat(pieces): add Insert Multiple Rows to Postgres, MySQL and SQL Server (#15946) --- bun.lock | 10 +- .../microsoft-sql-server/package.json | 2 +- .../microsoft-sql-server/src/index.ts | 2 + .../src/lib/actions/insert-rows.ts | 318 ++++++++++++++++++ .../src/lib/common/index.ts | 39 +++ .../src/lib/common/props.ts | 1 + .../src/lib/output-schemas.ts | 12 + packages/pieces/community/mysql/package.json | 2 +- .../community/mysql/src/lib/actions/index.ts | 2 + .../mysql/src/lib/actions/insert-rows.ts | 242 +++++++++++++ .../community/mysql/src/lib/common/index.ts | 22 ++ .../community/mysql/src/lib/output-schemas.ts | 12 + .../pieces/community/postgres/package.json | 2 +- .../pieces/community/postgres/src/index.ts | 3 +- .../postgres/src/lib/actions/insert-rows.ts | 244 ++++++++++++++ .../community/postgres/src/lib/common.ts | 18 + .../postgres/src/lib/output-schemas.ts | 11 + 17 files changed, 933 insertions(+), 9 deletions(-) create mode 100644 packages/pieces/community/microsoft-sql-server/src/lib/actions/insert-rows.ts create mode 100644 packages/pieces/community/mysql/src/lib/actions/insert-rows.ts create mode 100644 packages/pieces/community/postgres/src/lib/actions/insert-rows.ts diff --git a/bun.lock b/bun.lock index ce07ab442ba4..e31cb1320404 100644 --- a/bun.lock +++ b/bun.lock @@ -3954,7 +3954,7 @@ }, "packages/pieces/community/google-slides": { "name": "@activepieces/piece-google-slides", - "version": "0.2.0", + "version": "0.3.0", "dependencies": { "@activepieces/core-piece-types": "workspace:*", "@activepieces/core-utils": "workspace:*", @@ -5886,7 +5886,7 @@ }, "packages/pieces/community/microsoft-sql-server": { "name": "@activepieces/piece-microsoft-sql-server", - "version": "0.0.3", + "version": "0.1.0", "dependencies": { "@activepieces/core-piece-types": "workspace:*", "@activepieces/core-utils": "workspace:*", @@ -6233,7 +6233,7 @@ }, "packages/pieces/community/mysql": { "name": "@activepieces/piece-mysql", - "version": "0.2.7", + "version": "0.3.0", "dependencies": { "@activepieces/core-piece-types": "workspace:*", "@activepieces/core-utils": "workspace:*", @@ -7109,7 +7109,7 @@ }, "packages/pieces/community/postgres": { "name": "@activepieces/piece-postgres", - "version": "0.3.1", + "version": "0.4.0", "dependencies": { "@activepieces/core-piece-types": "workspace:*", "@activepieces/core-utils": "workspace:*", @@ -8386,7 +8386,7 @@ }, "packages/pieces/community/slack": { "name": "@activepieces/piece-slack", - "version": "0.20.0", + "version": "0.21.0", "dependencies": { "@activepieces/core-piece-types": "workspace:*", "@activepieces/core-utils": "workspace:*", diff --git a/packages/pieces/community/microsoft-sql-server/package.json b/packages/pieces/community/microsoft-sql-server/package.json index 6f8cc6e39903..4807527ba878 100644 --- a/packages/pieces/community/microsoft-sql-server/package.json +++ b/packages/pieces/community/microsoft-sql-server/package.json @@ -1,6 +1,6 @@ { "name": "@activepieces/piece-microsoft-sql-server", - "version": "0.0.3", + "version": "0.1.0", "main": "./dist/src/index.js", "types": "./dist/src/index.d.ts", "dependencies": { diff --git a/packages/pieces/community/microsoft-sql-server/src/index.ts b/packages/pieces/community/microsoft-sql-server/src/index.ts index e9745bbeb792..d88019ec8faa 100644 --- a/packages/pieces/community/microsoft-sql-server/src/index.ts +++ b/packages/pieces/community/microsoft-sql-server/src/index.ts @@ -2,6 +2,7 @@ import { PieceCategory, createPiece } from '@activepieces/pieces-framework'; import { mssqlAuth } from './lib/auth'; import { findRowsAction } from './lib/actions/find-rows'; import { insertRowAction } from './lib/actions/insert-row'; +import { insertRowsAction } from './lib/actions/insert-rows'; import { updateRowsAction } from './lib/actions/update-rows'; import { deleteRowsAction } from './lib/actions/delete-rows'; import { getTablesAction } from './lib/actions/get-tables'; @@ -20,6 +21,7 @@ export const mssql = createPiece({ actions: [ findRowsAction, insertRowAction, + insertRowsAction, updateRowsAction, deleteRowsAction, getTablesAction, diff --git a/packages/pieces/community/microsoft-sql-server/src/lib/actions/insert-rows.ts b/packages/pieces/community/microsoft-sql-server/src/lib/actions/insert-rows.ts new file mode 100644 index 000000000000..abf33f9a6850 --- /dev/null +++ b/packages/pieces/community/microsoft-sql-server/src/lib/actions/insert-rows.ts @@ -0,0 +1,318 @@ +import { createAction, InputPropertyMap, Property } from '@activepieces/pieces-framework'; +import sql from 'mssql'; +import { mssqlAuth } from '../auth'; +import { mssqlCommon, MssqlColumnDetail } from '../common'; +import { cursorUtils } from '../common/cursor'; +import { mssqlProps } from '../common/props'; +import { insertRowsActionOutputSchema } from '../output-schemas'; + +export const insertRowsAction = createAction({ + auth: mssqlAuth, + name: 'insert_rows', + classification: 'WRITE', + displayName: 'Insert Multiple Rows', + description: 'Inserts many rows into a table in a single transaction', + audience: 'both', + aiMetadata: { + description: + 'Inserts an array of rows into a SQL Server table, each row a map of column names to values, using multi-row INSERT statements inside one transaction: if any row fails, none are inserted. Columns a row omits get the column default. Returns the inserted row count and, when Return Inserted Rows is on, the stored rows including IDENTITY ids. Use instead of repeated single inserts for bulk loads. Not idempotent: each call appends new rows, so repeating it adds duplicates or fails on a unique-key collision.', + idempotent: false, + }, + props: { + table: mssqlProps.table(), + batch: Property.DynamicProperties({ + auth: mssqlAuth, + displayName: 'Rows', + required: true, + refreshers: ['table'], + props: async ({ auth, table }): Promise => { + const target = mssqlProps.asTable(table); + if (!auth || target === null) { + return {}; + } + const pool = await mssqlCommon.connect({ auth }); + try { + const columns = await mssqlCommon.getColumnDetails({ pool, table: target }); + return { + rows: Property.Array({ + displayName: 'Rows', + description: + 'One item per row. Leave a field empty to use the column default.', + required: true, + properties: Object.fromEntries( + columns.map((column) => [ + column.name, + Property.ShortText({ + displayName: column.name, + description: describeColumn({ column }), + required: isRequiredColumn({ column }), + }), + ]) + ), + }), + }; + } finally { + await pool.close().catch(() => undefined); + } + }, + }), + keep_empty_strings: Property.Checkbox({ + displayName: 'Keep Empty Strings', + description: + 'Insert empty fields as empty strings instead of the column default.', + required: false, + defaultValue: false, + }), + return_rows: Property.Checkbox({ + displayName: 'Return Inserted Rows', + description: + 'Return the inserted rows, including identity ids. Fails on tables with triggers.', + required: false, + defaultValue: false, + }), + }, + outputSchema: insertRowsActionOutputSchema, + async run(context) { + const { table, batch, keep_empty_strings, return_rows } = context.propsValue; + const records = insertRowsUtils.toRecords({ + rows: batch?.['rows'], + keepEmptyStrings: keep_empty_strings === true, + }); + const columns = insertRowsUtils.collectColumns({ records }); + const target = mssqlCommon.quoteTable(table); + + const pool = await mssqlCommon.connect({ auth: context.auth }); + pool.on('error', () => undefined); + try { + const output = + return_rows === true + ? `OUTPUT ${cursorUtils.exactProjection({ + columns: (await mssqlCommon.getTableMeta({ pool, table })).columns, + prefix: 'INSERTED', + })}` + : ''; + const statements = insertRowsUtils.buildStatements({ + quotedTable: target, + columns, + records, + output, + }); + const results = await runInTransaction({ pool, statements, target }); + const rowCount = results.reduce( + (total, result) => total + sumOf({ counts: result.rowsAffected }), + 0 + ); + return return_rows === true + ? { rows: results.flatMap((result) => result.recordset ?? []), row_count: rowCount } + : { row_count: rowCount }; + } finally { + await pool.close().catch(() => undefined); + } + }, +}); + +async function runInTransaction({ + pool, + statements, + target, +}: { + pool: sql.ConnectionPool; + statements: InsertStatement[]; + target: string; +}): Promise>[]> { + const transaction = new sql.Transaction(pool); + await transaction.begin(); + try { + const results: sql.IResult>[] = []; + for (const statement of statements) { + const request = transaction.request(); + statement.values.forEach((value, index) => + request.input(`p${index}`, value) + ); + results.push(await request.query>(statement.text)); + } + await transaction.commit(); + return results; + } catch (error) { + await transaction.rollback().catch(() => undefined); + if (mssqlCommon.isOutputBlockedByTrigger(error)) { + throw new Error( + `${target} has enabled triggers, so SQL Server refuses to return the inserted rows. Turn off Return Inserted Rows and the batch will insert normally.` + ); + } + const reason = error instanceof Error ? error.message : String(error); + throw new Error( + `Insert failed and the whole batch was rolled back: ${reason}` + ); + } +} + +function sumOf({ counts }: { counts: number[] | undefined }): number { + return (counts ?? []).reduce((total, count) => total + count, 0); +} + +function isRequiredColumn({ column }: { column: MssqlColumnDetail }): boolean { + const isFilledByDatabase = + column.hasDefault || + column.isIdentity || + column.isComputed || + column.typeName === 'timestamp'; + return !column.isNullable && !isFilledByDatabase; +} + +function describeColumn({ column }: { column: MssqlColumnDetail }): string { + return isRequiredColumn({ column }) + ? `${column.typeName} · required` + : column.typeName; +} + +function toRecords({ + rows, + keepEmptyStrings, +}: { + rows: unknown; + keepEmptyStrings: boolean; +}): Record[] { + if (!Array.isArray(rows) || rows.length === 0) { + throw new Error('Rows must be a non-empty array of objects.'); + } + return rows.map((row, index) => { + if (typeof row !== 'object' || row === null || Array.isArray(row)) { + throw new Error( + `Row ${index + 1} must be an object of column names mapped to values.` + ); + } + return Object.fromEntries( + Object.entries(row).filter( + ([, value]) => value !== undefined && (keepEmptyStrings || value !== '') + ) + ); + }); +} + +function collectColumns({ + records, +}: { + records: Record[]; +}): string[] { + const columns = [ + ...new Set(records.flatMap((record) => Object.keys(record))), + ]; + if (columns.length === 0) { + throw new Error('Rows must contain at least one column.'); + } + if (columns.length > MAX_PARAMETERS) { + throw new Error( + `Rows use ${columns.length} columns, but SQL Server allows at most ${MAX_PARAMETERS} parameters per statement.` + ); + } + return columns; +} + +function toBindValue(value: unknown): unknown { + const isStructured = + typeof value === 'object' && + value !== null && + !(value instanceof Date) && + !Buffer.isBuffer(value); + return isStructured ? JSON.stringify(value) : value; +} + +function buildStatements({ + quotedTable, + columns, + records, + output, +}: { + quotedTable: string; + columns: string[]; + records: Record[]; + output: string; +}): InsertStatement[] { + const rowsPerStatement = Math.min( + MAX_ROWS_PER_STATEMENT, + Math.floor(MAX_PARAMETERS / columns.length) + ); + const quotedColumns = columns + .map((column) => mssqlCommon.quoteId(column)) + .join(', '); + + return splitIntoBatches({ records, maxRows: rowsPerStatement }).map((batch) => { + const values: unknown[] = []; + const tuples = batch.map((record) => { + const placeholders = columns.map((column) => { + const value = record[column]; + if (value === undefined) { + return 'DEFAULT'; + } + values.push(toBindValue(value) ?? null); + return `@p${values.length - 1}`; + }); + return `(${placeholders.join(', ')})`; + }); + return { + text: `INSERT INTO ${quotedTable} (${quotedColumns}) ${output} VALUES ${tuples.join(', ')}`, + values, + }; + }); +} + +function splitIntoBatches({ + records, + maxRows, +}: { + records: Record[]; + maxRows: number; +}): Record[][] { + const batches: Record[][] = []; + let current: Record[] = []; + let currentBytes = 0; + for (const record of records) { + const bytes = estimateBytes({ record }); + const isFull = + current.length >= maxRows || currentBytes + bytes > MAX_STATEMENT_BYTES; + if (current.length > 0 && isFull) { + batches.push(current); + current = []; + currentBytes = 0; + } + current.push(record); + currentBytes += bytes; + } + return current.length > 0 ? [...batches, current] : batches; +} + +function estimateBytes({ record }: { record: Record }): number { + return Object.values(record).reduce( + (total, value) => total + sizeOf({ value }), + 0 + ); +} + +function sizeOf({ value }: { value: unknown }): number { + if (typeof value === 'string') { + return Buffer.byteLength(value); + } + if (Buffer.isBuffer(value)) { + return value.length; + } + if (typeof value === 'object' && value !== null && !(value instanceof Date)) { + return Buffer.byteLength(JSON.stringify(value)); + } + return 16; +} + +const MAX_PARAMETERS = 2098; +const MAX_ROWS_PER_STATEMENT = 1000; +const MAX_STATEMENT_BYTES = 1024 * 1024; + +export const insertRowsUtils = { + toRecords, + collectColumns, + buildStatements, +}; + +export type InsertStatement = { + text: string; + values: unknown[]; +}; diff --git a/packages/pieces/community/microsoft-sql-server/src/lib/common/index.ts b/packages/pieces/community/microsoft-sql-server/src/lib/common/index.ts index 2fad7f1ce3c0..7b940048a305 100644 --- a/packages/pieces/community/microsoft-sql-server/src/lib/common/index.ts +++ b/packages/pieces/community/microsoft-sql-server/src/lib/common/index.ts @@ -276,6 +276,35 @@ async function getTableMeta({ }; } +async function getColumnDetails({ + pool, + table, +}: { + pool: sql.ConnectionPool; + table: MssqlTable; +}): Promise { + const result = await pool + .request() + .input('schema', table.table_schema) + .input('name', table.table_name) + .query>( + `SELECT c.name, TYPE_NAME(c.system_type_id) AS type_name, c.is_nullable, + c.is_identity, c.is_computed, + CASE WHEN c.default_object_id <> 0 THEN 1 ELSE 0 END AS has_default + FROM sys.columns c + WHERE c.object_id = OBJECT_ID(QUOTENAME(@schema) + '.' + QUOTENAME(@name)) + ORDER BY c.column_id` + ); + return result.recordset.map((row) => ({ + name: String(row['name']), + typeName: String(row['type_name']).toLowerCase(), + isNullable: Boolean(row['is_nullable']), + isIdentity: Boolean(row['is_identity']), + isComputed: Boolean(row['is_computed']), + hasDefault: Boolean(row['has_default']), + })); +} + function bindParameters({ request, parameters, @@ -320,6 +349,7 @@ export const mssqlCommon = { connect, getTables, getTableMeta, + getColumnDetails, bindParameters, writeReturningRows, }; @@ -346,3 +376,12 @@ export type MssqlTableMeta = { identity?: string; keyColumns: string[]; }; + +export type MssqlColumnDetail = { + name: string; + typeName: string; + isNullable: boolean; + isIdentity: boolean; + isComputed: boolean; + hasDefault: boolean; +}; diff --git a/packages/pieces/community/microsoft-sql-server/src/lib/common/props.ts b/packages/pieces/community/microsoft-sql-server/src/lib/common/props.ts index 895317cf7439..3750e952e9ff 100644 --- a/packages/pieces/community/microsoft-sql-server/src/lib/common/props.ts +++ b/packages/pieces/community/microsoft-sql-server/src/lib/common/props.ts @@ -111,4 +111,5 @@ export const warningMarkdown = Property.MarkDown({ export const mssqlProps = { table, column, + asTable, }; diff --git a/packages/pieces/community/microsoft-sql-server/src/lib/output-schemas.ts b/packages/pieces/community/microsoft-sql-server/src/lib/output-schemas.ts index f76d50ae8e25..d26f2d2b8b9a 100644 --- a/packages/pieces/community/microsoft-sql-server/src/lib/output-schemas.ts +++ b/packages/pieces/community/microsoft-sql-server/src/lib/output-schemas.ts @@ -52,6 +52,18 @@ export const insertRowActionOutputSchema: OutputSchema = { ], }; +export const insertRowsActionOutputSchema: OutputSchema = { + fields: [ + { + key: 'rows', + label: 'Inserted Rows', + description: + 'The inserted rows as stored by the database, including IDENTITY ids and defaults; each row\'s fields are the columns of the target table. Only present when Return Inserted Rows is enabled.', + }, + { key: 'row_count', label: 'Inserted Row Count', format: 'number' }, + ], +}; + export const newOrUpdatedRowTriggerOutputSchema: OutputSchema = { fields: [ { diff --git a/packages/pieces/community/mysql/package.json b/packages/pieces/community/mysql/package.json index 06a2431c3889..bf14ab4ea892 100644 --- a/packages/pieces/community/mysql/package.json +++ b/packages/pieces/community/mysql/package.json @@ -1,6 +1,6 @@ { "name": "@activepieces/piece-mysql", - "version": "0.2.7", + "version": "0.3.0", "main": "./dist/src/index.js", "types": "./dist/src/index.d.ts", "dependencies": { diff --git a/packages/pieces/community/mysql/src/lib/actions/index.ts b/packages/pieces/community/mysql/src/lib/actions/index.ts index eef26ac9bbce..e467b8df0fc5 100644 --- a/packages/pieces/community/mysql/src/lib/actions/index.ts +++ b/packages/pieces/community/mysql/src/lib/actions/index.ts @@ -1,6 +1,7 @@ import executeQuery from './execute-query'; import getTables from './get-tables'; import insertRow from './insert-row'; +import { insertRows } from './insert-rows'; import updateRows from './update-row'; import deleteRows from './delete-row'; import selectRows from './find-rows'; @@ -8,6 +9,7 @@ import selectRows from './find-rows'; export default [ selectRows, insertRow, + insertRows, updateRows, deleteRows, getTables, diff --git a/packages/pieces/community/mysql/src/lib/actions/insert-rows.ts b/packages/pieces/community/mysql/src/lib/actions/insert-rows.ts new file mode 100644 index 000000000000..65d52dae96a4 --- /dev/null +++ b/packages/pieces/community/mysql/src/lib/actions/insert-rows.ts @@ -0,0 +1,242 @@ +import { createAction, InputPropertyMap, Property } from '@activepieces/pieces-framework'; +import { Connection } from 'promise-mysql'; +import { mysqlAuth } from '../..'; +import { MysqlColumn, mysqlCommon, mysqlConnect, mysqlGetColumns, sanitizeColumnName } from '../common'; +import { insertRowsOutputSchema } from '../output-schemas'; + +export const insertRows = createAction({ + auth: mysqlAuth, + name: 'insert_rows', + classification: 'WRITE', + displayName: 'Insert Multiple Rows', + description: 'Inserts many rows into a table in a single transaction', + audience: 'both', + outputSchema: insertRowsOutputSchema, + aiMetadata: { + description: 'Inserts an array of rows into a MySQL table, each row a map of column names to values, using multi-row INSERT statements inside one transaction: if any row fails, none are inserted (InnoDB tables only). Columns a row omits get the column default. Returns the inserted row count and the auto-increment id of the first inserted row. Use instead of repeated single inserts for bulk loads. Not idempotent: each call appends new rows, so repeating it adds duplicates or errors on a unique-key collision.', + idempotent: false, + }, + props: { + timezone: mysqlCommon.timezone, + table: mysqlCommon.table(), + batch: Property.DynamicProperties({ + auth: mysqlAuth, + displayName: 'Rows', + required: true, + refreshers: ['table'], + props: async ({ auth, table }): Promise => { + if (!auth || typeof table !== 'string' || table === '') { + return {}; + } + const conn = await mysqlConnect(auth, {}); + try { + const columns = await mysqlGetColumns({ conn, table }); + return { + rows: Property.Array({ + displayName: 'Rows', + description: 'One item per row. Leave a field empty to use the column default.', + required: true, + properties: Object.fromEntries( + columns.map((column) => [ + column.column_name, + Property.ShortText({ + displayName: column.column_name, + description: describeColumn({ column }), + required: isRequiredColumn({ column }), + }), + ]), + ), + }), + }; + } finally { + await conn.end(); + } + }, + }), + keep_empty_strings: Property.Checkbox({ + displayName: 'Keep Empty Strings', + description: 'Insert empty fields as empty strings instead of the column default.', + required: false, + defaultValue: false, + }), + }, + async run(context) { + const { table, batch, keep_empty_strings } = context.propsValue; + const records = insertRowsUtils.toRecords({ rows: batch?.['rows'], keepEmptyStrings: keep_empty_strings === true }); + const columns = insertRowsUtils.collectColumns({ records }); + const statements = insertRowsUtils.buildStatements({ + quotedTable: sanitizeColumnName(table), + columns, + records, + }); + + const conn = await mysqlConnect(context.auth, context.propsValue); + conn.on('error', () => undefined); + try { + await assertTransactionalTable({ conn, table }); + return await insertInTransaction({ conn, statements }); + } finally { + await conn.end().catch(() => conn.destroy()); + } + }, +}); + +async function assertTransactionalTable({ conn, table }: { conn: Connection; table: string | undefined }): Promise { + const rows: Record[] = await conn.query( + 'SELECT t.engine AS engine, e.transactions AS transactions FROM information_schema.tables t JOIN information_schema.engines e ON e.engine = t.engine WHERE t.table_schema = DATABASE() AND t.table_name = ?', + [table] + ); + const [row] = rows; + if (row !== undefined && String(row['transactions']).toUpperCase() === 'NO') { + throw new Error(`Table "${table}" uses the ${String(row['engine'])} storage engine, which does not support transactions, so a failed batch could leave some rows inserted. Use an InnoDB table.`); + } +} + +async function insertInTransaction({ conn, statements }: { conn: Connection; statements: InsertStatement[] }) { + try { + await conn.beginTransaction(); + const results: MysqlWriteResult[] = []; + for (const statement of statements) { + results.push(await conn.query(statement.text, statement.values)); + } + await conn.commit(); + return { + affectedRows: results.reduce((total, result) => total + result.affectedRows, 0), + firstInsertId: results[0]?.insertId ?? 0, + }; + } catch (error) { + await conn.rollback().catch(() => undefined); + const reason = error instanceof Error ? error.message : String(error); + throw new Error(`Insert failed and the whole batch was rolled back: ${reason}`); + } +} + +function isRequiredColumn({ column }: { column: MysqlColumn }): boolean { + const isFilledByDatabase = + column.column_default !== null || + column.extra.includes('auto_increment') || + /(VIRTUAL|STORED) GENERATED/.test(column.extra); + return column.is_nullable === 'NO' && !isFilledByDatabase; +} + +function describeColumn({ column }: { column: MysqlColumn }): string { + return isRequiredColumn({ column }) ? `${column.data_type} · required` : column.data_type; +} + +function toRecords({ rows, keepEmptyStrings }: { rows: unknown; keepEmptyStrings: boolean }): Record[] { + if (!Array.isArray(rows) || rows.length === 0) { + throw new Error('Rows must be a non-empty array of objects.'); + } + return rows.map((row, index) => { + if (typeof row !== 'object' || row === null || Array.isArray(row)) { + throw new Error(`Row ${index + 1} must be an object of column names mapped to values.`); + } + return Object.fromEntries(Object.entries(row).filter(([, value]) => value !== undefined && (keepEmptyStrings || value !== ''))); + }); +} + +function collectColumns({ records }: { records: Record[] }): string[] { + const columns = [...new Set(records.flatMap((record) => Object.keys(record)))]; + if (columns.length === 0) { + throw new Error('Rows must contain at least one column.'); + } + return columns; +} + +function toBindValue(value: unknown): unknown { + const isStructured = typeof value === 'object' && value !== null && !(value instanceof Date) && !Buffer.isBuffer(value); + return isStructured ? JSON.stringify(value) : value; +} + +function buildStatements({ quotedTable, columns, records }: { + quotedTable: string; + columns: string[]; + records: Record[]; +}): InsertStatement[] { + const rowsPerStatement = Math.max(1, Math.min(MAX_ROWS_PER_STATEMENT, Math.floor(MAX_PARAMETERS / columns.length))); + const quotedColumns = columns.map((column) => sanitizeColumnName(column)).join(', '); + + return splitIntoBatches({ records, maxRows: rowsPerStatement }).map((batch) => { + const values: unknown[] = []; + const tuples = batch.map((record) => { + const placeholders = columns.map((column) => { + const value = record[column]; + if (value === undefined) { + return 'DEFAULT'; + } + values.push(toBindValue(value)); + return '?'; + }); + return `(${placeholders.join(', ')})`; + }); + return { + text: `INSERT INTO ${quotedTable} (${quotedColumns}) VALUES ${tuples.join(', ')}`, + values, + }; + }); +} + +function splitIntoBatches({ + records, + maxRows, +}: { + records: Record[]; + maxRows: number; +}): Record[][] { + const batches: Record[][] = []; + let current: Record[] = []; + let currentBytes = 0; + for (const record of records) { + const bytes = estimateBytes({ record }); + const isFull = + current.length >= maxRows || currentBytes + bytes > MAX_STATEMENT_BYTES; + if (current.length > 0 && isFull) { + batches.push(current); + current = []; + currentBytes = 0; + } + current.push(record); + currentBytes += bytes; + } + return current.length > 0 ? [...batches, current] : batches; +} + +function estimateBytes({ record }: { record: Record }): number { + return Object.values(record).reduce( + (total, value) => total + sizeOf({ value }), + 0 + ); +} + +function sizeOf({ value }: { value: unknown }): number { + if (typeof value === 'string') { + return Buffer.byteLength(value); + } + if (Buffer.isBuffer(value)) { + return value.length; + } + if (typeof value === 'object' && value !== null && !(value instanceof Date)) { + return Buffer.byteLength(JSON.stringify(value)); + } + return 16; +} + +const MAX_PARAMETERS = 65535; +const MAX_ROWS_PER_STATEMENT = 1000; +const MAX_STATEMENT_BYTES = 1024 * 1024; + +export const insertRowsUtils = { + toRecords, + collectColumns, + buildStatements, +}; + +export type InsertStatement = { + text: string; + values: unknown[]; +}; + +type MysqlWriteResult = { + affectedRows: number; + insertId: number; +}; diff --git a/packages/pieces/community/mysql/src/lib/common/index.ts b/packages/pieces/community/mysql/src/lib/common/index.ts index 1b05be931007..42e0259d5849 100644 --- a/packages/pieces/community/mysql/src/lib/common/index.ts +++ b/packages/pieces/community/mysql/src/lib/common/index.ts @@ -35,6 +35,20 @@ export async function mysqlGetTableNames(conn: Connection): Promise { return result.map((row: Record) => row[Object.keys(row)[0]]); } +export async function mysqlGetColumns({ conn, table }: { conn: Connection; table: string }): Promise { + const rows: Record[] = await conn.query( + 'SELECT column_name AS column_name, data_type AS data_type, is_nullable AS is_nullable, column_default AS column_default, extra AS extra FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = ? ORDER BY ordinal_position', + [table] + ); + return rows.map((row) => ({ + column_name: String(row['column_name']), + data_type: String(row['data_type']), + is_nullable: String(row['is_nullable']), + column_default: row['column_default'] === null ? null : String(row['column_default']), + extra: String(row['extra']), + })); +} + export const mysqlCommon = { timezone: Property.ShortText({ displayName: 'Timezone', @@ -81,3 +95,11 @@ export function sanitizeColumnName(name: string | undefined): string { } return sqlstring.escapeId(name); } + +export type MysqlColumn = { + column_name: string; + data_type: string; + is_nullable: string; + column_default: string | null; + extra: string; +}; diff --git a/packages/pieces/community/mysql/src/lib/output-schemas.ts b/packages/pieces/community/mysql/src/lib/output-schemas.ts index 232ff3f6003f..f4f6e1423ae5 100644 --- a/packages/pieces/community/mysql/src/lib/output-schemas.ts +++ b/packages/pieces/community/mysql/src/lib/output-schemas.ts @@ -14,6 +14,18 @@ export const insertRowOutputSchema: OutputSchema = { ], }; +export const insertRowsOutputSchema: OutputSchema = { + fields: [ + { key: 'affectedRows', label: 'Inserted Rows', format: 'number' }, + { + key: 'firstInsertId', + label: 'First Insert ID', + format: 'number', + description: 'Auto-increment id of the first inserted row (0 when the table has no auto-increment column). MySQL reports only this one id; the other rows follow it in sequence.', + }, + ], +}; + export const updateRowOutputSchema: OutputSchema = { fields: [ { diff --git a/packages/pieces/community/postgres/package.json b/packages/pieces/community/postgres/package.json index 94a03a3a0a20..48a5a7d229c0 100644 --- a/packages/pieces/community/postgres/package.json +++ b/packages/pieces/community/postgres/package.json @@ -1,6 +1,6 @@ { "name": "@activepieces/piece-postgres", - "version": "0.3.1", + "version": "0.4.0", "main": "./dist/src/index.js", "types": "./dist/src/index.d.ts", "dependencies": { diff --git a/packages/pieces/community/postgres/src/index.ts b/packages/pieces/community/postgres/src/index.ts index 0f7d26adc0f4..79672aede899 100644 --- a/packages/pieces/community/postgres/src/index.ts +++ b/packages/pieces/community/postgres/src/index.ts @@ -7,6 +7,7 @@ import { AppConnectionType, PieceCategory } from '@activepieces/pieces-framework import { runQuery } from './lib/actions/run-query'; import { findRows } from './lib/actions/find-rows'; import { insertRow } from './lib/actions/insert-row'; +import { insertRows } from './lib/actions/insert-rows'; import { updateRow } from './lib/actions/update-row'; import { deleteRow } from './lib/actions/delete-row'; import { getTables } from './lib/actions/get-tables'; @@ -96,6 +97,6 @@ export const postgres = createPiece({ logoUrl: 'https://cdn.activepieces.com/pieces/postgres.png', authors: ["AbdullahBitar", "Willianwg", "dentych", "kishanprmr", "AbdulTheActivePiecer", "khaledmashaly", "abuaboud"], auth: postgresAuth, - actions: [findRows, insertRow, updateRow, deleteRow, getTables, runQuery], + actions: [findRows, insertRow, insertRows, updateRow, deleteRow, getTables, runQuery], triggers: [newRow], }); diff --git a/packages/pieces/community/postgres/src/lib/actions/insert-rows.ts b/packages/pieces/community/postgres/src/lib/actions/insert-rows.ts new file mode 100644 index 000000000000..1cb8f3fb96d1 --- /dev/null +++ b/packages/pieces/community/postgres/src/lib/actions/insert-rows.ts @@ -0,0 +1,244 @@ +import { createAction, InputPropertyMap, Property } from '@activepieces/pieces-framework'; +import { QueryResult } from 'pg'; +import { postgresAuth } from '../..'; +import { pgClient, PostgresColumn, postgresCommon, postgresUtils } from '../common'; +import { insertRowsOutputSchema } from '../output-schemas'; + +export const insertRows = createAction({ + auth: postgresAuth, + name: 'insert-rows', + classification: 'WRITE', + displayName: 'Insert Multiple Rows', + description: 'Inserts many rows into a table in a single transaction', + audience: 'both', + outputSchema: insertRowsOutputSchema, + aiMetadata: { + description: 'Inserts an array of rows into a PostgreSQL table, each row a map of column names to values, using multi-row INSERT statements inside one transaction: if any row fails, none are inserted. Columns a row omits get the column default. Returns the inserted count and, when Return Inserted Rows is on, the stored rows including generated ids. Use instead of repeated single inserts for bulk loads. Not idempotent: each call appends new rows, so repeating it adds duplicates or errors on a unique-key collision.', + idempotent: false, + }, + props: { + table: postgresCommon.table, + batch: Property.DynamicProperties({ + auth: postgresAuth, + displayName: 'Rows', + required: true, + refreshers: ['table'], + props: async ({ auth, table }): Promise => { + if (!auth || !postgresUtils.isPostgresTable(table)) { + return {}; + } + const client = await pgClient(auth); + try { + const columns = await postgresUtils.listColumnDetails({ client, table }); + return { + rows: Property.Array({ + displayName: 'Rows', + description: 'One item per row. Leave a field empty to use the column default.', + required: true, + properties: Object.fromEntries( + columns.map((column) => [ + column.column_name, + Property.ShortText({ + displayName: column.column_name, + description: describeColumn({ column }), + required: isRequiredColumn({ column }), + }), + ]), + ), + }), + }; + } finally { + await client.end(); + } + }, + }), + keep_empty_strings: Property.Checkbox({ + displayName: 'Keep Empty Strings', + description: 'Insert empty fields as empty strings instead of the column default.', + required: false, + defaultValue: false, + }), + return_rows: Property.Checkbox({ + displayName: 'Return Inserted Rows', + description: 'Return the inserted rows, including generated ids. Needs SELECT permission on the table.', + required: false, + defaultValue: false, + }), + }, + async run(context) { + const { table, batch, keep_empty_strings, return_rows } = context.propsValue; + const submittedRecords = insertRowsUtils.toRecords({ rows: batch?.['rows'], keepEmptyStrings: keep_empty_strings === true }); + insertRowsUtils.collectColumns({ records: submittedRecords }); + + const client = await pgClient(context.auth); + client.on('error', () => undefined); + try { + const tableColumns = await postgresUtils.listColumnDetails({ client, table }); + const records = insertRowsUtils.parseArrayColumns({ records: submittedRecords, tableColumns }); + const statements = insertRowsUtils.buildStatements({ + qualifiedTable: postgresUtils.qualifiedName(table), + columns: insertRowsUtils.collectColumns({ records }), + records, + returnRows: return_rows === true, + }); + await client.query('BEGIN'); + const results: QueryResult[] = []; + for (const statement of statements) { + results.push(await client.query(statement.text, statement.values)); + } + await client.query('COMMIT'); + const rowCount = results.reduce((total, result) => total + (result.rowCount ?? 0), 0); + return return_rows === true + ? { rows: results.flatMap((result) => result.rows), rowCount } + : { rowCount }; + } catch (error) { + await client.query('ROLLBACK').catch(() => undefined); + const reason = error instanceof Error ? error.message : String(error); + throw new Error(`Insert failed and the whole batch was rolled back: ${reason}`); + } finally { + await client.end(); + } + }, +}); + +function isRequiredColumn({ column }: { column: PostgresColumn }): boolean { + const isFilledByDatabase = column.column_default !== null || column.is_identity === 'YES' || column.is_generated === 'ALWAYS'; + return column.is_nullable === 'NO' && !isFilledByDatabase; +} + +function describeColumn({ column }: { column: PostgresColumn }): string { + return isRequiredColumn({ column }) ? `${column.data_type} · required` : column.data_type; +} + +function toRecords({ rows, keepEmptyStrings }: { rows: unknown; keepEmptyStrings: boolean }): Record[] { + if (!Array.isArray(rows) || rows.length === 0) { + throw new Error('Rows must be a non-empty array of objects.'); + } + return rows.map((row, index) => { + if (typeof row !== 'object' || row === null || Array.isArray(row)) { + throw new Error(`Row ${index + 1} must be an object of column names mapped to values.`); + } + return Object.fromEntries(Object.entries(row).filter(([, value]) => value !== undefined && (keepEmptyStrings || value !== ''))); + }); +} + +function parseArrayColumns({ records, tableColumns }: { records: Record[]; tableColumns: PostgresColumn[] }): Record[] { + const arrayColumns = new Set(tableColumns.filter((column) => column.data_type === 'ARRAY').map((column) => column.column_name)); + if (arrayColumns.size === 0) { + return records; + } + return records.map((record) => Object.fromEntries( + Object.entries(record).map(([column, value]) => [column, arrayColumns.has(column) ? parseJsonArray({ value }) : value]), + )); +} + +function parseJsonArray({ value }: { value: unknown }): unknown { + if (typeof value !== 'string' || !value.trimStart().startsWith('[')) { + return value; + } + try { + const parsed: unknown = JSON.parse(value); + return Array.isArray(parsed) ? parsed : value; + } catch { + return value; + } +} + +function collectColumns({ records }: { records: Record[] }): string[] { + const columns = [...new Set(records.flatMap((record) => Object.keys(record)))]; + if (columns.length === 0) { + throw new Error('Rows must contain at least one column.'); + } + return columns; +} + +function buildStatements({ qualifiedTable, columns, records, returnRows }: { + qualifiedTable: string; + columns: string[]; + records: Record[]; + returnRows: boolean; +}): InsertStatement[] { + const rowsPerStatement = Math.max(1, Math.min(MAX_ROWS_PER_STATEMENT, Math.floor(MAX_PARAMETERS / columns.length))); + const quotedColumns = columns.map((column) => postgresUtils.quoteIdentifier(column)).join(', '); + const suffix = returnRows ? ' RETURNING *' : ''; + + return splitIntoBatches({ records, maxRows: rowsPerStatement }).map((batch) => { + const values: unknown[] = []; + const tuples = batch.map((record) => { + const placeholders = columns.map((column) => { + const value = record[column]; + if (value === undefined) { + return 'DEFAULT'; + } + values.push(value); + return `$${values.length}`; + }); + return `(${placeholders.join(', ')})`; + }); + return { + text: `INSERT INTO ${qualifiedTable} (${quotedColumns}) VALUES ${tuples.join(', ')}${suffix}`, + values, + }; + }); +} + +function splitIntoBatches({ + records, + maxRows, +}: { + records: Record[]; + maxRows: number; +}): Record[][] { + const batches: Record[][] = []; + let current: Record[] = []; + let currentBytes = 0; + for (const record of records) { + const bytes = estimateBytes({ record }); + const isFull = + current.length >= maxRows || currentBytes + bytes > MAX_STATEMENT_BYTES; + if (current.length > 0 && isFull) { + batches.push(current); + current = []; + currentBytes = 0; + } + current.push(record); + currentBytes += bytes; + } + return current.length > 0 ? [...batches, current] : batches; +} + +function estimateBytes({ record }: { record: Record }): number { + return Object.values(record).reduce( + (total, value) => total + sizeOf({ value }), + 0 + ); +} + +function sizeOf({ value }: { value: unknown }): number { + if (typeof value === 'string') { + return Buffer.byteLength(value); + } + if (Buffer.isBuffer(value)) { + return value.length; + } + if (typeof value === 'object' && value !== null && !(value instanceof Date)) { + return Buffer.byteLength(JSON.stringify(value)); + } + return 16; +} + +const MAX_PARAMETERS = 65535; +const MAX_ROWS_PER_STATEMENT = 1000; +const MAX_STATEMENT_BYTES = 1024 * 1024; + +export const insertRowsUtils = { + toRecords, + collectColumns, + parseArrayColumns, + buildStatements, +}; + +export type InsertStatement = { + text: string; + values: unknown[]; +}; diff --git a/packages/pieces/community/postgres/src/lib/common.ts b/packages/pieces/community/postgres/src/lib/common.ts index 0d2048887748..fe517ad2246b 100644 --- a/packages/pieces/community/postgres/src/lib/common.ts +++ b/packages/pieces/community/postgres/src/lib/common.ts @@ -51,6 +51,14 @@ const listColumns = async ({ client, table }: { client: Client, table: PostgresT return result.rows.map((row) => row.column_name); } +const listColumnDetails = async ({ client, table }: { client: Client, table: PostgresTable }): Promise => { + const result = await client.query( + `SELECT column_name, data_type, is_nullable, column_default, is_identity, is_generated FROM information_schema.columns WHERE table_schema = $1 AND table_name = $2 ORDER BY ordinal_position`, + [table.table_schema, table.table_name] + ); + return result.rows; +} + const isPostgresTable = (value: unknown): value is PostgresTable => { if (typeof value !== 'object' || value === null) { return false; @@ -180,6 +188,7 @@ export const postgresCommon = { export const postgresUtils = { listTables, listColumns, + listColumnDetails, isPostgresTable, qualifiedName, quoteIdentifier, @@ -189,3 +198,12 @@ export type PostgresTable = { table_schema: string; table_name: string; }; + +export type PostgresColumn = { + column_name: string; + data_type: string; + is_nullable: string; + column_default: string | null; + is_identity: string; + is_generated: string; +}; diff --git a/packages/pieces/community/postgres/src/lib/output-schemas.ts b/packages/pieces/community/postgres/src/lib/output-schemas.ts index 2b0d20a63615..580f46257bef 100644 --- a/packages/pieces/community/postgres/src/lib/output-schemas.ts +++ b/packages/pieces/community/postgres/src/lib/output-schemas.ts @@ -26,6 +26,17 @@ export const insertRowOutputSchema: OutputSchema = { ], }; +export const insertRowsOutputSchema: OutputSchema = { + fields: [ + { + key: 'rows', + label: 'Inserted Rows', + description: 'The inserted rows as stored by the database, including defaults and generated ids; each row\'s fields are the columns of the target table. Only present when Return Inserted Rows is enabled.', + }, + { key: 'rowCount', label: 'Inserted Row Count', format: 'number' }, + ], +}; + export const updateRowOutputSchema: OutputSchema = { fields: [ { From de6c7d12324ef255b30c03e675ca0aeeebc11296 Mon Sep 17 00:00:00 2001 From: Daniel Poon <17039704+danielpoonwj@users.noreply.github.com> Date: Wed, 30 Sep 2026 14:29:29 +0800 Subject: [PATCH 2/3] feat(duckdb): Usability enhancements (#15584) Co-authored-by: Kishan Parmar <135701940+kishanprmr@users.noreply.github.com> --- bun.lock | 4 +- packages/pieces/community/duckdb/package.json | 9 +- .../src/lib/actions/create-and-query-db.ts | 72 ++++-- .../duckdb/test/create-and-query-db.test.ts | 215 ++++++++++++++++++ .../pieces/community/duckdb/vitest.config.ts | 17 ++ 5 files changed, 299 insertions(+), 18 deletions(-) create mode 100644 packages/pieces/community/duckdb/test/create-and-query-db.test.ts create mode 100644 packages/pieces/community/duckdb/vitest.config.ts diff --git a/bun.lock b/bun.lock index e31cb1320404..5e6d3e85e638 100644 --- a/bun.lock +++ b/bun.lock @@ -2786,16 +2786,18 @@ }, "packages/pieces/community/duckdb": { "name": "@activepieces/piece-duckdb", - "version": "0.0.8", + "version": "0.0.9", "dependencies": { "@activepieces/core-piece-types": "workspace:*", "@activepieces/core-utils": "workspace:*", "@activepieces/pieces-common": "workspace:*", "@activepieces/pieces-framework": "workspace:*", "@duckdb/node-api": "1.4.3-r.2", + "csv-parse": "5.6.0", }, "devDependencies": { "tslib": "^2.3.0", + "vitest": "3.2.6", }, }, "packages/pieces/community/dumpling-ai": { diff --git a/packages/pieces/community/duckdb/package.json b/packages/pieces/community/duckdb/package.json index a96832fb42fa..352cf80c7bca 100644 --- a/packages/pieces/community/duckdb/package.json +++ b/packages/pieces/community/duckdb/package.json @@ -1,6 +1,6 @@ { "name": "@activepieces/piece-duckdb", - "version": "0.0.8", + "version": "0.0.9", "type": "commonjs", "main": "./dist/src/index.js", "types": "./dist/src/index.d.ts", @@ -8,15 +8,18 @@ "@activepieces/pieces-common": "workspace:*", "@activepieces/pieces-framework": "workspace:*", "@duckdb/node-api": "1.4.3-r.2", + "csv-parse": "5.6.0", "@activepieces/core-piece-types": "workspace:*", "@activepieces/core-utils": "workspace:*" }, "devDependencies": { - "tslib": "^2.3.0" + "tslib": "^2.3.0", + "vitest": "3.2.6" }, "scripts": { "build": "tsc -p tsconfig.lib.json && cp package.json dist/", "bundle": "node ../../../../dist/packages/cli/src/index.js pieces bundle", - "lint": "eslint 'src/**/*.ts'" + "lint": "eslint 'src/**/*.ts'", + "test": "vitest run" } } diff --git a/packages/pieces/community/duckdb/src/lib/actions/create-and-query-db.ts b/packages/pieces/community/duckdb/src/lib/actions/create-and-query-db.ts index 3be14a2abfd1..66b627c30464 100644 --- a/packages/pieces/community/duckdb/src/lib/actions/create-and-query-db.ts +++ b/packages/pieces/community/duckdb/src/lib/actions/create-and-query-db.ts @@ -1,6 +1,8 @@ -import { createAction, Property } from '@activepieces/pieces-framework'; +import { ApFile, createAction, Property } from '@activepieces/pieces-framework'; import { DuckDBInstance } from '@duckdb/node-api'; +import { quotedIdentifier } from '@duckdb/node-api/lib/sql'; +import { parse as parseCsv } from 'csv-parse/sync'; export const createAndQueryDB = createAction({ name: 'createAndQueryDB', @@ -29,16 +31,24 @@ export const createAndQueryDB = createAction({ data: Property.Json({ displayName: 'JSON Data', description: - 'Table data in the form of a JSON array of objects. Object keys would be taken as column names.', - required: true, + 'Table data as a JSON array of objects. Object keys are taken as column names. Takes priority over the file below if set.', + required: false, defaultValue: [], }), + file: Property.File({ + displayName: 'CSV/JSON File', + description: + 'Upload a CSV or JSON file as the table source, used only if JSON Data above is empty. Format is detected from the file extension.', + required: false, + }), schema: Property.Json({ displayName: 'Schema in JSON format', description: ` Provide the schema as a JSON object. Note that only the specified columns would be loaded. Leave empty to autodetect schema, although it is recommended to specify the schema for consistency and avoid unexpected behaviour. +A column type that doesn't match the actual shape of the row data (e.g. a STRUCT/array type for a column that's a plain string, which is always the case for CSV-sourced tables) silently loads as null instead of erroring - double check this against autodetected schema if a column is unexpectedly empty. + More information on data types and accepted values: - https://duckdb.org/2023/03/03/json - https://duckdb.org/docs/stable/sql/data_types/overview @@ -46,6 +56,13 @@ More information on data types and accepted values: required: false, defaultValue: {}, }), + keepNestedFields: Property.Checkbox({ + displayName: 'Keep Nested Fields', + description: + 'When off, a nested object/struct field is split into individual top-level columns (e.g. "meta": {"a":1} becomes column "a"). Turn on to keep it as a single nested column instead, queryable with dot notation (e.g. "meta.a").', + required: false, + defaultValue: false, + }), }, }), query: Property.ShortText({ @@ -72,10 +89,12 @@ More information on data types and accepted values: const dbTables: any[] = context.propsValue.tables ?? []; for (const dbTable of dbTables) { - const dbData = JSON.stringify(dbTable.data); + const tableName = quotedIdentifier(dbTable.name); + const dbData = JSON.stringify(resolveTableRows(dbTable)); let dbSchema = null; - if (dbTable.schema) { + const hasSchema = dbTable.schema && Object.keys(dbTable.schema).length > 0; + if (hasSchema) { dbSchema = JSON.stringify([dbTable.schema]); } else { const schemaResult = await connection.run( @@ -87,16 +106,21 @@ More information on data types and accepted values: dbSchema = detectedSchema[0][0]; } - await connection.run( - ` - CREATE TABLE ${dbTable.name} AS + const shouldFlatten = !dbTable.keepNestedFields; + const createTableQuery = shouldFlatten + ? ` + CREATE TABLE ${tableName} AS SELECT UNNEST(JSON_TRANSFORM($sourceData, $sourceSchema), recursive := true); - `, - { - sourceData: dbData, - sourceSchema: dbSchema, - } - ); + ` + : ` + CREATE TABLE ${tableName} AS + SELECT row.* FROM (SELECT UNNEST(JSON_TRANSFORM($sourceData, $sourceSchema)) AS row) t; + `; + + await connection.run(createTableQuery, { + sourceData: dbData, + sourceSchema: dbSchema, + }); } const queryArgs = context.propsValue.args ?? []; @@ -118,3 +142,23 @@ More information on data types and accepted values: return results; }, }); + +function resolveTableRows(dbTable: { data?: unknown; file?: ApFile }): unknown { + if (Array.isArray(dbTable.data) ? dbTable.data.length > 0 : dbTable.data) { + return dbTable.data; + } + + if (!dbTable.file) { + return dbTable.data; + } + + const text = dbTable.file.data.toString('utf-8'); + return dbTable.file.extension?.toLowerCase() === 'csv' + ? parseCsv(text, { + columns: true, + skip_empty_lines: true, + bom: true, + relax_column_count: true, + }) + : JSON.parse(text.replace(/^/, '')); +} diff --git a/packages/pieces/community/duckdb/test/create-and-query-db.test.ts b/packages/pieces/community/duckdb/test/create-and-query-db.test.ts new file mode 100644 index 000000000000..1537c703aae7 --- /dev/null +++ b/packages/pieces/community/duckdb/test/create-and-query-db.test.ts @@ -0,0 +1,215 @@ +import { ApFile } from '@activepieces/pieces-framework'; +import { describe, expect, it } from 'vitest'; +import { createAndQueryDB } from '../src/lib/actions/create-and-query-db'; + +function runAction(propsValue: Record) { + return createAndQueryDB.run({ propsValue } as never); +} + +function table(overrides: Record) { + return { name: 't', data: [], schema: {}, ...overrides }; +} + +describe('JSON Data prop', () => { + it('loads rows straight from the JSON Data array', async () => { + const rows = await runAction({ + tables: [ + table({ + data: [{ id: 1, name: 'a' }, { id: 2, name: 'b' }], + schema: { id: 'INTEGER', name: 'VARCHAR' }, + }), + ], + query: 'SELECT * FROM t ORDER BY id', + }); + + expect(rows).toEqual([{ id: 1, name: 'a' }, { id: 2, name: 'b' }]); + }); + + it('an explicit schema restricts the table to only the listed columns', async () => { + const rows = await runAction({ + tables: [ + table({ + data: [{ id: 1, name: 'a', extra: 'dropped' }], + schema: { id: 'INTEGER', name: 'VARCHAR' }, + }), + ], + query: 'SELECT * FROM t', + }); + + expect(rows).toEqual([{ id: 1, name: 'a' }]); + }); +}); + +describe('File prop (CSV/JSON)', () => { + it('loads rows from an uploaded JSON file when JSON Data is empty', async () => { + const file = new ApFile('rows.json', Buffer.from(JSON.stringify([{ id: 1, name: 'a' }])), 'json'); + + const rows = await runAction({ + tables: [table({ data: [], file, schema: { id: 'INTEGER', name: 'VARCHAR' } })], + query: 'SELECT * FROM t', + }); + + expect(rows).toEqual([{ id: 1, name: 'a' }]); + }); + + it('loads rows from an uploaded CSV file, format detected from the extension', async () => { + const file = new ApFile('rows.csv', Buffer.from('id,name\n1,a\n2,b\n'), 'csv'); + + const rows = await runAction({ + tables: [table({ data: [], file, schema: { id: 'INTEGER', name: 'VARCHAR' } })], + query: 'SELECT * FROM t ORDER BY id', + }); + + expect(rows).toEqual([{ id: 1, name: 'a' }, { id: 2, name: 'b' }]); + }); + + it('strips a UTF-8 BOM from a CSV header instead of corrupting the first column name', async () => { + const file = new ApFile('rows.csv', Buffer.from('id,name\n1,a\n'), 'csv'); + + const rows = await runAction({ + tables: [table({ data: [], file, schema: { id: 'INTEGER', name: 'VARCHAR' } })], + query: 'SELECT * FROM t', + }); + + expect(rows).toEqual([{ id: 1, name: 'a' }]); + }); + + it('strips a UTF-8 BOM from a JSON file instead of failing to parse', async () => { + const file = new ApFile( + 'rows.json', + Buffer.from('' + JSON.stringify([{ id: 1, name: 'a' }])), + 'json' + ); + + const rows = await runAction({ + tables: [table({ data: [], file, schema: { id: 'INTEGER', name: 'VARCHAR' } })], + query: 'SELECT * FROM t', + }); + + expect(rows).toEqual([{ id: 1, name: 'a' }]); + }); + + it('a ragged CSV row does not abort the whole file', async () => { + const file = new ApFile('rows.csv', Buffer.from('id,name\n1,a,extra\n2,b\n'), 'csv'); + + const rows = await runAction({ + tables: [table({ data: [], file, schema: { id: 'INTEGER', name: 'VARCHAR' } })], + query: 'SELECT * FROM t ORDER BY id', + }); + + expect(rows).toEqual([{ id: 1, name: 'a' }, { id: 2, name: 'b' }]); + }); + + it('JSON Data takes priority over the file when both are set', async () => { + const file = new ApFile('rows.csv', Buffer.from('id,name\n99,ignored\n'), 'csv'); + + const rows = await runAction({ + tables: [table({ data: [{ id: 1, name: 'from-data' }], file, schema: { id: 'INTEGER', name: 'VARCHAR' } })], + query: 'SELECT * FROM t', + }); + + expect(rows).toEqual([{ id: 1, name: 'from-data' }]); + }); +}); + +describe('Nested schema and Keep Nested Fields', () => { + const NESTED_DATA = [{ id: 1, meta: { a: 'x', b: 2 }, tags: ['a', 'b'] }]; + const NESTED_SCHEMA = { id: 'INTEGER', meta: 'STRUCT(a VARCHAR, b INTEGER)', tags: 'VARCHAR[]' }; + + it('flattens nested struct fields into top-level columns by default', async () => { + const rows = await runAction({ + tables: [table({ data: NESTED_DATA, schema: NESTED_SCHEMA })], + query: 'SELECT * FROM t', + }); + + expect(rows).toEqual([{ id: 1, a: 'x', b: 2, tags: ['a', 'b'] }]); + }); + + it('a fresh array row seeds Keep Nested Fields to false (builder default for a checkbox sub-prop), so it still flattens', async () => { + const rows = await runAction({ + tables: [table({ data: NESTED_DATA, schema: NESTED_SCHEMA, keepNestedFields: false })], + query: 'SELECT * FROM t', + }); + + expect(rows).toEqual([{ id: 1, a: 'x', b: 2, tags: ['a', 'b'] }]); + }); + + it('keepNestedFields: true keeps the nested object as a single struct column', async () => { + const rows = await runAction({ + tables: [table({ data: NESTED_DATA, schema: NESTED_SCHEMA, keepNestedFields: true })], + query: 'SELECT * FROM t', + }); + + expect(rows).toEqual([{ id: 1, meta: { a: 'x', b: 2 }, tags: ['a', 'b'] }]); + }); + + it('a schema/data shape mismatch silently nulls the column rather than erroring (known DuckDB behavior)', async () => { + const rows = await runAction({ + tables: [ + table({ + data: [{ id: 1, meta: 'not-a-struct' }], + schema: { id: 'INTEGER', meta: 'STRUCT(a VARCHAR, b INTEGER)' }, + }), + ], + query: 'SELECT * FROM t', + }); + + expect(rows).toEqual([{ id: 1, a: null, b: null }]); + }); +}); + +describe('Table Name is a SQL identifier, not a value (cannot be parameterized)', () => { + it('quotes the table name so multi-statement SQL injection cannot execute', async () => { + const maliciousName = + "z(v INT); CREATE TABLE injected_evil AS SELECT 'pwned' AS marker; CREATE TABLE legit_final"; + + const rows = await runAction({ + tables: [table({ name: maliciousName, data: [{ id: 1 }] })], + query: 'SELECT table_name FROM information_schema.tables', + }); + + expect(rows).toEqual([{ table_name: maliciousName }]); + }); + + it('accepts a plain alphanumeric/underscore table name', async () => { + const rows = await runAction({ + tables: [table({ name: 'valid_table_1', data: [{ id: 1 }] })], + query: 'SELECT * FROM valid_table_1', + }); + + expect(rows).toEqual([{ id: '1' }]); + }); +}); + +describe('Multiple tables', () => { + it('loads more than one table and can join across them in a single query', async () => { + const rows = await runAction({ + tables: [ + table({ + name: 'users', + data: [{ id: 1, name: 'a' }], + schema: { id: 'INTEGER', name: 'VARCHAR' }, + }), + table({ + name: 'orders', + data: [{ id: 10, user_id: 1, total: 5 }], + schema: { id: 'INTEGER', user_id: 'INTEGER', total: 'INTEGER' }, + }), + ], + query: 'SELECT u.name, o.total FROM users u JOIN orders o ON o.user_id = u.id', + }); + + expect(rows).toEqual([{ name: 'a', total: 5 }]); + }); +}); + +describe('Schema autodetection (schema left at its default {})', () => { + it('does not error on a fresh table row whose schema is still the default empty object, and autodetects types (note: integer columns come back as strings, since json_structure infers UBIGINT and getRowObjectsJson stringifies 64-bit ints to avoid precision loss)', async () => { + const rows = await runAction({ + tables: [table({ data: [{ id: 1, name: 'a' }] })], + query: 'SELECT * FROM t', + }); + + expect(rows).toEqual([{ id: '1', name: 'a' }]); + }); +}); diff --git a/packages/pieces/community/duckdb/vitest.config.ts b/packages/pieces/community/duckdb/vitest.config.ts new file mode 100644 index 000000000000..ba8ade4a1780 --- /dev/null +++ b/packages/pieces/community/duckdb/vitest.config.ts @@ -0,0 +1,17 @@ +import path from 'path' +import { defineConfig } from 'vitest/config' + +const repoRoot = path.resolve(__dirname, '../../../..') + +export default defineConfig({ + test: { + globals: true, + environment: 'node', + }, + resolve: { + alias: { + '@activepieces/pieces-framework': path.resolve(repoRoot, 'packages/pieces/framework/src/index.ts'), + '@activepieces/pieces-common': path.resolve(repoRoot, 'packages/pieces/common/src/index.ts'), + }, + }, +}) From 6881d9f2f86e7d75a25b7458737e45c713f5b65d Mon Sep 17 00:00:00 2001 From: Kishan Parmar <135701940+kishanprmr@users.noreply.github.com> Date: Wed, 30 Sep 2026 14:02:38 +0530 Subject: [PATCH 3/3] ci: skip redis-memory-server compile in piece validate and release pieces workflows (#15844) --- .github/workflows/release-pieces.yml | 2 ++ .github/workflows/validate-publishable-packages.yml | 2 ++ packages/pieces/community/figma/package.json | 2 +- 3 files changed, 5 insertions(+), 1 deletion(-) diff --git a/.github/workflows/release-pieces.yml b/.github/workflows/release-pieces.yml index f04fc6a4da31..aa8ebb804d4d 100644 --- a/.github/workflows/release-pieces.yml +++ b/.github/workflows/release-pieces.yml @@ -42,6 +42,8 @@ jobs: - name: Install dependencies run: bun install + env: + REDISMS_DISABLE_POSTINSTALL: "1" - name: copy project .npmrc to user level run: cp .npmrc $HOME/.npmrc diff --git a/.github/workflows/validate-publishable-packages.yml b/.github/workflows/validate-publishable-packages.yml index 8453de1358e3..46d3b35493d1 100644 --- a/.github/workflows/validate-publishable-packages.yml +++ b/.github/workflows/validate-publishable-packages.yml @@ -36,6 +36,8 @@ jobs: - name: Install dependencies run: bun install + env: + REDISMS_DISABLE_POSTINSTALL: "1" - name: validate publishable packages run: npx ts-node -r tsconfig-paths/register -P packages/server/engine/tsconfig.lib.json tools/scripts/validate-publishable-packages.ts diff --git a/packages/pieces/community/figma/package.json b/packages/pieces/community/figma/package.json index 172d14008230..ffd651a19834 100644 --- a/packages/pieces/community/figma/package.json +++ b/packages/pieces/community/figma/package.json @@ -1,6 +1,6 @@ { "name": "@activepieces/piece-figma", - "version": "0.6.2", + "version": "0.6.3", "main": "./dist/src/index.js", "types": "./dist/src/index.d.ts", "scripts": {