Skip to content

fix(cli): remove unused sql and errorMessage OpenLineage facets #2391

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rafal-hawrylak merged 1 commit into from
Oct 20, 2026
Merged

fix(cli): remove unused sql and errorMessage OpenLineage facets #2391

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 2 additions & 15 deletions cli/api/lineage/emitter_test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -147,11 +147,6 @@ suite("LineageEmitter", () => {
new Date(2000).toISOString(),
);

// SQL job facet verified
expect(openLineage.job.facets.sql.query).to.equal(
"CREATE TABLE target_table AS SELECT * FROM source_table",
);

// GCP lineage job facet verified
expect(openLineage.job.facets.gcp_lineage.displayName).to.equal(
"BigQuery Pipelines action target_dataset.target_table",
Expand Down Expand Up @@ -225,12 +220,6 @@ suite("LineageEmitter", () => {
"target-project.us.cli.my-dataform-project-0b5d3e86.target_dataset.failing_table",
);

// Error message run facet verified
expect(openLineage.run.facets.errorMessage.message).to.equal(
'bigquery error: Syntax error: Unexpected "\\" at [3:15]',
);
expect(openLineage.run.facets.errorMessage.programmingLanguage).to.equal("typescript");

// Nominal time run facet verified (nominalEndTime matches failure timing)
expect(openLineage.run.facets.nominalTime.nominalStartTime).to.equal(
new Date(1000).toISOString(),
Expand Down Expand Up @@ -721,7 +710,7 @@ suite("LineageEmitter", () => {
}
});

test("emits FAIL with errorMessage facet for a failing assertion", async () => {
test("emits FAIL event for a failing assertion", async () => {
const mockClient = new MockLineageClient();
const emitter = new LineageEmitter(
credentials,
Expand All @@ -746,9 +735,7 @@ suite("LineageEmitter", () => {
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).",
);
expect(event.job.facets.gcp_bq_pipelines_job.actionType).to.equal("ASSERTION");
});

test("emits for property graph actions with PROPERTY_GRAPH actionType", async () => {
Expand Down
72 changes: 21 additions & 51 deletions cli/api/lineage/payload_builder.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,9 +14,6 @@ const PARENT_RUN_FACET_SCHEMA =
"https://openlineage.io/spec/facets/1-2-0/ParentRunFacet.json#/$defs/ParentRunFacet";
const EXTERNAL_QUERY_FACET_SCHEMA =
"https://openlineage.io/spec/facets/1-0-2/ExternalQueryRunFacet.json";
const ERROR_MESSAGE_FACET_SCHEMA =
"https://openlineage.io/spec/facets/1-0-1/ErrorMessageRunFacet.json";
const SQL_JOB_FACET_SCHEMA = "https://openlineage.io/spec/facets/1-1-0/SQLJobFacet.json";
const GCP_LINEAGE_JOB_FACET_SCHEMA =
"https://openlineage.io/spec/facets/1-0-0/GcpLineageJobFacet.json#/$defs/GcpLineageJobFacet";
const JOB_TYPE_FACET_SCHEMA =
Expand Down Expand Up @@ -142,55 +139,28 @@ export class LineagePayloadBuilder {
}
}

if (eventType === "FAIL") {
const errorMessages = actionResult.tasks
?.map((t) => t.errorMessage)
.filter((msg) => !!msg)
.join("; ");
if (errorMessages) {
runFacets.errorMessage = {
_schemaURL: ERROR_MESSAGE_FACET_SCHEMA,
message: errorMessages,
programmingLanguage: "typescript",
};
}
}

const sqlStatements = action.tasks
?.map((task) => task.statement)
.filter((stmt) => !!stmt)
.join(";\n");

const jobFacets: any = {};
if (sqlStatements) {
jobFacets.sql = {
_schemaURL: SQL_JOB_FACET_SCHEMA,
query: sqlStatements,
};
}

jobFacets.gcp_lineage = {
_producer: PRODUCER_URL,
_schemaURL: GCP_LINEAGE_JOB_FACET_SCHEMA,
displayName: `BigQuery Pipelines action ${canonicalActionTarget}`,
origin: {
name: `projects/${projectId}/locations/${location}/cli/${workdirIdentifier}`,
sourceType: "BIGQUERY_PIPELINES",
const jobFacets: any = {
gcp_lineage: {
_producer: PRODUCER_URL,
_schemaURL: GCP_LINEAGE_JOB_FACET_SCHEMA,
displayName: `BigQuery Pipelines action ${canonicalActionTarget}`,
origin: {
name: `projects/${projectId}/locations/${location}/cli/${workdirIdentifier}`,
sourceType: "BIGQUERY_PIPELINES",
},
},
jobType: {
_producer: PRODUCER_URL,
_schemaURL: JOB_TYPE_FACET_SCHEMA,
integration: "BIGQUERY_PIPELINES",
jobType: "ACTION",
processingType: "BATCH",
},
gcp_bq_pipelines_job: {
dataformCoreVersion: version,
actionType,
actionName: canonicalActionTarget,
},
};

jobFacets.jobType = {
_producer: PRODUCER_URL,
_schemaURL: JOB_TYPE_FACET_SCHEMA,
integration: "BIGQUERY_PIPELINES",
jobType: "ACTION",
processingType: "BATCH",
};

jobFacets.gcp_bq_pipelines_job = {
dataformCoreVersion: version,
actionType,
actionName: canonicalActionTarget,
};

return {
Expand Down
77 changes: 3 additions & 74 deletions cli/api/lineage/payload_builder_test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -180,92 +180,21 @@ suite("LineagePayloadBuilder", () => {
);
});

test("errorMessage run facet appears only on FAIL with non-empty task errorMessage", () => {
const builder = new LineagePayloadBuilder("/tmp/proj");
const failedWithMsg = builder.build(
ACTION,
dataform.ActionResult.create({
status: dataform.ActionResult.ExecutionStatus.FAILED,
tasks: [{ errorMessage: "boom" }, { errorMessage: "kaboom" }],
}),
"proj",
"us",
);
expect(failedWithMsg.run.facets.errorMessage).to.deep.include({
message: "boom; kaboom",
programmingLanguage: "typescript",
});
const failedWithoutMsg = builder.build(
ACTION,
dataform.ActionResult.create({
status: dataform.ActionResult.ExecutionStatus.FAILED,
}),
"proj",
"us",
);
expect(failedWithoutMsg.run.facets.errorMessage).to.equal(undefined);
});

test("errorMessage aggregates non-empty task messages across multi-task action", () => {
test("does not emit sql job facet or errorMessage run facet", () => {
const builder = new LineagePayloadBuilder("/tmp/proj");
const failed = builder.build(
ACTION,
dataform.ActionResult.create({
status: dataform.ActionResult.ExecutionStatus.FAILED,
tasks: [
{ errorMessage: "pre_operations failed: table not found" },
{ errorMessage: "" },
{ errorMessage: "main statement failed: syntax error at line 3" },
],
}),
"proj",
"us",
);
expect(failed.run.facets.errorMessage).to.deep.equal({
_schemaURL: "https://openlineage.io/spec/facets/1-0-1/ErrorMessageRunFacet.json",
message:
"pre_operations failed: table not found; main statement failed: syntax error at line 3",
programmingLanguage: "typescript",
});
});

test("errorMessage facet omitted when every task has empty or undefined errorMessage", () => {
const builder = new LineagePayloadBuilder("/tmp/proj");
const failed = builder.build(
ACTION,
dataform.ActionResult.create({
status: dataform.ActionResult.ExecutionStatus.FAILED,
tasks: [{ errorMessage: "" }, {}, { errorMessage: "" }],
tasks: [{ errorMessage: "boom" }],
}),
"proj",
"us",
);
expect(failed.job.facets.sql).to.equal(undefined);
expect(failed.run.facets.errorMessage).to.equal(undefined);
});

test("sql job facet emits joined task statements", () => {
const multiTaskAction = dataform.ExecutionAction.create({
target: { database: "proj", schema: "schema", name: "table" },
type: "table",
tasks: [{ statement: "CREATE TABLE t AS SELECT 1" }, { statement: "SELECT 2" }],
});
const builder = new LineagePayloadBuilder("/tmp/proj");
const payload = builder.build(multiTaskAction, START_RESULT, "proj", "us");
expect(payload.job.facets.sql).to.deep.include({
query: "CREATE TABLE t AS SELECT 1;\nSELECT 2",
});
});

test("sql job facet omitted when action has no statements", () => {
const emptyAction = dataform.ExecutionAction.create({
target: { database: "proj", schema: "schema", name: "table" },
type: "table",
});
const builder = new LineagePayloadBuilder("/tmp/proj");
const payload = builder.build(emptyAction, START_RESULT, "proj", "us");
expect(payload.job.facets.sql).to.equal(undefined);
});

test("jobType job facet declares BigQuery Pipelines integration", () => {
const builder = new LineagePayloadBuilder("/tmp/proj");
const payload = builder.build(ACTION, START_RESULT, "proj", "us");
Expand Down
Loading