Execute dynamic DML inside onSchemaChange procedure (#2307)

[apilaskowski]

Fixes #2307 (together with #2319, which this is stacked on — review that first).

Problem

ExecutionSql.publishTasks emitted the schema-change stored procedure and then fell through to a static INSERT / MERGE / INSERT_OVERWRITE built from tableMetadata.fields. tableMetadata is fetched once, before execution, so the DML always used the pre-alteration column list:

• EXTEND — newly added columns were omitted from the INSERT list and silently left NULL.
• SYNCHRONIZE — dropped columns were still referenced, failing at runtime with Unrecognized name: <dropped_col>.

The correct column list only exists after the schema-change DDL has run, i.e. at BigQuery script runtime. It cannot be computed by the CLI at compile time, so the DML has to move inside the procedure.

Solution

Generate the DML dynamically inside the stored procedure, matching the behaviour of the managed Dataform service:

1. publishTasks now breaks instead of falling through, so no static DML is emitted for FAIL / EXTEND / SYNCHRONIZE. IGNORE is unchanged.
2. dataform_columns_list (plus dataform_columns_merge for merge-strategy tables) is declared up front and SET from temp_table_columns after the schema-change DDL.
3. The incremental query is materialised into a _temp staging table.
4. The INSERT / MERGE / INSERT_OVERWRITE runs via EXECUTE IMMEDIATE, splicing in the runtime column lists.
5. _empty and _temp are cleaned up on completion and on error.

Generated SQL for EXTEND — before

CREATE OR REPLACE PROCEDURE df_osc_x() BEGIN
  ... ALTER TABLE `t` ADD COLUMN IF NOT EXISTS field2 STRING ...
END;
CALL df_osc_x();
insert into `t` (`id`,`field1`)                        -- stale: field2 missing
select `id`,`field1` from (...) as insertions

Generated SQL for EXTEND — after

CREATE OR REPLACE PROCEDURE df_osc_x() BEGIN
  ... ALTER TABLE `t` ADD COLUMN IF NOT EXISTS field2 STRING ...

  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 `t_df_temp_x_temp` AS ( ... );

  EXECUTE IMMEDIATE (
    "INSERT INTO `t` (" || dataform_columns_list || ") " ||
    "SELECT " || dataform_columns_list || " FROM `t_df_temp_x_temp`"
  );
END;
CALL df_osc_x();

Behaviour change

Projects using EXTEND or SYNCHRONIZE will see a different generated script. The incremental query is now materialised into a temporary staging table before the DML instead of being inlined into it, so the query text appears twice in the script (once as the LIMIT 0 schema probe, once as the staging table). IGNORE is unaffected.

Because the procedure queries INFORMATION_SCHEMA.COLUMNS twice and creates the _empty probe table, an incremental run with onSchemaChange set bills a small fixed amount (~40 MiB observed) even for a trivial query. This matches the managed service.

Test coverage

Incremental strategy	onSchemaChange	Golden	Unit	Live E2E
INSERT (no uniqueKey)	FAIL, EXTEND	✅	✅	✅ (EXTEND)
MERGE (uniqueKey)	SYNCHRONIZE	✅	✅	✅
INSERT_OVERWRITE	EXTEND	✅	✅	❌ (#2328)
error path (FAIL trips)	FAIL	✅	✅	❌

• Unit tests (cli/api/execution_sql_test.ts): updated goldens for on_schema_change_{fail,extend,synchronize}.sql and insert_overwrite_extend.sql; a test asserting the dynamic DML is emitted inside the procedure and no static DML referencing stale tableMetadata columns is emitted after it; and a test asserting every table dropped in the outer error handler is fully qualified.
• E2E (cli/index_run_e2e_test.ts): the onSchemaChange suite is upgraded from --dry-run only to executing EXTEND and SYNCHRONIZE live against BigQuery, asserting via Dataform assertions that added columns (field2) are populated with non-null data and dropped columns (field1) are removed. These require GCP credentials and are tagged integration (Mark tests that require valid GCP project credentials as integration tests #2289).

Verified manually against the exact repro from #2307 (type: "incremental", onSchemaChange: "FAIL", SELECT 1 AS a) on BigQuery: run 1 creates the table, run 2 succeeds and appends correctly, and changing the query to SELECT 1 AS a, 2 AS b now fails with the intended message:

Schema mismatch defined by on_schema_change = 'FAIL'. Added columns: [("b", "INT64")], removed columns: []

---

Reviewer note: intentional commit split

1. style(cli): format index_run_e2e_test.ts with prettier — mechanical Prettier formatting of pre-existing code in cli/index_run_e2e_test.ts, required by scripts/lint, with no logic changes. Kept separate so it does not pollute the functional diff.
2. fix(cli): execute dynamic DML inside onSchemaChange procedure (#2307) — the functional change, goldens, unit test and E2E tests.
3. fix(cli): do not drop the procedure-scoped staging table from the error handler — follow-up fix found while testing this branch live; see below.

Reviewing commits 2 and 3 individually gives the cleanest diff.

On commit 3: the outer EXCEPTION handler dropped the _temp staging table by its unqualified name. That table is scoped to the stored procedure, so it is not resolvable from the calling script, and BigQuery resolves table names before evaluating IF EXISTS. The handler therefore raised Table "..._temp" must be qualified with a dataset, which masked the original error and leaked the df_osc_* procedure (because DROP PROCEDURE runs after it). It was most visible with onSchemaChange: "FAIL", where the intended schema-mismatch message never reached the user. The statement was also redundant — cleanupSql already drops the table inside the procedure, and BigQuery drops procedure-scoped temp tables on exit.