diff --git a/front/temporal/relocation/activities/destination_region/front/sql.ts b/front/temporal/relocation/activities/destination_region/front/sql.ts index 8a046bf1b61a..87723d91af1b 100644 --- a/front/temporal/relocation/activities/destination_region/front/sql.ts +++ b/front/temporal/relocation/activities/destination_region/front/sql.ts @@ -8,6 +8,7 @@ import type { CoreEntitiesRelocationBlob, RelocationBlob, } from "@app/temporal/relocation/activities/types"; +import { isArrayOfPlainObjects } from "@app/temporal/relocation/activities/types"; import { deleteFromRelocationStorage, readFromRelocationStorage, @@ -116,7 +117,10 @@ export async function processFrontTableChunk({ for (const { sql, params } of statements) { await frontSequelize.transaction(async (transaction) => frontSequelize.query(sql, { - bind: params, + // TODO(2025-01-31): Remove this once current batch of data is processed. + bind: params.map((p) => + isArrayOfPlainObjects(p) ? JSON.stringify(p) : p + ), type: QueryTypes.INSERT, transaction, }) diff --git a/front/temporal/relocation/activities/types.ts b/front/temporal/relocation/activities/types.ts index c4783f052822..7497287bb05c 100644 --- a/front/temporal/relocation/activities/types.ts +++ b/front/temporal/relocation/activities/types.ts @@ -4,6 +4,7 @@ import type { CoreAPITableBlob, ModelId, } from "@dust-tt/types"; +import { isPlainObject } from "lodash"; import type { RegionType } from "@app/lib/api/regions/config"; @@ -61,3 +62,9 @@ export type CoreTableAPIRelocationBlob = APIRelocationBlob< "tables", CoreAPITableBlob >; + +export function isArrayOfPlainObjects(value: unknown) { + return ( + Array.isArray(value) && value.every((element) => isPlainObject(element)) + ); +} diff --git a/front/temporal/relocation/lib/sql/insert.ts b/front/temporal/relocation/lib/sql/insert.ts index cb083b53ac50..9d823005ec27 100644 --- a/front/temporal/relocation/lib/sql/insert.ts +++ b/front/temporal/relocation/lib/sql/insert.ts @@ -1,3 +1,5 @@ +import { isArrayOfPlainObjects } from "@app/temporal/relocation/activities/types"; + const DEFAULT_CHUNK_SIZE = 250; export function generateParameterizedInsertStatements( @@ -29,7 +31,13 @@ export function generateParameterizedInsertStatements( const rowPlaceholders: string[] = []; for (const col of columns) { rowPlaceholders.push(`$${paramIndex++}`); - params.push(row[col]); + + // Array of plain objects are serialized to JSON strings. + const serialized = isArrayOfPlainObjects(row[col]) + ? JSON.stringify(row[col]) + : row[col]; + + params.push(serialized); } placeholders.push(`(${rowPlaceholders.join(",")})`); }