From cce1da6dba4f4301b900bad2cf75cf61022003c8 Mon Sep 17 00:00:00 2001 From: Artur Piotr Izaak Laskowski Date: Tue, 7 Jul 2026 10:33:52 +0200 Subject: [PATCH 1/2] Move DefaultIcebergConfig message to avoid circular dependencies (#2104) * Define DefaultIcebergConfig message Move DefaultIcebergConfig message from configs.proto to core.proto. This now prevents using core.proto messages in config.proto. The way protos are used across the project lets this change work without additional changes in code. * Remove DefaultIcebergConfig message Removed DefaultIcebergConfig message definition from configs.proto. * Remove import of configs.proto from core.proto * Remove unused import from configs.proto Removed import statement for core.proto. --- protos/configs.proto | 18 ------------------ protos/core.proto | 20 +++++++++++++++++++- 2 files changed, 19 insertions(+), 19 deletions(-) diff --git a/protos/configs.proto b/protos/configs.proto index a0aca3f39..cf81bdcf3 100644 --- a/protos/configs.proto +++ b/protos/configs.proto @@ -69,24 +69,6 @@ message WorkflowSettings { bool include_tests_in_compiled_graph = 16; } -message DefaultIcebergConfig { - // Optional. Bucket name used to construct a storage URI when creating an - // Iceberg table. - string bucket_name = 1; - - // Optional. Table folder root used to construct a storage URI when creating - // an Iceberg table. - string table_folder_root = 2; - - // Optional. Table folder subpath used to construct a storage URI when - // creating an Iceberg table. - string table_folder_subpath = 3; - - // Optional. The connection specifying the credentials to be used to read and - // write to external storage, such as Cloud Storage. - string connection = 4; -} - // Action configs defines the contents of `actions.yaml` configuration files. message ActionConfigs { repeated ActionConfig actions = 1; diff --git a/protos/core.proto b/protos/core.proto index 3e3b451a2..e44c2e013 100644 --- a/protos/core.proto +++ b/protos/core.proto @@ -2,8 +2,8 @@ syntax = "proto3"; package dataform; -import "configs.proto"; import "extension.proto"; + import "google/protobuf/struct.proto"; option java_package = "com.dataform.protos"; @@ -41,6 +41,24 @@ message ProjectConfig { reserved 3, 4, 6, 8, 10, 12, 13; } +message DefaultIcebergConfig { + // Optional. Bucket name used to construct a storage URI when creating an + // Iceberg table. + string bucket_name = 1; + + // Optional. Table folder root used to construct a storage URI when creating + // an Iceberg table. + string table_folder_root = 2; + + // Optional. Table folder subpath used to construct a storage URI when + // creating an Iceberg table. + string table_folder_subpath = 3; + + // Optional. The connection specifying the credentials to be used to read and + // write to external storage, such as Cloud Storage. + string connection = 4; +} + message CompileConfig { // Required. string project_dir = 1; From dfbad6dfa04d0b1b380ab5ad4951507f18ce39f1 Mon Sep 17 00:00:00 2001 From: Edvin <56413069+SuchodolskiEdvin@users.noreply.github.com> Date: Mon, 13 Jul 2026 10:45:26 +0200 Subject: [PATCH 2/2] Support incremental_strategy parameter and new insert_overwrite strategy (#2195) * Support incremental_strategy parameter and new insert_overwrite strategy - updated proto with new parameters - added new tests - added validation for chosen incremental_strategies - added new insert_overwrite strategy logic * Implement incrementalPredicates config for BigQuery incremental tables - Replaced legacy 'T' and 'S' MERGE aliases with explicit 'DATAFORM_DEST' and 'DATAFORM_SOURCE' in the BigQuery adapter. - Updated `incremental_table_test.ts` assertions. - Documented the new `incrementalPredicates` property in the OSS API reference. --- cli/api/BUILD | 1 - cli/api/dbadapters/execution_sql.ts | 122 +++++++-- cli/api/execution_sql_test.ts | 30 +++ cli/api/goldens/insert_overwrite_extend.sql | 99 ++++++++ cli/api/goldens/insert_overwrite_ignore.sql | 25 ++ .../insert_overwrite_simple_column.sql | 25 ++ cli/api/goldens/on_schema_change_ignore.sql | 10 +- .../goldens/on_schema_change_synchronize.sql | 10 +- core/actions/incremental_table.ts | 80 ++++++ core/actions/incremental_table_test.ts | 232 ++++++++++++++++++ core/actions/index.ts | 5 + core/actions/table.ts | 1 + core/actions/table_test.ts | 24 ++ core/actions/view_test.ts | 24 ++ core/main_test.ts | 3 + docs/reference/configs.md | 1 + protos/configs.proto | 18 ++ protos/core.proto | 16 ++ 18 files changed, 690 insertions(+), 36 deletions(-) create mode 100644 cli/api/goldens/insert_overwrite_extend.sql create mode 100644 cli/api/goldens/insert_overwrite_ignore.sql create mode 100644 cli/api/goldens/insert_overwrite_simple_column.sql diff --git a/cli/api/BUILD b/cli/api/BUILD index 111ff2a89..7e65d1efa 100644 --- a/cli/api/BUILD +++ b/cli/api/BUILD @@ -66,7 +66,6 @@ ts_test_suite( data = [ ":node_modules", "//packages/@dataform/core:package_tar", - "//test_credentials:bigquery.json", "@nodejs//:node", "@nodejs//:npm", ] + glob(["goldens/**"]), diff --git a/cli/api/dbadapters/execution_sql.ts b/cli/api/dbadapters/execution_sql.ts index 3745e7a2f..7e048a840 100644 --- a/cli/api/dbadapters/execution_sql.ts +++ b/cli/api/dbadapters/execution_sql.ts @@ -154,24 +154,8 @@ from (${query}) as insertions`; this.buildIncrementalSchemaChangeTasks(tasks, table); // Fall through to run the static DML after the procedure alters the schema case dataform.OnSchemaChange.IGNORE: - default: - tasks.add( - Task.statement( - table.uniqueKey && table.uniqueKey.length > 0 - ? this.mergeInto( - table.target, - tableMetadata?.fields.map(f => f.name), - this.getIncrementalQuery(table), - table.uniqueKey, - table.bigquery && table.bigquery.updatePartitionFilter - ) - : this.insertInto( - table.target, - tableMetadata?.fields.map(f => f.name).map(column => `\`${column}\``), - this.getIncrementalQuery(table) - ) - ) - ); + const columns = tableMetadata?.fields.map(f => f.name) || []; + tasks.add(Task.statement(this.getIncrementalDmlStatement(table, columns))); break; } } @@ -222,6 +206,16 @@ from (${query}) as insertions`; return `drop ${this.tableTypeAsSql(type)} if exists ${this.resolveTarget(target)}`; } + private buildIncrementalPredicatesString( + incrementalPredicates?: string[] | null + ): string { + const validPredicates = incrementalPredicates ? incrementalPredicates.filter(p => p.trim() !== "") : []; + if (validPredicates.length === 0) { + return ""; + } + return `and ${validPredicates.map(p => `(${p})`).join(" and ")}`; + } + private buildIncrementalSchemaChangeTasks(tasks: Tasks, table: dataform.ITable) { const uniqueId = this.uniqueIdGenerator(); @@ -451,25 +445,103 @@ DROP TABLE IF EXISTS ${emptyTempTableName}; create or replace view ${this.resolveTarget(target)} as ${query}`; } - private mergeInto( + private mergeInto( target: dataform.ITarget, columns: string[], query: string, uniqueKey: string[], - updatePartitionFilter: string + bigquery: dataform.IBigQueryOptions ) { + const updatePartitionFilter = bigquery && bigquery.updatePartitionFilter; + const incrementalPredicates = bigquery && bigquery.incrementalPredicates; + const incrementalPredicatesString = this.buildIncrementalPredicatesString(incrementalPredicates); const backtickedColumns = columns.map(column => `\`${column}\``); return ` -merge ${this.resolveTarget(target)} T +merge ${this.resolveTarget(target)} DATAFORM_DEST using (${query} -) S -on ${uniqueKey.map(uniqueKeyCol => `T.${uniqueKeyCol} = S.${uniqueKeyCol}`).join(` and `)} - ${updatePartitionFilter ? `and T.${updatePartitionFilter}` : ""} +) DATAFORM_SOURCE +on ${uniqueKey.map(uniqueKeyCol => `DATAFORM_DEST.${uniqueKeyCol} = DATAFORM_SOURCE.${uniqueKeyCol}`).join(` and `)} ${updatePartitionFilter ? `and DATAFORM_DEST.${updatePartitionFilter}` : ""} +${incrementalPredicatesString ? ` ${incrementalPredicatesString}` : ""} when matched then - update set ${columns.map(column => `\`${column}\` = S.${column}`).join(",")} + update set ${columns.map(column => `\`${column}\` = DATAFORM_SOURCE.${column}`).join(",")} when not matched then insert (${backtickedColumns.join(",")}) values (${backtickedColumns.join(",")})`; } + + private insertOverwrite( + target: dataform.ITarget, + columns: string[], + query: string, + bigquery: dataform.IBigQueryOptions + ): string { + const partitionBy = bigquery && bigquery.partitionBy; + const updatePartitionFilter = bigquery && bigquery.updatePartitionFilter; + const incrementalPredicates = bigquery && bigquery.incrementalPredicates; + const incrementalPredicatesString = this.buildIncrementalPredicatesString(incrementalPredicates); + const uniqueId = this.uniqueIdGenerator(); + const stagingTableUnqualified = `staging_table_temp_${uniqueId}`; + const backtickedColumns = columns.map(column => `\`${column}\``); + const resolveTargetTable = this.resolveTarget(target); + + return `CREATE OR REPLACE TEMP TABLE \`${stagingTableUnqualified}\` AS ( + ${query} +); + +BEGIN + DECLARE partitions_for_replacement DEFAULT ( + ARRAY( + SELECT DISTINCT ${partitionBy} + FROM \`${stagingTableUnqualified}\` + WHERE ${partitionBy} IS NOT NULL + ) + ); + + MERGE ${resolveTargetTable} DATAFORM_DEST + USING \`${stagingTableUnqualified}\` DATAFORM_SOURCE + ON FALSE + WHEN NOT MATCHED BY SOURCE AND ${partitionBy} IN UNNEST(partitions_for_replacement) ${updatePartitionFilter ? `and DATAFORM_DEST.${updatePartitionFilter}` : ""} + ${incrementalPredicatesString ? ` ${incrementalPredicatesString}` : ""} + THEN + DELETE + WHEN NOT MATCHED BY TARGET THEN + INSERT (${backtickedColumns.join(",")}) VALUES (${backtickedColumns.join(",")}); +END; + +DROP TABLE IF EXISTS \`${stagingTableUnqualified}\`;`; + } + + private getIncrementalDmlStatement( + table: dataform.ITable, + columns: string[] + ): string { + const incrementalQuery = this.getIncrementalQuery(table); + + switch (table.incrementalStrategy) { + case dataform.IncrementalStrategy.INSERT_OVERWRITE: + return this.insertOverwrite( + table.target, + columns, + incrementalQuery, + table.bigquery + ); + case dataform.IncrementalStrategy.MERGE: + default: + if (table.uniqueKey && table.uniqueKey.length > 0) { + return this.mergeInto( + table.target, + columns, + incrementalQuery, + table.uniqueKey, + table.bigquery + ); + } + return this.insertInto( + table.target, + columns.map(column => `\`${column}\``), + incrementalQuery + ); + } + } } export function collectEvaluationQueries( diff --git a/cli/api/execution_sql_test.ts b/cli/api/execution_sql_test.ts index 4a8cf5568..fbb87fe1c 100644 --- a/cli/api/execution_sql_test.ts +++ b/cli/api/execution_sql_test.ts @@ -86,4 +86,34 @@ suite("ExecutionSql with 'onSchemaChange'", () => { const expectedSql = fs.readFileSync("cli/api/goldens/on_schema_change_ignore.sql", "utf8"); expect(procedureSql).to.equal(expectedSql.trim()); }); + + test("generates INSERT_OVERWRITE script for IGNORE strategy", () => { + const table = { + ...baseTable, + incrementalStrategy: dataform.IncrementalStrategy.INSERT_OVERWRITE, + bigquery: { + partitionBy: "DATE(ts)", + incrementalPredicates: ["DATAFORM_DEST.ts >= '2024-01-01'", "DATAFORM_SOURCE.ts >= '2024-01-01'"] + } + }; + const tasks = executionSql.publishTasks(table, { fullRefresh: false }, tableMetadata); + const sql = tasks.build().map(t => t.statement).join("\n;\n"); + const expectedSql = fs.readFileSync("cli/api/goldens/insert_overwrite_ignore.sql", "utf8"); + expect(sql).to.equal(expectedSql.trim()); + }); + + test("generates INSERT_OVERWRITE script for EXTEND strategy", () => { + const table = { + ...baseTable, + incrementalStrategy: dataform.IncrementalStrategy.INSERT_OVERWRITE, + onSchemaChange: dataform.OnSchemaChange.EXTEND, + bigquery: { + partitionBy: "DATE(ts)" + } + }; + const tasks = executionSql.publishTasks(table, { fullRefresh: false }, tableMetadata); + const sql = tasks.build().map(t => t.statement).join("\n;\n"); + const expectedSql = fs.readFileSync("cli/api/goldens/insert_overwrite_extend.sql", "utf8"); + expect(sql).to.equal(expectedSql.trim()); + }); }); diff --git a/cli/api/goldens/insert_overwrite_extend.sql b/cli/api/goldens/insert_overwrite_extend.sql new file mode 100644 index 000000000..caf971162 --- /dev/null +++ b/cli/api/goldens/insert_overwrite_extend.sql @@ -0,0 +1,99 @@ +CREATE OR REPLACE PROCEDURE `project-id.dataset-id.df_osc_test_uuid`() +OPTIONS(strict_mode=false) +BEGIN + +-- Create empty table to extract schema of new query. +CREATE OR REPLACE TABLE `project-id.dataset-id.incremental_on_schema_change_df_temp_test_uuid_empty` AS ( + SELECT * FROM (select 1 as id, 'a' as field1, 'new' as field2) AS insertions LIMIT 0 +); + + +-- Compare schemas +DECLARE dataform_columns ARRAY; +DECLARE temp_table_columns ARRAY>; +DECLARE columns_added ARRAY>; +DECLARE columns_removed ARRAY; + +SET dataform_columns = ( + SELECT IFNULL(ARRAY_AGG(DISTINCT column_name), []) + FROM `project-id.dataset-id.INFORMATION_SCHEMA.COLUMNS` + WHERE table_name = 'incremental_on_schema_change' +); + +SET temp_table_columns = ( + SELECT IFNULL(ARRAY_AGG(STRUCT(column_name, data_type)), []) + FROM `project-id.dataset-id.INFORMATION_SCHEMA.COLUMNS` + WHERE table_name = 'incremental_on_schema_change_df_temp_test_uuid_empty' +); + +SET columns_added = ( + SELECT IFNULL(ARRAY_AGG(column_info), []) + FROM UNNEST(temp_table_columns) AS column_info + WHERE column_info.column_name NOT IN UNNEST(dataform_columns) +); +SET columns_removed = ( + SELECT IFNULL(ARRAY_AGG(column_name), []) + FROM UNNEST(dataform_columns) AS column_name + WHERE column_name NOT IN (SELECT col.column_name FROM UNNEST(temp_table_columns) AS col) +); + + +-- Apply schema change strategy (EXTEND). +IF ARRAY_LENGTH(columns_removed) > 0 THEN + RAISE USING MESSAGE = FORMAT( + "Column removals are not allowed when on_schema_change = 'EXTEND'. Removed columns: %T", + columns_removed + ); +END IF; + +IF ARRAY_LENGTH(columns_added) > 0 THEN + EXECUTE IMMEDIATE ( + "ALTER TABLE `project-id.dataset-id.incremental_on_schema_change` " || + ( + SELECT STRING_AGG(FORMAT("ADD COLUMN IF NOT EXISTS %s %s", column_info.column_name, column_info.data_type), ", ") + FROM UNNEST(columns_added) AS column_info + ) + ); +END IF; + + + +-- Cleanup temporary tables. +DROP TABLE IF EXISTS `project-id.dataset-id.incremental_on_schema_change_df_temp_test_uuid_empty`; + +END +; +BEGIN + CALL `project-id.dataset-id.df_osc_test_uuid`(); +EXCEPTION WHEN ERROR THEN + DROP TABLE IF EXISTS `project-id.dataset-id.incremental_on_schema_change_df_temp_test_uuid_empty`; + DROP PROCEDURE IF EXISTS `project-id.dataset-id.df_osc_test_uuid`; + RAISE; +END; +DROP PROCEDURE IF EXISTS `project-id.dataset-id.df_osc_test_uuid` +; +CREATE OR REPLACE TEMP TABLE `staging_table_temp_test_uuid` AS ( + select 1 as id, 'a' as field1, 'new' as field2 +); + +BEGIN + DECLARE partitions_for_replacement DEFAULT ( + ARRAY( + SELECT DISTINCT DATE(ts) + FROM `staging_table_temp_test_uuid` + WHERE DATE(ts) IS NOT NULL + ) + ); + + MERGE `project-id.dataset-id.incremental_on_schema_change` DATAFORM_DEST + USING `staging_table_temp_test_uuid` DATAFORM_SOURCE + ON FALSE + WHEN NOT MATCHED BY SOURCE AND DATE(ts) IN UNNEST(partitions_for_replacement) + + THEN + DELETE + WHEN NOT MATCHED BY TARGET THEN + INSERT (`id`,`field1`) VALUES (`id`,`field1`); +END; + +DROP TABLE IF EXISTS `staging_table_temp_test_uuid` diff --git a/cli/api/goldens/insert_overwrite_ignore.sql b/cli/api/goldens/insert_overwrite_ignore.sql new file mode 100644 index 000000000..86032c3c3 --- /dev/null +++ b/cli/api/goldens/insert_overwrite_ignore.sql @@ -0,0 +1,25 @@ +CREATE OR REPLACE TEMP TABLE `staging_table_temp_test_uuid` AS ( + select 1 as id, 'a' as field1, 'new' as field2 +); + +BEGIN + DECLARE partitions_for_replacement DEFAULT ( + ARRAY( + SELECT DISTINCT DATE(ts) + FROM `staging_table_temp_test_uuid` + WHERE DATE(ts) IS NOT NULL + ) + ); + + MERGE `project-id.dataset-id.incremental_on_schema_change` DATAFORM_DEST + USING `staging_table_temp_test_uuid` DATAFORM_SOURCE + ON FALSE + WHEN NOT MATCHED BY SOURCE AND DATE(ts) IN UNNEST(partitions_for_replacement) + and (DATAFORM_DEST.ts >= '2024-01-01') and (DATAFORM_SOURCE.ts >= '2024-01-01') + THEN + DELETE + WHEN NOT MATCHED BY TARGET THEN + INSERT (`id`,`field1`) VALUES (`id`,`field1`); +END; + +DROP TABLE IF EXISTS `staging_table_temp_test_uuid` diff --git a/cli/api/goldens/insert_overwrite_simple_column.sql b/cli/api/goldens/insert_overwrite_simple_column.sql new file mode 100644 index 000000000..f11241a8c --- /dev/null +++ b/cli/api/goldens/insert_overwrite_simple_column.sql @@ -0,0 +1,25 @@ +CREATE OR REPLACE TEMP TABLE `staging_table_temp_test_uuid` AS ( + select 1 as id, 'a' as field1, 'new' as field2 +); + +BEGIN + DECLARE partitions_for_replacement DEFAULT ( + ARRAY( + SELECT DISTINCT date_col + FROM `staging_table_temp_test_uuid` + WHERE date_col IS NOT NULL + ) + ); + + MERGE `project-id.dataset-id.incremental_on_schema_change` DATAFORM_DEST + USING `staging_table_temp_test_uuid` DATAFORM_SOURCE + ON FALSE + WHEN NOT MATCHED BY SOURCE AND date_col IN UNNEST(partitions_for_replacement) + + THEN + DELETE + WHEN NOT MATCHED BY TARGET THEN + INSERT (`id`,`field1`) VALUES (`id`,`field1`); +END; + +DROP TABLE IF EXISTS `staging_table_temp_test_uuid`; diff --git a/cli/api/goldens/on_schema_change_ignore.sql b/cli/api/goldens/on_schema_change_ignore.sql index 0f9c016ba..5d0353e10 100644 --- a/cli/api/goldens/on_schema_change_ignore.sql +++ b/cli/api/goldens/on_schema_change_ignore.sql @@ -1,9 +1,9 @@ -merge `project-id.dataset-id.incremental_on_schema_change` T +merge `project-id.dataset-id.incremental_on_schema_change` DATAFORM_DEST using (select 1 as id, 'a' as field1, 'new' as field2 -) S -on T.id = S.id - +) DATAFORM_SOURCE +on DATAFORM_DEST.id = DATAFORM_SOURCE.id + when matched then - update set `id` = S.id,`field1` = S.field1 + update set `id` = DATAFORM_SOURCE.id,`field1` = DATAFORM_SOURCE.field1 when not matched then insert (`id`,`field1`) values (`id`,`field1`) \ No newline at end of file diff --git a/cli/api/goldens/on_schema_change_synchronize.sql b/cli/api/goldens/on_schema_change_synchronize.sql index bcfc63eac..bf87ae8ca 100644 --- a/cli/api/goldens/on_schema_change_synchronize.sql +++ b/cli/api/goldens/on_schema_change_synchronize.sql @@ -87,12 +87,12 @@ EXCEPTION WHEN ERROR THEN END; DROP PROCEDURE IF EXISTS `project-id.dataset-id.df_osc_test_uuid` ; -merge `project-id.dataset-id.incremental_on_schema_change` T +merge `project-id.dataset-id.incremental_on_schema_change` DATAFORM_DEST using (select 1 as id, 'a' as field1, 'new' as field2 -) S -on T.id = S.id - +) DATAFORM_SOURCE +on DATAFORM_DEST.id = DATAFORM_SOURCE.id + when matched then - update set `id` = S.id,`field1` = S.field1 + update set `id` = DATAFORM_SOURCE.id,`field1` = DATAFORM_SOURCE.field1 when not matched then insert (`id`,`field1`) values (`id`,`field1`) \ No newline at end of file diff --git a/core/actions/incremental_table.ts b/core/actions/incremental_table.ts index 9abd4477b..3ae9560f8 100644 --- a/core/actions/incremental_table.ts +++ b/core/actions/incremental_table.ts @@ -202,6 +202,7 @@ export class IncrementalTable extends ActionBuilder { partitionExpirationDays: config.partitionExpirationDays, requirePartitionFilter: config.requirePartitionFilter, additionalOptions: config.additionalOptions, + incrementalPredicates: config.incrementalPredicates, ...(config.iceberg ? { connection: getConnectionForIcebergTable( config.iceberg.connection, @@ -217,6 +218,10 @@ export class IncrementalTable extends ActionBuilder { } : {}), }); this.proto.onSchemaChange = this.mapOnSchemaChange(config.onSchemaChange); + this.proto.incrementalStrategy = this.mapIncrementalStrategy(config.incrementalStrategy); + + this.checkIncrementalStrategyRequirements(config); + this.checkMutuallyExclusivePredicates(config); if (config.reservation) { if (!this.proto.actionDescriptor) { @@ -669,6 +674,7 @@ export class IncrementalTable extends ActionBuilder { "partitionExpirationDays", "requirePartitionFilter", "additionalOptions", + "incrementalPredicates", "iceberg" ]), "BigQuery table config" @@ -736,6 +742,80 @@ export class IncrementalTable extends ActionBuilder { throw new Error(`OnSchemaChange value "${onSchemaChange}" is not supported`); } } + + private mapIncrementalStrategy( + incrementalStrategy?: string | number + ): dataform.IncrementalStrategy { + if (!incrementalStrategy) { + return dataform.IncrementalStrategy.INCREMENTAL_STRATEGY_UNSPECIFIED; + } + + if (typeof incrementalStrategy === "number") { + switch (incrementalStrategy) { + case dataform.ActionConfig.IncrementalStrategy.INCREMENTAL_STRATEGY_UNSPECIFIED: + return dataform.IncrementalStrategy.INCREMENTAL_STRATEGY_UNSPECIFIED; + case dataform.ActionConfig.IncrementalStrategy.INCREMENTAL_STRATEGY_MERGE: + return dataform.IncrementalStrategy.MERGE; + case dataform.ActionConfig.IncrementalStrategy.INCREMENTAL_STRATEGY_INSERT_OVERWRITE: + return dataform.IncrementalStrategy.INSERT_OVERWRITE; + default: + throw new Error(`IncrementalStrategy value "${incrementalStrategy}" is not supported`); + } + } + + switch (incrementalStrategy.toString().toUpperCase()) { + case "INCREMENTAL_STRATEGY_UNSPECIFIED": + return dataform.IncrementalStrategy.INCREMENTAL_STRATEGY_UNSPECIFIED; + case "MERGE": + return dataform.IncrementalStrategy.MERGE; + case "INSERT_OVERWRITE": + return dataform.IncrementalStrategy.INSERT_OVERWRITE; + default: + throw new Error(`IncrementalStrategy value "${incrementalStrategy}" is not supported`); + } + } + + private checkIncrementalStrategyRequirements(config: dataform.ActionConfig.IIncrementalTableConfig) { + switch (this.proto.incrementalStrategy) { + case dataform.IncrementalStrategy.INSERT_OVERWRITE: + if (!this.proto.bigquery || !this.proto.bigquery.partitionBy) { + this.session.compileError( + new Error(`IncrementalStrategy 'insert_overwrite' requires 'partitionBy' to be set`), + config.filename, + this.proto.target + ); + } + break; + case dataform.IncrementalStrategy.MERGE: + if (!this.proto.uniqueKey || this.proto.uniqueKey.length === 0) { + this.session.compileError( + new Error(`IncrementalStrategy 'merge' requires 'uniqueKey' to be set`), + config.filename, + this.proto.target + ); + } + break; + default: + break; + } + } + + private checkMutuallyExclusivePredicates(config: dataform.ActionConfig.IIncrementalTableConfig) { + if ( + this.proto.bigquery && + this.proto.bigquery.updatePartitionFilter && + this.proto.bigquery.incrementalPredicates && + this.proto.bigquery.incrementalPredicates.length > 0 + ) { + this.session.compileError( + new Error( + `incrementalPredicates and updatePartitionFilter cannot be both set. Use only incrementalPredicates.` + ), + config.filename, + this.proto.target + ); + } + } } /** diff --git a/core/actions/incremental_table_test.ts b/core/actions/incremental_table_test.ts index 9b444fbb9..b80ceab43 100644 --- a/core/actions/incremental_table_test.ts +++ b/core/actions/incremental_table_test.ts @@ -62,6 +62,7 @@ actions: onSchemaChange: "IGNORE", query: "SELECT 1", incrementalQuery: "SELECT 1", + incrementalStrategy: "INCREMENTAL_STRATEGY_UNSPECIFIED", type: "incremental", enumType: "INCREMENTAL", protected: true, @@ -185,6 +186,7 @@ SELECT 1` fileName: `definitions/${testParameters.filename}`, query: "\n\n\nSELECT 1", incrementalQuery: "\n\n\nSELECT 1", + incrementalStrategy: "INCREMENTAL_STRATEGY_UNSPECIFIED", actionDescriptor: { ...exampleActionDescriptor.outputActionDescriptor, reservation: "reservation", @@ -251,9 +253,36 @@ SELECT 1`; fileName: `definitions/${minimalIncrementalTableName}.sqlx`, query: "\n\n\nSELECT 1", incrementalQuery: "\n\n\nSELECT 1", + incrementalStrategy: "INCREMENTAL_STRATEGY_UNSPECIFIED", } ]); }) + + test("throws an error when incrementalPredicates is not an array", () => { + const projectDir = tmpDirFixture.createNewTmpDir(); + fs.writeFileSync( + path.join(projectDir, "workflow_settings.yaml"), + VALID_WORKFLOW_SETTINGS_YAML + ); + fs.mkdirSync(path.join(projectDir, "definitions")); + fs.writeFileSync( + path.join(projectDir, "definitions/invalid_predicates.sqlx"), + `config { + type: "incremental", + bigquery: { + incrementalPredicates: "not_an_array" + } +} +SELECT 1` + ); + + const result = runMainInVm(coreExecutionRequestFromPath(projectDir)); + + expect(result.compile.compiledGraph.graphErrors.compilationErrors.length).equals(1); + expect(result.compile.compiledGraph.graphErrors.compilationErrors[0].message).contains("incrementalPredicates"); + }); + + }); test("action config options", () => { @@ -348,6 +377,7 @@ actions: fileName: "definitions/filename.sql", query: "SELECT 1", incrementalQuery: "SELECT 1", + incrementalStrategy: "INCREMENTAL_STRATEGY_UNSPECIFIED", actionDescriptor: { bigqueryLabels: { key: "val" @@ -964,6 +994,7 @@ select \${incremental()} as is_incremental` hermeticity: "NON_HERMETIC", onSchemaChange: "IGNORE", fileName: "definitions/incremental.js", + incrementalStrategy: "INCREMENTAL_STRATEGY_UNSPECIFIED", jitCode: '(ctx) => Promise.resolve({query: "select 1", incrementalQuery: "select 1"})', actionDescriptor: { compilationMode: "ACTION_COMPILATION_MODE_JIT" @@ -987,4 +1018,205 @@ select \${incremental()} as is_incremental` expect(result.compile.compiledGraph.graphErrors.compilationErrors.some(e => e.message.includes("Cannot mix AoT and JiT compilation"))).equals(true); }); }); + + suite("incrementalStrategy", () => { + test("compiles successfully with insert_overwrite and partitionBy", () => { + const projectDir = tmpDirFixture.createNewTmpDir(); + fs.writeFileSync( + path.join(projectDir, "workflow_settings.yaml"), + VALID_WORKFLOW_SETTINGS_YAML + ); + fs.mkdirSync(path.join(projectDir, "definitions")); + fs.writeFileSync( + path.join(projectDir, "definitions/incremental.sqlx"), + `config { + type: "incremental", + incrementalStrategy: "insert_overwrite", + partitionBy: "DATE(ts)" + } + SELECT 1` + ); + + const result = runMainInVm(coreExecutionRequestFromPath(projectDir)); + + expect(result.compile.compiledGraph.graphErrors.compilationErrors).deep.equals([]); + expect(asPlainObject(result.compile.compiledGraph.tables)).deep.equals( + asPlainObject([ + { + target: { + database: "defaultProject", + schema: "defaultDataset", + name: "incremental" + }, + canonicalTarget: { + database: "defaultProject", + schema: "defaultDataset", + name: "incremental" + }, + fileName: "definitions/incremental.sqlx", + hermeticity: "NON_HERMETIC", + onSchemaChange: "IGNORE", + query: "\n SELECT 1", + incrementalQuery: "\n SELECT 1", + incrementalStrategy: "INSERT_OVERWRITE", + type: "incremental", + enumType: "INCREMENTAL", + protected: false, + disabled: false, + bigquery: { + partitionBy: "DATE(ts)" + } + } + ]) + ); + }); + + test("compilation fails with insert_overwrite and missing partitionBy", () => { + const projectDir = tmpDirFixture.createNewTmpDir(); + fs.writeFileSync( + path.join(projectDir, "workflow_settings.yaml"), + VALID_WORKFLOW_SETTINGS_YAML + ); + fs.mkdirSync(path.join(projectDir, "definitions")); + fs.writeFileSync( + path.join(projectDir, "definitions/incremental.sqlx"), + `config { + type: "incremental", + incrementalStrategy: "insert_overwrite" + } + SELECT 1` + ); + + const result = runMainInVm(coreExecutionRequestFromPath(projectDir)); + + expect(result.compile.compiledGraph.graphErrors.compilationErrors.length).greaterThan(0); + expect(result.compile.compiledGraph.graphErrors.compilationErrors[0].message).contains( + "IncrementalStrategy 'insert_overwrite' requires 'partitionBy' to be set" + ); + }); + + test("compiles successfully with merge and uniqueKey", () => { + const projectDir = tmpDirFixture.createNewTmpDir(); + fs.writeFileSync( + path.join(projectDir, "workflow_settings.yaml"), + VALID_WORKFLOW_SETTINGS_YAML + ); + fs.mkdirSync(path.join(projectDir, "definitions")); + fs.writeFileSync( + path.join(projectDir, "definitions/incremental.sqlx"), + `config { + type: "incremental", + incrementalStrategy: "merge", + uniqueKey: ["id"] + } + SELECT 1` + ); + + const result = runMainInVm(coreExecutionRequestFromPath(projectDir)); + + expect(result.compile.compiledGraph.graphErrors.compilationErrors).deep.equals([]); + expect(asPlainObject(result.compile.compiledGraph.tables)).deep.equals( + asPlainObject([ + { + target: { + database: "defaultProject", + schema: "defaultDataset", + name: "incremental" + }, + canonicalTarget: { + database: "defaultProject", + schema: "defaultDataset", + name: "incremental" + }, + fileName: "definitions/incremental.sqlx", + hermeticity: "NON_HERMETIC", + onSchemaChange: "IGNORE", + query: "\n SELECT 1", + incrementalQuery: "\n SELECT 1", + incrementalStrategy: "MERGE", + type: "incremental", + enumType: "INCREMENTAL", + protected: false, + disabled: false, + uniqueKey: ["id"] + } + ]) + ); + }); + + test("compilation fails with merge and missing uniqueKey", () => { + const projectDir = tmpDirFixture.createNewTmpDir(); + fs.writeFileSync( + path.join(projectDir, "workflow_settings.yaml"), + VALID_WORKFLOW_SETTINGS_YAML + ); + fs.mkdirSync(path.join(projectDir, "definitions")); + fs.writeFileSync( + path.join(projectDir, "definitions/incremental.sqlx"), + `config { + type: "incremental", + incrementalStrategy: "merge" + } + SELECT 1` + ); + + const result = runMainInVm(coreExecutionRequestFromPath(projectDir)); + + expect(result.compile.compiledGraph.graphErrors.compilationErrors.length).greaterThan(0); + expect(result.compile.compiledGraph.graphErrors.compilationErrors[0].message).contains( + "IncrementalStrategy 'merge' requires 'uniqueKey' to be set" + ); + }); + + test("compilation fails with invalid incrementalStrategy", () => { + const projectDir = tmpDirFixture.createNewTmpDir(); + fs.writeFileSync( + path.join(projectDir, "workflow_settings.yaml"), + VALID_WORKFLOW_SETTINGS_YAML + ); + fs.mkdirSync(path.join(projectDir, "definitions")); + fs.writeFileSync( + path.join(projectDir, "definitions/incremental.sqlx"), + `config { + type: "incremental", + incrementalStrategy: "invalid_strategy" + } + SELECT 1` + ); + + const result = runMainInVm(coreExecutionRequestFromPath(projectDir)); + + expect(result.compile.compiledGraph.graphErrors.compilationErrors.length).greaterThan(0); + expect(result.compile.compiledGraph.graphErrors.compilationErrors[0].message).contains( + 'IncrementalStrategy value "invalid_strategy" is not supported' + ); + }); + + test("compilation fails when both incrementalPredicates and updatePartitionFilter are provided", () => { + const projectDir = tmpDirFixture.createNewTmpDir(); + fs.writeFileSync( + path.join(projectDir, "workflow_settings.yaml"), + VALID_WORKFLOW_SETTINGS_YAML + ); + fs.mkdirSync(path.join(projectDir, "definitions")); + fs.writeFileSync( + path.join(projectDir, "definitions/incremental.sqlx"), + `config { + type: "incremental", + bigquery: { + updatePartitionFilter: "T.foo = 1", + incrementalPredicates: ["bar = 2"] + } + } + SELECT 1` + ); + + const result = runMainInVm(coreExecutionRequestFromPath(projectDir)); + + expect(result.compile.compiledGraph.graphErrors.compilationErrors.length).equals(1); + expect(result.compile.compiledGraph.graphErrors.compilationErrors[0].message).contains( + "incrementalPredicates and updatePartitionFilter cannot be both set. Use only incrementalPredicates." + ); + }); + }); }); diff --git a/core/actions/index.ts b/core/actions/index.ts index 63f214ca3..28b3eabfb 100644 --- a/core/actions/index.ts +++ b/core/actions/index.ts @@ -371,6 +371,7 @@ export interface ILegacyBigQueryOptions { tableFolderRoot?: string; tableFolderSubpath?: string; } + incrementalPredicates?: string[]; } /** @@ -472,6 +473,10 @@ export class LegacyConfigConverter { legacyConfig.iceberg = legacyConfig.bigquery.iceberg; delete legacyConfig.bigquery.iceberg; } + if (!!legacyConfig.bigquery.incrementalPredicates) { + legacyConfig.incrementalPredicates = legacyConfig.bigquery.incrementalPredicates; + delete legacyConfig.bigquery.incrementalPredicates; + } // To prevent skipping throwing an error when there are additional, unused fields, only delete // the legacy bigquery object if there are no more fields left on it. if (Object.keys(legacyConfig.bigquery).length === 0) { diff --git a/core/actions/table.ts b/core/actions/table.ts index ee1562527..07bec675a 100644 --- a/core/actions/table.ts +++ b/core/actions/table.ts @@ -651,6 +651,7 @@ export class Table extends ActionBuilder { "partitionExpirationDays", "requirePartitionFilter", "additionalOptions", + "incrementalPredicates", "iceberg" ]), "BigQuery table config" diff --git a/core/actions/table_test.ts b/core/actions/table_test.ts index 38a010d1e..aaaa3c098 100644 --- a/core/actions/table_test.ts +++ b/core/actions/table_test.ts @@ -484,6 +484,30 @@ ${exampleBuiltInAssertionsAsYaml.inputActionConfigBlock} ); }); + test("fails compilation if incrementalStrategy is set on standard table", () => { + const projectDir = tmpDirFixture.createNewTmpDir(); + fs.writeFileSync( + path.join(projectDir, "workflow_settings.yaml"), + VALID_WORKFLOW_SETTINGS_YAML + ); + fs.mkdirSync(path.join(projectDir, "definitions")); + fs.writeFileSync( + path.join(projectDir, "definitions/table.sqlx"), + `config { + type: "table", + incrementalStrategy: "merge" + } + SELECT 1` + ); + + const result = runMainInVm(coreExecutionRequestFromPath(projectDir)); + + expect(result.compile.compiledGraph.graphErrors.compilationErrors.length).greaterThan(0); + expect(result.compile.compiledGraph.graphErrors.compilationErrors[0].message).contains( + 'Unexpected property "incrementalStrategy"' + ); + }); + suite("Iceberg table options", () => { const setupFiles = ( projectDir: string, diff --git a/core/actions/view_test.ts b/core/actions/view_test.ts index b72c4444e..cc4c3fbdd 100644 --- a/core/actions/view_test.ts +++ b/core/actions/view_test.ts @@ -282,6 +282,30 @@ ${exampleBuiltInAssertionsAsYaml.inputActionConfigBlock} ); }); + test("fails compilation if incrementalStrategy is set on view", () => { + const projectDir = tmpDirFixture.createNewTmpDir(); + fs.writeFileSync( + path.join(projectDir, "workflow_settings.yaml"), + VALID_WORKFLOW_SETTINGS_YAML + ); + fs.mkdirSync(path.join(projectDir, "definitions")); + fs.writeFileSync( + path.join(projectDir, "definitions/view.sqlx"), + `config { + type: "view", + incrementalStrategy: "merge" + } + SELECT 1` + ); + + const result = runMainInVm(coreExecutionRequestFromPath(projectDir)); + + expect(result.compile.compiledGraph.graphErrors.compilationErrors.length).greaterThan(0); + expect(result.compile.compiledGraph.graphErrors.compilationErrors[0].message).contains( + 'Unexpected property "incrementalStrategy"' + ); + }); + suite("jit compilation", () => { test("jit compilation is supported", () => { const projectDir = tmpDirFixture.createNewTmpDir(); diff --git a/core/main_test.ts b/core/main_test.ts index 18d39c2ad..76d365b4a 100644 --- a/core/main_test.ts +++ b/core/main_test.ts @@ -1328,6 +1328,7 @@ publish("name", { incrementalPostOps: ["post_op"], incrementalPreOps: ["pre_op"], incrementalQuery: "SELECT 1", + incrementalStrategy: "INCREMENTAL_STRATEGY_UNSPECIFIED", protected: false, onSchemaChange: "IGNORE" } @@ -1393,6 +1394,7 @@ publish("name", { ...(tableType === "incremental" ? { incrementalQuery: "SELECT * FROM `defaultProject.defaultDataset.operation`", + incrementalStrategy: "INCREMENTAL_STRATEGY_UNSPECIFIED", protected: false, onSchemaChange: "IGNORE" } @@ -1442,6 +1444,7 @@ publish("name", { incrementalQuery: "SELECT 1", protected: true, onSchemaChange: "IGNORE", + incrementalStrategy: "INCREMENTAL_STRATEGY_UNSPECIFIED", } ]); }); diff --git a/docs/reference/configs.md b/docs/reference/configs.md index de618ae70..40ef8ca53 100644 --- a/docs/reference/configs.md +++ b/docs/reference/configs.md @@ -134,6 +134,7 @@ Action config defines the contents of `actions.yaml` configuration files. | partitionExpirationDays | [int32](#int32) | | The number of days for which BigQuery stores data in each partition. The setting applies to all partitions in a table, but is calculated independently for each partition based on the partition time. | | requirePartitionFilter | [bool](#bool) | | Declares whether the partitioned table requires a WHERE clause predicate filter that filters the partitioning column. | | updatePartitionFilter | [string](#string) | | SQL-based filter for when incremental updates are applied. | +| incrementalPredicates | [string](#string) | repeated | Array of SQL-based predicates for when incremental updates are applied. | | clusterBy | [string](#string) | repeated | The keys by which to cluster partitions by. See https://cloud.google.com/dataform/docs/partitions-clusters. | | labels | [ActionConfig.IncrementalTableConfig.LabelsEntry](#dataform-ActionConfig-IncrementalTableConfig-LabelsEntry) | repeated | Key-value pairs for BigQuery labels. | | additionalOptions | [ActionConfig.IncrementalTableConfig.AdditionalOptionsEntry](#dataform-ActionConfig-IncrementalTableConfig-AdditionalOptionsEntry) | repeated | Key-value pairs of additional options to pass to the BigQuery API. Some options, for example, partitionExpirationDays, have dedicated type/validity checked fields. For such options, use the dedicated fields. | diff --git a/protos/configs.proto b/protos/configs.proto index cf81bdcf3..7dd2cadac 100644 --- a/protos/configs.proto +++ b/protos/configs.proto @@ -364,6 +364,18 @@ message ActionConfig { SYNCHRONIZE = 3; } + enum IncrementalStrategy { + // Strategy is unspecified. Defaults to MERGE if uniqueKey is configured, + // or APPEND otherwise. + INCREMENTAL_STRATEGY_UNSPECIFIED = 0; + // MERGE strategy. Matches rows based on uniqueKey to update existing rows + // or insert new ones. + INCREMENTAL_STRATEGY_MERGE = 1; + // INSERT_OVERWRITE strategy. Overwrites entire partitions in the target + // table that are present in the new staging data. + INCREMENTAL_STRATEGY_INSERT_OVERWRITE = 2; + } + message IncrementalTableConfig { // The name of the incremental table. string name = 1; @@ -467,6 +479,12 @@ message ActionConfig { // If unset, the value from workflow_settings.yaml is used. If neither is set, default BigQuery behavior applies. // Dataform CLI only (GCP Dataform support pending). string reservation = 27; + + // Optional. The incremental strategy to use when updating the table. + IncrementalStrategy incremental_strategy = 28; + + // Array of SQL-based predicates for when incremental updates are applied. + repeated string incremental_predicates = 29; } message AssertionConfig { diff --git a/protos/core.proto b/protos/core.proto index e44c2e013..5b8528395 100644 --- a/protos/core.proto +++ b/protos/core.proto @@ -110,6 +110,7 @@ message BigQueryOptions { TableFormat table_format = 9; FileFormat file_format = 10; string storage_uri = 11; + repeated string incremental_predicates = 12; } message GraphErrors { @@ -194,6 +195,18 @@ enum OnSchemaChange { SYNCHRONIZE = 3; } +enum IncrementalStrategy { + // Strategy is unspecified. Defaults to MERGE if uniqueKey is configured, + // or APPEND otherwise. + INCREMENTAL_STRATEGY_UNSPECIFIED = 0; + // MERGE strategy. Matches rows based on uniqueKey to update existing row + // or insert new ones. + MERGE = 1; + // INSERT_OVERWRITE strategy. Overwrites entire partitions in the target + // table that are present in the new staging data. + INSERT_OVERWRITE = 2; +} + message Table { // For legacy compatability reasons, both incremental tables and views are // configured in compiled graphs via the Table proto. @@ -230,6 +243,9 @@ message Table { string jit_code = 38; + // The incremental strategy to use when updating the incremental table. + IncrementalStrategy incremental_strategy = 39; + // Warehouse specific features. BigQueryOptions bigquery = 22;