feat(cli): emit lineage for assertions and property graphs

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.

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;
+}

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);
+    }
+  });
+});

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;
     }
 

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();
 

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<string, string[]>();
+    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(

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;