diff --git a/cli/api/dbadapters/execution_sql.ts b/cli/api/dbadapters/execution_sql.ts index b513fcac4..d5b736f44 100644 --- a/cli/api/dbadapters/execution_sql.ts +++ b/cli/api/dbadapters/execution_sql.ts @@ -153,7 +153,7 @@ from (${query}) as insertions`; case dataform.OnSchemaChange.EXTEND: case dataform.OnSchemaChange.SYNCHRONIZE: this.buildIncrementalSchemaChangeTasks(tasks, table); - // Fall through to run the static DML after the procedure alters the schema + break; case dataform.OnSchemaChange.IGNORE: const columns = tableMetadata?.fields.map((f) => f.name) || []; tasks.add(Task.statement(this.getIncrementalDmlStatement(table, columns))); @@ -233,12 +233,14 @@ from (${query}) as insertions`; ...table.target, name: `${table.target.name}_df_temp_${uniqueId}_empty`, }; + const tempTableName = `${table.target.name}_df_temp_${uniqueId}_temp`; const procedureName = this.createProcedureName(table.target, uniqueId); const procedureBody = this.incrementalSchemaChangeBody( table, this.resolveTarget(table.target), emptyTempTableTarget, + tempTableName, ); const createProcedureSql = `CREATE OR REPLACE PROCEDURE ${procedureName}() @@ -274,10 +276,23 @@ END; DROP PROCEDURE IF EXISTS ${procedureName};`; } - private declareSchemaChangeVariablesSql(onSchemaChange: dataform.OnSchemaChange): string { + private declareSchemaChangeVariablesSql(table: dataform.ITable): string { + const onSchemaChange = table.onSchemaChange || dataform.OnSchemaChange.IGNORE; + const isMerge = + table.incrementalStrategy !== dataform.IncrementalStrategy.INSERT_OVERWRITE && + table.uniqueKey && + table.uniqueKey.length > 0; + let sql = ` -- Declare variables for schema comparison and strategy execution. DECLARE dataform_columns ARRAY; +DECLARE dataform_columns_list STRING;`; + + if (isMerge) { + sql += `\nDECLARE dataform_columns_merge STRING;`; + } + + sql += ` DECLARE temp_table_columns ARRAY>; DECLARE columns_added ARRAY>; DECLARE columns_removed ARRAY;`; @@ -401,10 +416,128 @@ ${this.alterTableAddColumnsSql(qualifiedTargetTableName)} END IF;`; } - private cleanupSql(emptyTempTableName: string): string { + private escapeSqlString(str: string): string { + return str.replace(/\\/g, "\\\\").replace(/"/g, '\\"'); + } + + private executeDynamicDmlSql( + table: dataform.ITable, + qualifiedTargetTableName: string, + tempTableName: string, + query: string, + ): string { + const isMerge = + table.incrementalStrategy !== dataform.IncrementalStrategy.INSERT_OVERWRITE && + table.uniqueKey && + table.uniqueKey.length > 0; + + let sql = ` +-- Prepare dynamic column lists and staging table for DML. +SET dataform_columns_list = ( + SELECT STRING_AGG(FORMAT("\`%s\`", column_info.column_name), ", ") + FROM UNNEST(temp_table_columns) AS column_info +);`; + + if (isMerge) { + sql += ` +SET dataform_columns_merge = ( + SELECT STRING_AGG(FORMAT("\`%s\` = DATAFORM_SOURCE.\`%s\`", column_info.column_name, column_info.column_name), ", ") + FROM UNNEST(temp_table_columns) AS column_info +);`; + } + + sql += ` + +CREATE OR REPLACE TEMP TABLE \`${tempTableName}\` AS ( + ${query} +); +`; + + switch (table.incrementalStrategy) { + case dataform.IncrementalStrategy.INSERT_OVERWRITE: { + const partitionBy = table.bigquery && table.bigquery.partitionBy; + const updatePartitionFilter = table.bigquery && table.bigquery.updatePartitionFilter; + const incrementalPredicates = table.bigquery && table.bigquery.incrementalPredicates; + const incrementalPredicatesString = + this.buildIncrementalPredicatesString(incrementalPredicates); + const notMatchedBySourceCondition = [ + `${partitionBy} IN UNNEST(partitions_for_replacement)`, + updatePartitionFilter ? `and DATAFORM_DEST.${updatePartitionFilter}` : "", + incrementalPredicatesString, + ] + .filter(Boolean) + .join(" "); + + sql += ` +BEGIN + DECLARE partitions_for_replacement DEFAULT ( + ARRAY( + SELECT DISTINCT ${partitionBy} + FROM \`${tempTableName}\` + WHERE ${partitionBy} IS NOT NULL + ) + ); + + EXECUTE IMMEDIATE ( + "MERGE ${this.escapeSqlString(qualifiedTargetTableName)} DATAFORM_DEST " || + "USING \`${tempTableName}\` DATAFORM_SOURCE " || + "ON FALSE " || + "WHEN NOT MATCHED BY SOURCE AND ${this.escapeSqlString(notMatchedBySourceCondition)} THEN " || + "DELETE " || + "WHEN NOT MATCHED BY TARGET THEN " || + "INSERT (" || dataform_columns_list || ") VALUES (" || dataform_columns_list || ")" + ); +END;`; + break; + } + case dataform.IncrementalStrategy.MERGE: + default: { + if (isMerge) { + const updatePartitionFilter = table.bigquery && table.bigquery.updatePartitionFilter; + const incrementalPredicates = table.bigquery && table.bigquery.incrementalPredicates; + const incrementalPredicatesString = + this.buildIncrementalPredicatesString(incrementalPredicates); + const onCondition = [ + table.uniqueKey + .map( + (uniqueKeyCol) => `DATAFORM_DEST.${uniqueKeyCol} = DATAFORM_SOURCE.${uniqueKeyCol}`, + ) + .join(" and "), + updatePartitionFilter ? `and DATAFORM_DEST.${updatePartitionFilter}` : "", + incrementalPredicatesString, + ] + .filter(Boolean) + .join(" "); + + sql += ` +EXECUTE IMMEDIATE ( + "MERGE ${this.escapeSqlString(qualifiedTargetTableName)} DATAFORM_DEST " || + "USING \`${tempTableName}\` DATAFORM_SOURCE " || + "ON ${this.escapeSqlString(onCondition)} " || + "WHEN MATCHED THEN " || + "UPDATE SET " || dataform_columns_merge || " " || + "WHEN NOT MATCHED THEN " || + "INSERT (" || dataform_columns_list || ") VALUES (" || dataform_columns_list || ")" +);`; + } else { + sql += ` +EXECUTE IMMEDIATE ( + "INSERT INTO ${this.escapeSqlString(qualifiedTargetTableName)} (" || dataform_columns_list || ") " || + "SELECT " || dataform_columns_list || " FROM \`${tempTableName}\`" +);`; + } + break; + } + } + + return sql; + } + + private cleanupSql(emptyTempTableName: string, tempTableName: string): string { return ` -- Cleanup temporary tables. DROP TABLE IF EXISTS ${emptyTempTableName}; +DROP TABLE IF EXISTS \`${tempTableName}\`; `; } @@ -412,16 +545,17 @@ DROP TABLE IF EXISTS ${emptyTempTableName}; table: dataform.ITable, qualifiedTargetTableName: string, emptyTempTableTarget: dataform.ITarget, + tempTableName: string, ): string { const emptyTempTableName = this.resolveTarget(emptyTempTableTarget); const query = this.getIncrementalQuery(table); - const onSchemaChange = table.onSchemaChange || dataform.OnSchemaChange.IGNORE; const statements: string[] = [ - this.declareSchemaChangeVariablesSql(onSchemaChange), + this.declareSchemaChangeVariablesSql(table), this.createEmptyTempTableSql(emptyTempTableName, query), this.compareSchemasSql(table.target, emptyTempTableTarget), this.applySchemaChangeStrategySql(table, qualifiedTargetTableName), - this.cleanupSql(emptyTempTableName), + this.executeDynamicDmlSql(table, qualifiedTargetTableName, tempTableName, query), + this.cleanupSql(emptyTempTableName, tempTableName), ]; return statements.join("\n\n"); diff --git a/cli/api/execution_sql_test.ts b/cli/api/execution_sql_test.ts index 2aee473e4..63f6f1df8 100644 --- a/cli/api/execution_sql_test.ts +++ b/cli/api/execution_sql_test.ts @@ -178,6 +178,64 @@ suite("ExecutionSql with 'onSchemaChange'", () => { } } }); + + test("executes dynamic DML inside procedure and does not emit static DML outside procedure for onSchemaChange", () => { + for (const strategy of [ + dataform.OnSchemaChange.FAIL, + dataform.OnSchemaChange.EXTEND, + dataform.OnSchemaChange.SYNCHRONIZE, + ]) { + const table = { + ...baseTable, + onSchemaChange: strategy, + uniqueKey: ["id"], + }; + const tasks = executionSql.publishTasks(table, { fullRefresh: false }, tableMetadata); + const builtTasks = tasks.build(); + expect(builtTasks.length).to.equal(1); + + const fullSql = builtTasks[0].statement; + const [procedurePart, ...rest] = fullSql.split("\nEND;\n"); + const afterProcedurePart = rest.join("\nEND;\n"); + expect(procedurePart).to.include("SET dataform_columns_list ="); + expect(procedurePart).to.include("SET dataform_columns_merge ="); + expect(procedurePart).to.include("EXECUTE IMMEDIATE"); + expect(afterProcedurePart).to.not.include("insert into"); + expect(afterProcedurePart).to.not.include("merge "); + expect(afterProcedurePart).to.not.include("field1"); + } + }); + + test("error handler outside the procedure only drops fully qualified tables", () => { + // The staging table is scoped to the stored procedure, so it is not resolvable + // from the calling script. Referencing it there raises "must be qualified with a + // dataset", which masks the original error and leaks the procedure. + for (const strategy of [ + dataform.OnSchemaChange.FAIL, + dataform.OnSchemaChange.EXTEND, + dataform.OnSchemaChange.SYNCHRONIZE, + ]) { + const table = { + ...baseTable, + onSchemaChange: strategy, + uniqueKey: ["id"], + }; + const fullSql = executionSql + .publishTasks(table, { fullRefresh: false }, tableMetadata) + .build()[0].statement; + + const errorHandler = fullSql.split("EXCEPTION WHEN ERROR THEN")[1]; + const droppedTables = [...errorHandler.matchAll(/DROP TABLE IF EXISTS `([^`]+)`/g)].map( + (match) => match[1], + ); + for (const droppedTable of droppedTables) { + expect( + droppedTable, + `error handler for ${dataform.OnSchemaChange[strategy]} drops an unqualified table`, + ).to.match(/^[^.]+\.[^.]+\.[^.]+$/); + } + } + }); }); suite("ExecutionSql for property graphs", () => { diff --git a/cli/api/goldens/insert_overwrite_extend.sql b/cli/api/goldens/insert_overwrite_extend.sql index 89e6c7255..6792a29e0 100644 --- a/cli/api/goldens/insert_overwrite_extend.sql +++ b/cli/api/goldens/insert_overwrite_extend.sql @@ -4,6 +4,7 @@ BEGIN -- Declare variables for schema comparison and strategy execution. DECLARE dataform_columns ARRAY; +DECLARE dataform_columns_list STRING; DECLARE temp_table_columns ARRAY>; DECLARE columns_added ARRAY>; DECLARE columns_removed ARRAY; @@ -60,19 +61,13 @@ 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 ( +-- Prepare dynamic column lists and staging table for DML. +SET dataform_columns_list = ( + SELECT STRING_AGG(FORMAT("`%s`", column_info.column_name), ", ") + FROM UNNEST(temp_table_columns) AS column_info +); + +CREATE OR REPLACE TEMP TABLE `incremental_on_schema_change_df_temp_test_uuid_temp` AS ( select 1 as id, 'a' as field1, 'new' as field2 ); @@ -80,20 +75,33 @@ BEGIN DECLARE partitions_for_replacement DEFAULT ( ARRAY( SELECT DISTINCT DATE(ts) - FROM `staging_table_temp_test_uuid` + FROM `incremental_on_schema_change_df_temp_test_uuid_temp` 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`); + EXECUTE IMMEDIATE ( + "MERGE `project-id.dataset-id.incremental_on_schema_change` DATAFORM_DEST " || + "USING `incremental_on_schema_change_df_temp_test_uuid_temp` 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 (" || dataform_columns_list || ") VALUES (" || dataform_columns_list || ")" + ); END; -DROP TABLE IF EXISTS `staging_table_temp_test_uuid`; + +-- Cleanup temporary tables. +DROP TABLE IF EXISTS `project-id.dataset-id.incremental_on_schema_change_df_temp_test_uuid_empty`; +DROP TABLE IF EXISTS `incremental_on_schema_change_df_temp_test_uuid_temp`; + +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`; diff --git a/cli/api/goldens/on_schema_change_extend.sql b/cli/api/goldens/on_schema_change_extend.sql index 0b6b520d0..8d548843f 100644 --- a/cli/api/goldens/on_schema_change_extend.sql +++ b/cli/api/goldens/on_schema_change_extend.sql @@ -4,6 +4,7 @@ BEGIN -- Declare variables for schema comparison and strategy execution. DECLARE dataform_columns ARRAY; +DECLARE dataform_columns_list STRING; DECLARE temp_table_columns ARRAY>; DECLARE columns_added ARRAY>; DECLARE columns_removed ARRAY; @@ -60,8 +61,25 @@ END IF; +-- Prepare dynamic column lists and staging table for DML. +SET dataform_columns_list = ( + SELECT STRING_AGG(FORMAT("`%s`", column_info.column_name), ", ") + FROM UNNEST(temp_table_columns) AS column_info +); + +CREATE OR REPLACE TEMP TABLE `incremental_on_schema_change_df_temp_test_uuid_temp` AS ( + select 1 as id, 'a' as field1, 'new' as field2 +); + +EXECUTE IMMEDIATE ( + "INSERT INTO `project-id.dataset-id.incremental_on_schema_change` (" || dataform_columns_list || ") " || + "SELECT " || dataform_columns_list || " FROM `incremental_on_schema_change_df_temp_test_uuid_temp`" +); + + -- Cleanup temporary tables. DROP TABLE IF EXISTS `project-id.dataset-id.incremental_on_schema_change_df_temp_test_uuid_empty`; +DROP TABLE IF EXISTS `incremental_on_schema_change_df_temp_test_uuid_temp`; END; BEGIN @@ -72,7 +90,3 @@ EXCEPTION WHEN ERROR THEN RAISE; END; DROP PROCEDURE IF EXISTS `project-id.dataset-id.df_osc_test_uuid`; -insert into `project-id.dataset-id.incremental_on_schema_change` -(`id`,`field1`) -select `id`,`field1` -from (select 1 as id, 'a' as field1, 'new' as field2) as insertions \ No newline at end of file diff --git a/cli/api/goldens/on_schema_change_fail.sql b/cli/api/goldens/on_schema_change_fail.sql index 17196b41a..57b1094c2 100644 --- a/cli/api/goldens/on_schema_change_fail.sql +++ b/cli/api/goldens/on_schema_change_fail.sql @@ -4,6 +4,7 @@ BEGIN -- Declare variables for schema comparison and strategy execution. DECLARE dataform_columns ARRAY; +DECLARE dataform_columns_list STRING; DECLARE temp_table_columns ARRAY>; DECLARE columns_added ARRAY>; DECLARE columns_removed ARRAY; @@ -51,8 +52,25 @@ END IF; +-- Prepare dynamic column lists and staging table for DML. +SET dataform_columns_list = ( + SELECT STRING_AGG(FORMAT("`%s`", column_info.column_name), ", ") + FROM UNNEST(temp_table_columns) AS column_info +); + +CREATE OR REPLACE TEMP TABLE `incremental_on_schema_change_df_temp_test_uuid_temp` AS ( + select 1 as id, 'a' as field1, 'new' as field2 +); + +EXECUTE IMMEDIATE ( + "INSERT INTO `project-id.dataset-id.incremental_on_schema_change` (" || dataform_columns_list || ") " || + "SELECT " || dataform_columns_list || " FROM `incremental_on_schema_change_df_temp_test_uuid_temp`" +); + + -- Cleanup temporary tables. DROP TABLE IF EXISTS `project-id.dataset-id.incremental_on_schema_change_df_temp_test_uuid_empty`; +DROP TABLE IF EXISTS `incremental_on_schema_change_df_temp_test_uuid_temp`; END; BEGIN @@ -63,7 +81,3 @@ EXCEPTION WHEN ERROR THEN RAISE; END; DROP PROCEDURE IF EXISTS `project-id.dataset-id.df_osc_test_uuid`; -insert into `project-id.dataset-id.incremental_on_schema_change` -(`id`,`field1`) -select `id`,`field1` -from (select 1 as id, 'a' as field1, 'new' as field2) as insertions \ 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 026e9f4dc..b56daaaa7 100644 --- a/cli/api/goldens/on_schema_change_synchronize.sql +++ b/cli/api/goldens/on_schema_change_synchronize.sql @@ -4,6 +4,8 @@ BEGIN -- Declare variables for schema comparison and strategy execution. DECLARE dataform_columns ARRAY; +DECLARE dataform_columns_list STRING; +DECLARE dataform_columns_merge STRING; DECLARE temp_table_columns ARRAY>; DECLARE columns_added ARRAY>; DECLARE columns_removed ARRAY; @@ -75,8 +77,34 @@ END IF; +-- Prepare dynamic column lists and staging table for DML. +SET dataform_columns_list = ( + SELECT STRING_AGG(FORMAT("`%s`", column_info.column_name), ", ") + FROM UNNEST(temp_table_columns) AS column_info +); +SET dataform_columns_merge = ( + SELECT STRING_AGG(FORMAT("`%s` = DATAFORM_SOURCE.`%s`", column_info.column_name, column_info.column_name), ", ") + FROM UNNEST(temp_table_columns) AS column_info +); + +CREATE OR REPLACE TEMP TABLE `incremental_on_schema_change_df_temp_test_uuid_temp` AS ( + select 1 as id, 'a' as field1, 'new' as field2 +); + +EXECUTE IMMEDIATE ( + "MERGE `project-id.dataset-id.incremental_on_schema_change` DATAFORM_DEST " || + "USING `incremental_on_schema_change_df_temp_test_uuid_temp` DATAFORM_SOURCE " || + "ON DATAFORM_DEST.id = DATAFORM_SOURCE.id " || + "WHEN MATCHED THEN " || + "UPDATE SET " || dataform_columns_merge || " " || + "WHEN NOT MATCHED THEN " || + "INSERT (" || dataform_columns_list || ") VALUES (" || dataform_columns_list || ")" +); + + -- Cleanup temporary tables. DROP TABLE IF EXISTS `project-id.dataset-id.incremental_on_schema_change_df_temp_test_uuid_empty`; +DROP TABLE IF EXISTS `incremental_on_schema_change_df_temp_test_uuid_temp`; END; BEGIN @@ -87,12 +115,3 @@ EXCEPTION WHEN ERROR THEN RAISE; END; DROP PROCEDURE IF EXISTS `project-id.dataset-id.df_osc_test_uuid`; -merge `project-id.dataset-id.incremental_on_schema_change` DATAFORM_DEST -using (select 1 as id, 'a' as field1, 'new' as field2 -) DATAFORM_SOURCE -on DATAFORM_DEST.id = DATAFORM_SOURCE.id - -when matched then - 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/index_run_e2e_test.ts b/cli/index_run_e2e_test.ts index 159e49fb2..0799898d2 100644 --- a/cli/index_run_e2e_test.ts +++ b/cli/index_run_e2e_test.ts @@ -10,7 +10,7 @@ import { INTEGRATION_TEST_PROJECT, INTEGRATION_TEST_RESERVATION, runCli, - setupProject + setupProject, } from "df/cli/index_test_base"; import { version } from "df/core/version"; import { suite, test, writeDefinitionFile } from "df/testing"; @@ -30,19 +30,16 @@ suite("run e2e", ({ afterEach }) => { ` config { type: "table", tags: ["someTag"] } select 1 as \${dataform.projectConfig.vars.testVar2} -` +`, ); // Compile the project using the CLI. - const compileResult = await runCli( - "compile", - [ - projectDir, - "--json", - "--vars=testVar1=testValue1,testVar2=testValue2", - "--schema-suffix=test_schema_suffix" - ] - ); + const compileResult = await runCli("compile", [ + projectDir, + "--json", + "--vars=testVar1=testValue1,testVar2=testValue2", + "--schema-suffix=test_schema_suffix", + ]); expect(compileResult.exitCode).equals(0); @@ -54,19 +51,19 @@ select 1 as \${dataform.projectConfig.vars.testVar2} target: { database: INTEGRATION_TEST_PROJECT, schema: "dataform_test_schema_suffix", - name: "example" + name: "example", }, canonicalTarget: { schema: "dataform", name: "example", - database: INTEGRATION_TEST_PROJECT + database: INTEGRATION_TEST_PROJECT, }, query: "\n\nselect 1 as testValue2\n", disabled: false, fileName: "definitions/example.sqlx", hermeticity: "NON_HERMETIC", - tags: ["someTag"] - } + tags: ["someTag"], + }, ], projectConfig: { warehouse: "bigquery", @@ -76,9 +73,9 @@ select 1 as \${dataform.projectConfig.vars.testVar2} defaultLocation: INTEGRATION_TEST_LOCATION, vars: { testVar1: "testValue1", - testVar2: "testValue2" + testVar2: "testValue2", }, - schemaSuffix: "test_schema_suffix" + schemaSuffix: "test_schema_suffix", }, graphErrors: {}, jitData: {}, @@ -87,26 +84,23 @@ select 1 as \${dataform.projectConfig.vars.testVar2} { database: INTEGRATION_TEST_PROJECT, schema: "dataform", - name: "example" - } - ] + name: "example", + }, + ], }); // Dry run the project. - const runResult = await runCli( - "run", - [ - projectDir, - "--credentials", - CREDENTIALS_PATH, - "--dry-run", - "--json", - "--vars=testVar1=testValue1,testVar2=testValue2", - "--default-location=europe", - "--tags=someTag,someOtherTag", - "--actions=example,someOtherAction" - ] - ); + const runResult = await runCli("run", [ + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--dry-run", + "--json", + "--vars=testVar1=testValue1,testVar2=testValue2", + "--default-location=europe", + "--tags=someTag,someOtherTag", + "--actions=example,someOtherAction", + ]); if (runResult.exitCode !== 0 || runResult.stdout.trim().length === 0) { console.error("GOLDEN PATH FAILED. STDERR:", runResult.stderr); @@ -122,17 +116,16 @@ select 1 as \${dataform.projectConfig.vars.testVar2} target: { database: INTEGRATION_TEST_PROJECT, name: "example", - schema: "dataform" + schema: "dataform", }, tasks: [ { - statement: - `create or replace table \`${INTEGRATION_TEST_PROJECT}.dataform.example\` as \n\nselect 1 as testValue2`, - type: "statement" - } + statement: `create or replace table \`${INTEGRATION_TEST_PROJECT}.dataform.example\` as \n\nselect 1 as testValue2`, + type: "statement", + }, ], - type: "table" - } + type: "table", + }, ], jitData: {}, projectConfig: { @@ -143,15 +136,15 @@ select 1 as \${dataform.projectConfig.vars.testVar2} warehouse: "bigquery", vars: { testVar1: "testValue1", - testVar2: "testValue2" - } + testVar2: "testValue2", + }, }, runConfig: { fullRefresh: false, tags: ["someTag", "someOtherTag"], - actions: ["example", "someOtherAction"] + actions: ["example", "someOtherAction"], }, - warehouseState: {} + warehouseState: {}, }); }); @@ -168,7 +161,7 @@ select 1 as \${dataform.projectConfig.vars.testVar2} ` config { type: "assertion" } SELECT 1 WHERE FALSE -` +`, ); writeDefinitionFile( @@ -182,7 +175,7 @@ config { } } SELECT 1 as id -` +`, ); }); @@ -195,16 +188,15 @@ SELECT 1 as id target: { database: INTEGRATION_TEST_PROJECT, name: "example_table", - schema: "dataform" + schema: "dataform", }, tasks: [ { - statement: - `create or replace table \`${INTEGRATION_TEST_PROJECT}.dataform.example_table\` as \n\nSELECT 1 as id`, - type: "statement" - } + statement: `create or replace table \`${INTEGRATION_TEST_PROJECT}.dataform.example_table\` as \n\nSELECT 1 as id`, + type: "statement", + }, ], - type: "table" + type: "table", }, { fileName: "definitions/test_assertion.sqlx", @@ -212,10 +204,10 @@ SELECT 1 as id target: { database: INTEGRATION_TEST_PROJECT, name: "test_assertion", - schema: "dataform_assertions" + schema: "dataform_assertions", }, - type: "assertion" - } + type: "assertion", + }, ], jitData: {}, projectConfig: { @@ -224,30 +216,27 @@ SELECT 1 as id defaultLocation: INTEGRATION_TEST_LOCATION, defaultSchema: "dataform", disableAssertions: true, - warehouse: "bigquery" + warehouse: "bigquery", }, runConfig: { actions: ["test_assertion", "example_table"], - fullRefresh: false + fullRefresh: false, }, - warehouseState: {} + warehouseState: {}, }; test("with --disable-assertions flag", async () => { alterWorkflowSettings(projectDir, { disableAssertions: false }); - const runResult = await runCli( - "run", - [ - projectDir, - "--credentials", - CREDENTIALS_PATH, - "--dry-run", - "--json", - "--disable-assertions", - "--actions=test_assertion,example_table" - ] - ); + const runResult = await runCli("run", [ + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--dry-run", + "--json", + "--disable-assertions", + "--actions=test_assertion,example_table", + ]); if (runResult.exitCode !== 0 || runResult.stdout.trim().length === 0) { console.error("ASSERTIONS TEST FAILED. STDERR:", runResult.stderr); @@ -259,17 +248,14 @@ SELECT 1 as id test("with disableAssertions set in workflow_settings.yaml", async () => { alterWorkflowSettings(projectDir, { disableAssertions: true }); - const runResult = await runCli( - "run", - [ - projectDir, - "--credentials", - CREDENTIALS_PATH, - "--dry-run", - "--json", - "--actions=test_assertion,example_table" - ] - ); + const runResult = await runCli("run", [ + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--dry-run", + "--json", + "--actions=test_assertion,example_table", + ]); if (runResult.exitCode !== 0 || runResult.stdout.trim().length === 0) { console.error("ASSERTIONS TEST FAILED. STDERR:", runResult.stderr); @@ -281,19 +267,16 @@ SELECT 1 as id test("with --job-labels flag", async () => { alterWorkflowSettings(projectDir, { disableAssertions: false }); - const runResult = await runCli( - "run", - [ - projectDir, - "--credentials", - CREDENTIALS_PATH, - "--dry-run", - "--json", - "--disable-assertions", - "--actions=test_assertion,example_table", - "--job-labels=env=testing,team=dataform" - ] - ); + const runResult = await runCli("run", [ + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--dry-run", + "--json", + "--disable-assertions", + "--actions=test_assertion,example_table", + "--job-labels=env=testing,team=dataform", + ]); if (runResult.exitCode !== 0 || runResult.stdout.trim().length === 0) { console.error("ASSERTIONS TEST FAILED. STDERR:", runResult.stderr); @@ -303,7 +286,6 @@ SELECT 1 as id }); }); - suite("--default-reservation flag", ({ beforeEach }) => { let projectDir: string; @@ -317,19 +299,16 @@ SELECT 1 as id ` config { type: "table" } SELECT 1 as id -` +`, ); }); test("--default-reservation flag is applied to projectConfig in compile output", async () => { - const compileResult = await runCli( - "compile", - [ - projectDir, - "--json", - `--default-reservation=${INTEGRATION_TEST_RESERVATION}` - ] - ); + const compileResult = await runCli("compile", [ + projectDir, + "--json", + `--default-reservation=${INTEGRATION_TEST_RESERVATION}`, + ]); expect(compileResult.exitCode).equals(0); const compiledGraph = JSON.parse(compileResult.stdout); @@ -339,23 +318,20 @@ SELECT 1 as id assertionSchema: "dataform_assertions", defaultDatabase: INTEGRATION_TEST_PROJECT, defaultLocation: INTEGRATION_TEST_LOCATION, - defaultReservation: INTEGRATION_TEST_RESERVATION + defaultReservation: INTEGRATION_TEST_RESERVATION, }); }); test("--default-reservation flag is applied to projectConfig in run (dry-run) output", async () => { - const runResult = await runCli( - "run", - [ - projectDir, - "--credentials", - CREDENTIALS_PATH, - "--dry-run", - "--json", - `--default-reservation=${INTEGRATION_TEST_RESERVATION}`, - "--actions=example_table" - ] - ); + const runResult = await runCli("run", [ + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--dry-run", + "--json", + `--default-reservation=${INTEGRATION_TEST_RESERVATION}`, + "--actions=example_table", + ]); expect(runResult.exitCode).equals(0); const executionGraph = JSON.parse(runResult.stdout); @@ -365,7 +341,7 @@ SELECT 1 as id assertionSchema: "dataform_assertions", defaultDatabase: INTEGRATION_TEST_PROJECT, defaultLocation: INTEGRATION_TEST_LOCATION, - defaultReservation: INTEGRATION_TEST_RESERVATION + defaultReservation: INTEGRATION_TEST_RESERVATION, }); }); }); @@ -378,12 +354,12 @@ SELECT 1 as id await setupProject(tmpDirFixture, projectDir); // Write a simple file to the project. writeDefinitionFile( - projectDir, - "example.sqlx", - ` + projectDir, + "example.sqlx", + ` config { type: "table" } select 1 -` +`, ); }); @@ -392,30 +368,29 @@ select 1 writeDefinitionFile( projectDir, "example_test.sqlx", - ` + ` config { type: "test", dataset: "example" } select 1 -` +`, ); // Run tests using the CLI. - const testResult = await runCli( - "test", - [ - projectDir, - "--credentials", - CREDENTIALS_PATH, - "--json" - ] - ); + const testResult = await runCli("test", [ + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--json", + ]); expect(testResult.exitCode).equals(0); - expect(JSON.parse(testResult.stdout)).deep.equals([ { - "name": "example_test", - "successful": true, - }]); - }); + expect(JSON.parse(testResult.stdout)).deep.equals([ + { + name: "example_test", + successful: true, + }, + ]); + }); test("golden with failed unit test", async () => { // Write a simple failing test to the project. @@ -425,38 +400,35 @@ select 1 ` config { type: "test", dataset: "example" } select 2 -` +`, ); // Run tests using the CLI. - const testResult = await runCli( - "test", - [ - projectDir, - "--credentials", - CREDENTIALS_PATH, - "--json" - ] - ); + const testResult = await runCli("test", [ + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--json", + ]); expect(testResult.exitCode).equals(1); - expect(JSON.parse(testResult.stdout)).deep.equals([{ - "name": "example_test", - "successful": false, - messages: [ - "For row 0 and column \"f0_\": expected \"2\", but saw \"1\"." - ] - }]); + expect(JSON.parse(testResult.stdout)).deep.equals([ + { + name: "example_test", + successful: false, + messages: ['For row 0 and column "f0_": expected "2", but saw "1".'], + }, + ]); }); - }); suite("onSchemaChange", ({ beforeEach }) => { let projectDir: string; - const uniqueDataset = `dataform_e2e_osc_${Math.random().toString(36).substring(7)}`; + let uniqueDataset: string; beforeEach("setup test project", async () => { + uniqueDataset = `dataform_e2e_osc_${Math.random().toString(36).substring(7)}`; projectDir = tmpDirFixture.createNewTmpDir(); await setupProject(tmpDirFixture, projectDir, { defaultDataset: uniqueDataset }); @@ -468,7 +440,7 @@ config { type: "operations" } CREATE OR REPLACE TABLE \`\${dataform.projectConfig.defaultDatabase}.\${dataform.projectConfig.defaultSchema}.example_incremental\` AS SELECT 1 AS id, 'old' AS field1 -` +`, ); writeDefinitionFile( @@ -480,7 +452,57 @@ config { onSchemaChange: "EXTEND" } SELECT 1 as id, 'new' as field1, 'new2' as field2 -` +`, + ); + + writeDefinitionFile( + projectDir, + "verify_extend.sqlx", + ` +config { + type: "assertion" +} +SELECT * FROM \${ref("example_incremental")} +WHERE (field1 = 'new' AND (field2 IS NULL OR field2 != 'new2')) + OR (SELECT COUNT(*) FROM \${ref("example_incremental")}) != 2 +`, + ); + + writeDefinitionFile( + projectDir, + "setup_synchronize_table.sqlx", + ` +config { + type: "operations" +} +CREATE OR REPLACE TABLE \`\${dataform.projectConfig.defaultDatabase}.\${dataform.projectConfig.defaultSchema}.example_synchronize\` AS SELECT 1 AS id, 'old' AS field1 +`, + ); + + writeDefinitionFile( + projectDir, + "example_synchronize.sqlx", + ` +config { + type: "incremental", + uniqueKey: ["id"], + onSchemaChange: "SYNCHRONIZE" +} +SELECT 1 as id, 'synced' as field2 +`, + ); + + writeDefinitionFile( + projectDir, + "verify_synchronize.sqlx", + ` +config { + type: "assertion" +} +SELECT * FROM \${ref("example_synchronize")} +WHERE id != 1 OR field2 IS NULL OR field2 != 'synced' + OR (SELECT COUNT(*) FROM \${ref("example_synchronize")}) != 1 +`, ); writeDefinitionFile( @@ -491,95 +513,111 @@ config { type: "operations" } DROP SCHEMA IF EXISTS \`\${dataform.projectConfig.defaultDatabase}.\${dataform.projectConfig.defaultSchema}\` CASCADE -` +`, ); }); - test("generates dynamic SQL for EXTEND when table exists in BigQuery", { timeout: 120000 }, async () => { - try { - // Run setup operation to create the table in BigQuery. - // Dataform will automatically create the uniqueDataset schema. - await runCli( - "run", - [ + test( + "executes EXTEND and SYNCHRONIZE strategies in BigQuery and populates dynamically altered columns", + { timeout: 120000 }, + async () => { + try { + // Run setup operations to create the initial tables in BigQuery. + // Dataform will automatically create the uniqueDataset schema. + const setupResult = await runCli("run", [ projectDir, "--credentials", CREDENTIALS_PATH, - "--actions=setup_table" - ] - ); - - // Run the incremental table in dry-run mode. - // Dataform will detect the table exists and generate the dynamic procedural SQL. - const runResult = await runCli( - "run", - [ + "--actions=setup_table,setup_synchronize_table", + ]); + expect(setupResult.exitCode).equals(0); + + // Run the incremental table in dry-run mode first to verify procedural SQL structure. + const dryRunResult = await runCli("run", [ projectDir, "--credentials", CREDENTIALS_PATH, "--dry-run", "--json", - "--actions=example_incremental" - ] - ); - - expect(runResult.exitCode).equals(0); - const executionGraph = JSON.parse(runResult.stdout); - const statement = executionGraph.actions[0].tasks[0].statement; - - const expectedRunResult = { - projectConfig: { - warehouse: "bigquery", - defaultSchema: uniqueDataset, - assertionSchema: "dataform_assertions", - defaultDatabase: INTEGRATION_TEST_PROJECT, - defaultLocation: INTEGRATION_TEST_LOCATION - }, - runConfig: { - actions: ["example_incremental"], - fullRefresh: false - }, - actions: [ - { - fileName: "definitions/example_incremental.sqlx", - hermeticity: "NON_HERMETIC", - tableType: "incremental", - target: { - database: INTEGRATION_TEST_PROJECT, - name: "example_incremental", - schema: uniqueDataset + "--actions=example_incremental", + ]); + + expect(dryRunResult.exitCode).equals(0); + const executionGraph = JSON.parse(dryRunResult.stdout); + const statement = executionGraph.actions[0].tasks[0].statement; + + const expectedRunResult = { + projectConfig: { + warehouse: "bigquery", + defaultSchema: uniqueDataset, + assertionSchema: "dataform_assertions", + defaultDatabase: INTEGRATION_TEST_PROJECT, + defaultLocation: INTEGRATION_TEST_LOCATION, + }, + runConfig: { + actions: ["example_incremental"], + fullRefresh: false, + }, + actions: [ + { + fileName: "definitions/example_incremental.sqlx", + hermeticity: "NON_HERMETIC", + tableType: "incremental", + target: { + database: INTEGRATION_TEST_PROJECT, + name: "example_incremental", + schema: uniqueDataset, + }, + tasks: [ + { + statement, + type: "statement", + }, + ], + type: "table", }, - tasks: [ - { - statement, - type: "statement" - } - ], - type: "table" - } - ], - jitData: {}, - warehouseState: executionGraph.warehouseState - }; - - expect(executionGraph).deep.equals(expectedRunResult); - expect(statement).to.include("CREATE OR REPLACE PROCEDURE"); - expect(statement).to.include("Column removals are not allowed when on_schema_change = 'EXTEND'."); - expect(statement).to.include("ALTER TABLE"); - expect(statement).to.include("ADD COLUMN IF NOT EXISTS"); - } finally { - // Teardown the schema completely, regardless of test success or failure. - await runCli( - "run", - [ + ], + jitData: {}, + warehouseState: executionGraph.warehouseState, + }; + + expect(executionGraph).deep.equals(expectedRunResult); + expect(statement).to.include("CREATE OR REPLACE PROCEDURE"); + expect(statement).to.include( + "Column removals are not allowed when on_schema_change = 'EXTEND'.", + ); + expect(statement).to.include("ALTER TABLE"); + expect(statement).to.include("ADD COLUMN IF NOT EXISTS"); + expect(statement).to.include("EXECUTE IMMEDIATE"); + + // Execute EXTEND live against BigQuery and verify new column is populated. + const extendRunResult = await runCli("run", [ projectDir, "--credentials", CREDENTIALS_PATH, - "--actions=teardown_schema" - ] - ); - } - }); + "--actions=example_incremental,verify_extend", + ]); + expect(extendRunResult.exitCode).equals(0); + + // Execute SYNCHRONIZE live against BigQuery (drops field1, adds field2) and verify result. + const syncRunResult = await runCli("run", [ + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--actions=example_synchronize,verify_synchronize", + ]); + expect(syncRunResult.exitCode).equals(0); + } finally { + // Teardown the schema completely, regardless of test success or failure. + await runCli("run", [ + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--actions=teardown_schema", + ]); + } + }, + ); }); suite("run --timeout deprecation", ({ beforeEach }) => { @@ -589,7 +627,7 @@ DROP SCHEMA IF EXISTS \`\${dataform.projectConfig.defaultDatabase}.\${dataform.p projectDir = tmpDirFixture.createNewTmpDir(); fs.writeFileSync( path.join(projectDir, "workflow_settings.yaml"), - `defaultProject: ${INTEGRATION_TEST_PROJECT}\ndefaultLocation: ${INTEGRATION_TEST_LOCATION}\n` + `defaultProject: ${INTEGRATION_TEST_PROJECT}\ndefaultLocation: ${INTEGRATION_TEST_LOCATION}\n`, ); }); @@ -597,39 +635,31 @@ DROP SCHEMA IF EXISTS \`\${dataform.projectConfig.defaultDatabase}.\${dataform.p // The notice fires in the run handler before compile. Yargs validation // requires workflow_settings.yaml to exist, but compile can fail after that // — we only assert on stderr for the notice line. - const runResult = await runCli( - "run", - [ - projectDir, - "--credentials", - CREDENTIALS_PATH, - "--timeout", - "30s" - ] - ); + const runResult = await runCli("run", [ + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--timeout", + "30s", + ]); expect(runResult.stderr).to.match( - /--timeout only bounds project compilation[\s\S]*use --execution-timeout/ + /--timeout only bounds project compilation[\s\S]*use --execution-timeout/, ); }); test("--timeout on run does NOT emit notice when --execution-timeout is also set", async () => { - const runResult = await runCli( - "run", - [ - projectDir, - "--credentials", - CREDENTIALS_PATH, - "--timeout", - "30s", - "--execution-timeout", - "10m" - ] - ); + const runResult = await runCli("run", [ + projectDir, + "--credentials", + CREDENTIALS_PATH, + "--timeout", + "30s", + "--execution-timeout", + "10m", + ]); - expect(runResult.stderr).to.not.match( - /--timeout only bounds project compilation/ - ); + expect(runResult.stderr).to.not.match(/--timeout only bounds project compilation/); }); - }) + }); });