From 76204c9aa0a8835b5a12aa39e20dc7c8cdb64367 Mon Sep 17 00:00:00 2001 From: Rafal Hawrylak Date: Tue, 13 Oct 2026 11:12:34 +0000 Subject: [PATCH] feat(cli): emit lineage for assertions and property graphs and align actionType with Backend Every action that runs a BigQuery job now emits lineage events. This covers tables, operations, assertions and property graphs. Declarations are still skipped. The gcp_bq_pipelines_job.actionType attribute now uses the same values as Dataform in Google Cloud: TABLE, OPERATIONS, ASSERTION and PROPERTY_GRAPH. For assertions the externalQuery facet references the CREATE VIEW job that writes the assertion view instead of the row count check query. --- cli/README.md | 2 + cli/api/lineage/action_type.ts | 31 ++++++ cli/api/lineage/action_type_test.ts | 35 ++++++ cli/api/lineage/emitter.ts | 10 +- cli/api/lineage/emitter_test.ts | 115 ++++++++++++++++++-- cli/api/lineage/lineage_integration_test.ts | 57 ++++++++++ cli/api/lineage/payload_builder.ts | 20 +++- cli/api/lineage/payload_builder_test.ts | 76 ++++++++++++- 8 files changed, 328 insertions(+), 18 deletions(-) create mode 100644 cli/api/lineage/action_type.ts create mode 100644 cli/api/lineage/action_type_test.ts diff --git a/cli/README.md b/cli/README.md index e2a171a5c..e151acc88 100644 --- a/cli/README.md +++ b/cli/README.md @@ -376,3 +376,5 @@ dataform run --job-prefix=nightly --job-labels=team=analytics,pipeline=nightly-e ```bash DATAFORM_LINEAGE_DEBUG=1 dataform run --emit-lineage ``` + +Lineage events are emitted for every action that runs a BigQuery job: tables (including views and incremental tables), operations, assertions and property graphs. Assertions appear as a process for the assertion view written to the assertions dataset. Declarations are not emitted. The action type is reported in the `gcp_bq_pipelines_job.actionType` attribute as `TABLE`, `OPERATIONS`, `ASSERTION` or `PROPERTY_GRAPH`, matching the values used by Dataform in Google Cloud. diff --git a/cli/api/lineage/action_type.ts b/cli/api/lineage/action_type.ts new file mode 100644 index 000000000..64523d63c --- /dev/null +++ b/cli/api/lineage/action_type.ts @@ -0,0 +1,31 @@ +import { dataform } from "df/protos/ts"; + +/** + * Lineage action-type rule shared with the Dataform Backend (DD §4.3): + * every action that runs a BigQuery job emits lineage events. Declarations + * and other action types that don't run a BigQuery job are excluded. + * + * Maps the CLI's execution action type to the value sent in + * `gcp_bq_pipelines_job.actionType`. Values mirror the Backend's + * `WorkflowExecutionActionInfo.DataformActionType` enum names so both + * runtimes report the same action type to Knowledge Catalog Lineage. + */ +const LINEAGE_ACTION_TYPES: { readonly [executionType: string]: string } = { + table: "TABLE", + operation: "OPERATIONS", + assertion: "ASSERTION", + propertyGraph: "PROPERTY_GRAPH", +}; + +/** Returns true if lineage events are emitted for the given action. */ +export function isLineageEligibleAction(action: dataform.IExecutionAction): boolean { + return Object.prototype.hasOwnProperty.call(LINEAGE_ACTION_TYPES, action.type || ""); +} + +/** + * Returns the `gcp_bq_pipelines_job.actionType` value for the given action, + * or undefined if the action type is not lineage-eligible. + */ +export function toLineageActionType(action: dataform.IExecutionAction): string | undefined { + return isLineageEligibleAction(action) ? LINEAGE_ACTION_TYPES[action.type] : undefined; +} diff --git a/cli/api/lineage/action_type_test.ts b/cli/api/lineage/action_type_test.ts new file mode 100644 index 000000000..135de3b11 --- /dev/null +++ b/cli/api/lineage/action_type_test.ts @@ -0,0 +1,35 @@ +import { expect } from "chai"; + +import { isLineageEligibleAction, toLineageActionType } from "df/cli/api/lineage/action_type"; +import { dataform } from "df/protos/ts"; +import { suite, test } from "df/testing"; + +function actionOfType(type: string): dataform.IExecutionAction { + return dataform.ExecutionAction.create({ + target: { database: "proj", schema: "schema", name: "name" }, + type, + }); +} + +suite("lineage action_type", () => { + test("actions that run a BigQuery job are eligible and map to Backend enum names", () => { + const mapped = ["table", "operation", "assertion", "propertyGraph"].map((type) => ({ + type, + eligible: isLineageEligibleAction(actionOfType(type)), + actionType: toLineageActionType(actionOfType(type)), + })); + expect(mapped).to.deep.equal([ + { type: "table", eligible: true, actionType: "TABLE" }, + { type: "operation", eligible: true, actionType: "OPERATIONS" }, + { type: "assertion", eligible: true, actionType: "ASSERTION" }, + { type: "propertyGraph", eligible: true, actionType: "PROPERTY_GRAPH" }, + ]); + }); + + test("declarations, unknown and missing types are not eligible", () => { + for (const type of ["declaration", "notebook", "unknown", "", "toString"]) { + expect(isLineageEligibleAction(actionOfType(type)), type).to.equal(false); + expect(toLineageActionType(actionOfType(type)), type).to.equal(undefined); + } + }); +}); diff --git a/cli/api/lineage/emitter.ts b/cli/api/lineage/emitter.ts index 7e835f755..637c49463 100644 --- a/cli/api/lineage/emitter.ts +++ b/cli/api/lineage/emitter.ts @@ -1,5 +1,6 @@ import { LineageClient } from "@google-cloud/lineage"; +import { isLineageEligibleAction } from "df/cli/api/lineage/action_type"; import { GLOBAL_LINEAGE_ENDPOINT, LineageEndpointRouter } from "df/cli/api/lineage/endpoint_router"; import { LineagePayloadBuilder, toProtoStruct } from "df/cli/api/lineage/payload_builder"; import { coerceAsError } from "df/common/errors/errors"; @@ -149,11 +150,10 @@ export class LineageEmitter { return; } - // Non-table/non-operation actions (e.g., assertions, declarations) are not - // emitted. This is a scope decision, not a user-visible misconfiguration, - // so it is intentionally silent. - const isEligibleType = action.type === "table" || action.type === "operation"; - if (!isEligibleType) { + // Only actions that run a BigQuery job emit lineage (see action_type.ts). + // Skipping the rest is a scope decision, not a user-visible + // misconfiguration, so it is intentionally silent. + if (!isLineageEligibleAction(action)) { return; } diff --git a/cli/api/lineage/emitter_test.ts b/cli/api/lineage/emitter_test.ts index 386aada4e..8c88c0fa0 100644 --- a/cli/api/lineage/emitter_test.ts +++ b/cli/api/lineage/emitter_test.ts @@ -613,12 +613,12 @@ suite("LineageEmitter", () => { expect(mockClient.processOpenLineageRunEventCalledWith.length).to.equal(0); }); - test("emits for operation-type actions and propagates action.type to gcp_bq_pipelines_job facet", async () => { - // Positive counterpart to "ineligible action types are skipped": operation - // is a first-class Dataform action (raw SQL block that runs but produces no + test("emits for operation-type actions and maps action.type to gcp_bq_pipelines_job facet", async () => { + // Positive counterpart to "declarations are skipped": operation is a + // first-class Dataform action (raw SQL block that runs but produces no // modeled table). It must not be caught by the eligibility filter, and its - // action.type string must reach the gcp_bq_pipelines_job.actionType facet - // verbatim so downstream consumers can distinguish tables from operations. + // action type must reach the gcp_bq_pipelines_job.actionType facet as the + // Backend's enum name so both runtimes report the same value. const mockClient = new MockLineageClient(); const emitter = new LineageEmitter( credentials, @@ -651,12 +651,12 @@ suite("LineageEmitter", () => { expect(gcpBqPipelinesJob.dataformCoreVersion).to.be.a("string"); expect(gcpBqPipelinesJob).to.deep.equal({ dataformCoreVersion: gcpBqPipelinesJob.dataformCoreVersion, - actionType: "operation", + actionType: "OPERATIONS", actionName: "s.op_action", }); }); - test("ineligible action types are skipped silently (no stderr line, no emission)", async () => { + test("emits START and COMPLETE for assertions with the assertion view as output", async () => { const mockClient = new MockLineageClient(); const stderr = new StderrCapture(); const emitter = new LineageEmitter( @@ -667,9 +667,107 @@ suite("LineageEmitter", () => { ); const assertion = dataform.ExecutionAction.create({ - target: { database: "proj", schema: "schema", name: "assertion" }, + target: { database: "proj", schema: "assertions", name: "assert_a" }, type: "assertion", + dependencyTargets: [{ database: "proj", schema: "schema", name: "table_a" }], + tasks: [ + { + type: "statement", + statement: "create or replace view `proj.assertions.assert_a` as ...", + }, + { + type: "assertion", + statement: "select sum(1) as row_count from `proj.assertions.assert_a`", + }, + ], }); + + emitter.emitForAction( + assertion, + dataform.ActionResult.create({ status: dataform.ActionResult.ExecutionStatus.RUNNING }), + ); + emitter.emitForAction( + assertion, + dataform.ActionResult.create({ status: dataform.ActionResult.ExecutionStatus.SUCCESSFUL }), + ); + await emitter.drain(); + + expect(stderr.writes).to.deep.equal([]); + const events = mockClient.processOpenLineageRunEventCalledWith.map((call) => + fromProtoStruct(call.openLineage), + ); + expect(events.map((e) => e.eventType)).to.deep.equal(["START", "COMPLETE"]); + for (const event of events) { + expect(event.inputs).to.deep.equal([{ namespace: "bigquery", name: "proj.schema.table_a" }]); + expect(event.outputs).to.deep.equal([ + { namespace: "bigquery", name: "proj.assertions.assert_a" }, + ]); + expect(event.job.facets.gcp_bq_pipelines_job.actionType).to.equal("ASSERTION"); + } + }); + + test("emits FAIL with errorMessage facet for a failing assertion", async () => { + const mockClient = new MockLineageClient(); + const emitter = new LineageEmitter( + credentials, + { lineageEnabled: true }, + () => mockClient as any, + ); + + const assertion = dataform.ExecutionAction.create({ + target: { database: "proj", schema: "assertions", name: "assert_a" }, + type: "assertion", + }); + + emitter.emitForAction( + assertion, + dataform.ActionResult.create({ + status: dataform.ActionResult.ExecutionStatus.FAILED, + tasks: [{ errorMessage: "bigquery error: Assertion failed: query returned 3 row(s)." }], + }), + ); + await emitter.drain(); + + expect(mockClient.processOpenLineageRunEventCalledWith.length).to.equal(1); + const event = fromProtoStruct(mockClient.processOpenLineageRunEventCalledWith[0].openLineage); + expect(event.eventType).to.equal("FAIL"); + expect(event.run.facets.errorMessage.message).to.equal( + "bigquery error: Assertion failed: query returned 3 row(s).", + ); + }); + + test("emits for property graph actions with PROPERTY_GRAPH actionType", async () => { + const mockClient = new MockLineageClient(); + const emitter = new LineageEmitter( + credentials, + { lineageEnabled: true }, + () => mockClient as any, + ); + + emitter.emitForAction( + dataform.ExecutionAction.create({ + target: { database: "proj", schema: "schema", name: "graph" }, + type: "propertyGraph", + }), + dataform.ActionResult.create({ status: dataform.ActionResult.ExecutionStatus.RUNNING }), + ); + await emitter.drain(); + + expect(mockClient.processOpenLineageRunEventCalledWith.length).to.equal(1); + const event = fromProtoStruct(mockClient.processOpenLineageRunEventCalledWith[0].openLineage); + expect(event.job.facets.gcp_bq_pipelines_job.actionType).to.equal("PROPERTY_GRAPH"); + }); + + test("declarations are skipped silently (no stderr line, no emission)", async () => { + const mockClient = new MockLineageClient(); + const stderr = new StderrCapture(); + const emitter = new LineageEmitter( + credentials, + { lineageEnabled: true }, + () => mockClient as any, + stderr, + ); + const declaration = dataform.ExecutionAction.create({ target: { database: "proj", schema: "schema", name: "declaration" }, type: "declaration", @@ -678,7 +776,6 @@ suite("LineageEmitter", () => { status: dataform.ActionResult.ExecutionStatus.RUNNING, }); - emitter.emitForAction(assertion, startResult); emitter.emitForAction(declaration, startResult); await emitter.drain(); diff --git a/cli/api/lineage/lineage_integration_test.ts b/cli/api/lineage/lineage_integration_test.ts index 6f5033542..63c19bec0 100644 --- a/cli/api/lineage/lineage_integration_test.ts +++ b/cli/api/lineage/lineage_integration_test.ts @@ -175,6 +175,63 @@ suite("--emit-lineage integration", { parallel: true }, () => { } }); + test("Case 1b — every BigQuery-job action type emits; declarations do not", async () => { + const recorder = new RecordingLineageClient(); + const emitter = createLineageEmitter( + factoryInput({ cliEmitLineage: true }), + new StderrCapture(), + recorderProvider(recorder), + ); + expect(emitter).to.not.equal(undefined); + + const graph = makeLinearGraph(); + graph.actions.push( + dataform.ExecutionAction.create({ + target: { database: "test-project", schema: "s_assertions", name: "assert_a" }, + type: "assertion", + tasks: [ + { type: "statement", statement: "CREATE OR REPLACE VIEW s_assertions.assert_a AS ..." }, + { type: "assertion", statement: "SELECT SUM(1) AS row_count FROM s_assertions.assert_a" }, + ], + dependencyTargets: [{ database: "test-project", schema: "s", name: "table_a" }], + }), + dataform.ExecutionAction.create({ + target: { database: "test-project", schema: "s", name: "graph" }, + type: "propertyGraph", + tasks: [{ type: "statement", statement: "CREATE OR REPLACE PROPERTY GRAPH s.graph ..." }], + dependencyTargets: [{ database: "test-project", schema: "s", name: "table_a" }], + }), + dataform.ExecutionAction.create({ + target: { database: "bigquery-public-data", schema: "samples", name: "shakespeare" }, + type: "declaration", + tasks: [], + }), + ); + + const runner = run(new FakeDbAdapter(), graph, { lineageEmitter: emitter }); + await runner.result(); + await emitter!.drain(); + + const eventsByActionType = new Map(); + for (const call of recorder.calls) { + const ol = fromProtoStruct(call.request.openLineage); + const actionType = ol.job.facets.gcp_bq_pipelines_job.actionType as string; + if (!eventsByActionType.has(actionType)) { + eventsByActionType.set(actionType, []); + } + eventsByActionType.get(actionType).push(ol.eventType as string); + } + // 2 tables + 1 assertion + 1 property graph, each START + COMPLETE. + expect(recorder.calls.length).to.equal(8); + expect([...eventsByActionType.keys()].sort()).to.deep.equal([ + "ASSERTION", + "PROPERTY_GRAPH", + "TABLE", + ]); + expect(eventsByActionType.get("ASSERTION").sort()).to.deep.equal(["COMPLETE", "START"]); + expect(eventsByActionType.get("PROPERTY_GRAPH").sort()).to.deep.equal(["COMPLETE", "START"]); + }); + test("Case 2 — workflow lineage.enabled=true (no CLI flag) enables emission", async () => { const recorder = new RecordingLineageClient(); const emitter = createLineageEmitter( diff --git a/cli/api/lineage/payload_builder.ts b/cli/api/lineage/payload_builder.ts index 45ae6e3bc..6ddf094b9 100644 --- a/cli/api/lineage/payload_builder.ts +++ b/cli/api/lineage/payload_builder.ts @@ -2,6 +2,7 @@ import { createHash, randomUUID } from "crypto"; import Long from "long"; import * as path from "path"; +import { toLineageActionType } from "df/cli/api/lineage/action_type"; import { version } from "df/core/version"; import { dataform } from "df/protos/ts"; @@ -125,7 +126,7 @@ export class LineagePayloadBuilder { }; if (eventType !== "START") { - const bqJobId = extractBqJobId(actionResult); + const bqJobId = extractTargetWritingBqJobId(action, actionResult); if (bqJobId && credentialsProjectId) { runFacets.externalQuery = { _producer: PRODUCER_URL, @@ -183,7 +184,7 @@ export class LineagePayloadBuilder { jobFacets.gcp_bq_pipelines_job = { dataformCoreVersion: version, - actionType: action.type, + actionType: toLineageActionType(action), actionName: canonicalActionTarget, }; @@ -231,11 +232,24 @@ function toMillis(val: Long | number | undefined | null): number | undefined { return undefined; } -function extractBqJobId(actionResult: dataform.IActionResult): string | undefined { +/** + * Returns the ID of the BigQuery job that wrote the action's target: the last + * task with a job ID, skipping assertion check tasks. Assertions run a + * `CREATE VIEW` task followed by a check query that only reads the view, so + * the check job carries no lineage for the assertion's dependencies. + * `actionResult.tasks` is index-aligned with `action.tasks`. + */ +function extractTargetWritingBqJobId( + action: dataform.IExecutionAction, + actionResult: dataform.IActionResult, +): string | undefined { if (!actionResult.tasks) { return undefined; } for (let i = actionResult.tasks.length - 1; i >= 0; i--) { + if (action.tasks?.[i]?.type === "assertion") { + continue; + } const jobId = actionResult.tasks[i]?.metadata?.bigquery?.jobId; if (jobId) { return jobId; diff --git a/cli/api/lineage/payload_builder_test.ts b/cli/api/lineage/payload_builder_test.ts index 888e80484..cd8e4cee1 100644 --- a/cli/api/lineage/payload_builder_test.ts +++ b/cli/api/lineage/payload_builder_test.ts @@ -106,6 +106,80 @@ suite("LineagePayloadBuilder", () => { expect(complete.run.facets.externalQuery).to.equal(undefined); }); + test("externalQuery uses the last job for multi-task tables", () => { + const multiTaskTable = dataform.ExecutionAction.create({ + target: { database: "proj", schema: "schema", name: "table" }, + type: "table", + tasks: [ + { type: "statement", statement: "CREATE SCHEMA IF NOT EXISTS ..." }, + { type: "statement", statement: "CREATE OR REPLACE TABLE ..." }, + ], + }); + const builder = new LineagePayloadBuilder("/tmp/proj"); + const complete = builder.build( + multiTaskTable, + dataform.ActionResult.create({ + status: dataform.ActionResult.ExecutionStatus.SUCCESSFUL, + tasks: [ + { metadata: { bigquery: { jobId: "job-schema" } } }, + { metadata: { bigquery: { jobId: "job-ctas" } } }, + ], + }), + "proj", + "us", + "cred-proj", + ); + expect(complete.run.facets.externalQuery.externalQueryId).to.equal("cred-proj.us.job-ctas"); + }); + + test("externalQuery uses the CREATE VIEW job for assertions, not the check job", () => { + const assertion = dataform.ExecutionAction.create({ + target: { database: "proj", schema: "assertions", name: "assert_a" }, + type: "assertion", + tasks: [ + { + type: "statement", + statement: "create or replace view `proj.assertions.assert_a` as ...", + }, + { + type: "assertion", + statement: "select sum(1) as row_count from `proj.assertions.assert_a`", + }, + ], + }); + const builder = new LineagePayloadBuilder("/tmp/proj"); + const tasks = [ + { metadata: { bigquery: { jobId: "job-create-view" } } }, + { metadata: { bigquery: { jobId: "job-check" } } }, + ]; + const complete = builder.build( + assertion, + dataform.ActionResult.create({ + status: dataform.ActionResult.ExecutionStatus.SUCCESSFUL, + tasks, + }), + "proj", + "us", + "cred-proj", + ); + const failed = builder.build( + assertion, + dataform.ActionResult.create({ + status: dataform.ActionResult.ExecutionStatus.FAILED, + tasks, + }), + "proj", + "us", + "cred-proj", + ); + expect(complete.run.facets.externalQuery.externalQueryId).to.equal( + "cred-proj.us.job-create-view", + ); + expect(failed.run.facets.externalQuery.externalQueryId).to.equal( + "cred-proj.us.job-create-view", + ); + }); + test("errorMessage run facet appears only on FAIL with non-empty task errorMessage", () => { const builder = new LineagePayloadBuilder("/tmp/proj"); const failedWithMsg = builder.build( @@ -206,7 +280,7 @@ suite("LineagePayloadBuilder", () => { const builder = new LineagePayloadBuilder("/tmp/proj"); const payload = builder.build(ACTION, START_RESULT, "proj", "us"); expect(payload.job.facets.gcp_bq_pipelines_job).to.deep.include({ - actionType: "table", + actionType: "TABLE", actionName: "schema.table", }); expect(payload.job.facets.gcp_bq_pipelines_job.dataformCoreVersion).to.be.a("string");