From 6bef0e102644a8a0eb73495efd963d7668b0958e Mon Sep 17 00:00:00 2001 From: Bart Maertens Date: Tue, 20 Oct 2026 14:28:34 +0200 Subject: [PATCH] Expose the pipeline hop graph on lineage events, fixes #9004 - add the PIPELINE_GRAPH event kind with PipelineGraphLineagePayload (enabled hops, in pipeline order) - emit it once per pipeline run, next to STARTED, with the same context - document that sinks skip event kinds they do not handle - the OpenLineage sink ignores the new kind, so its output does not change yet --- .../modules/ROOT/pages/lineage.adoc | 5 +- .../pages/technology/openlineage/index.adoc | 1 + .../lineage/LineageRunLifecycleEmitter.java | 53 ++++- .../hop/lineage/model/LineageEventKind.java | 15 +- .../model/PipelineGraphLineagePayload.java | 101 +++++++++ .../apache/hop/lineage/spi/ILineageSink.java | 4 + .../hop/lineage/LineagePipelineGraphTest.java | 199 ++++++++++++++++++ .../openlineage/OpenLineageEventMapper.java | 1 + .../OpenLineageEventMapperTest.java | 19 ++ 9 files changed, 393 insertions(+), 5 deletions(-) create mode 100644 engine/src/main/java/org/apache/hop/lineage/model/PipelineGraphLineagePayload.java create mode 100644 engine/src/test/java/org/apache/hop/lineage/LineagePipelineGraphTest.java diff --git a/docs/hop-dev-manual/modules/ROOT/pages/lineage.adoc b/docs/hop-dev-manual/modules/ROOT/pages/lineage.adoc index f5fdf37a0c0..4f13aac7499 100644 --- a/docs/hop-dev-manual/modules/ROOT/pages/lineage.adoc +++ b/docs/hop-dev-manual/modules/ROOT/pages/lineage.adoc @@ -105,7 +105,7 @@ Each observation is a **`LineageEvent`**: * **Event id** — Unique id (UUID by default). * **Timestamp** — Milliseconds since epoch. -* **`LineageEventKind`** — One of `RUN_LIFECYCLE`, `TRANSFORM_SCHEMA`, `FILE_IO`, `RELATIONAL_IO`, `HTTP_IO`. +* **`LineageEventKind`** — One of `RUN_LIFECYCLE`, `TRANSFORM_SCHEMA`, `FILE_IO`, `RELATIONAL_IO`, `HTTP_IO`, `PIPELINE_GRAPH`. New kinds are only ever appended: a sink must skip a kind it does not handle rather than fail the batch. * **`LineageContext`** — Who/what produced the observation (see above). * **`LineagePayload`** — Optional typed payload; may be null if the kind and context are enough. @@ -115,6 +115,7 @@ Payload types are plain Java types (strings, longs, enums) so sinks can serializ * **`RelationalIoLineagePayload`** — Relational/JDBC table access: operation (`READ`, `WRITE`, `DELETE`, `EXEC`), the resolved **dataset namespace**, the input/output **`RelationalTable`** list (catalog/schema/table), the raw SQL when the engine cannot resolve the tables itself, an optional column schema, the connection's default catalog and schema, optional per-column provenance (**`RelationalWriteColumn`**: target column, stream field, origin transform), and an optional **`RelationalLifecycle`** (`OVERWRITE`, `TRUNCATE`, `DROP`, `CREATE`). * **`HttpLineagePayload`** — `CLIENT` vs `SERVER`, method, URL, status, request/response bytes, duration, success, message. * **`RunLifecycleLineagePayload`** — `STARTED` / `FINISHED` plus optional detail text. +* **`PipelineGraphLineagePayload`** — The pipeline's hop graph as a list of `from` → `to` transform-name transitions: enabled hops with both ends named, in pipeline order. Informational hops are included. A sink uses it to attribute datasets to the transform that touched them and to propagate them along the graph, instead of pairing all of a pipeline's inputs with all of its outputs. * **`TransformSchemaLineagePayload`** — `INPUT` or `OUTPUT` plus a list of field names, Hop type names, length, and precision (boundary schema; emitted when a transform finishes). Fields usually come from runtime row metadata on the transform or row sets. If input was never observed (e.g. _Abort_ when `getRow()` immediately returns null), input is derived via `PipelineMeta#getPrevTransformFields`. If output was never written (e.g. _Detect empty stream_ with a non-empty stream), output is derived via `PipelineMeta#getTransformFields`. Graph-derived events set context attribute `transformSchemaSource` to `DESIGN_GRAPH`. Convenience factory: `LineageEvent.of(kind, context, payload)` assigns id and current time. @@ -143,7 +144,7 @@ The queue is allocated lazily on the first `emit` from the resolved configuratio The `org.apache.hop.lineage.xp` package registers extension points that **emit** `RUN_LIFECYCLE` events (via `LineageRunLifecycleEmitter`) and **flush** the hub where needed: -* **`PipelineStart`** — `LineageHubPipelineStartXp` — phase `STARTED` +* **`PipelineStart`** — `LineageHubPipelineStartXp` — phase `STARTED`, followed by one `PIPELINE_GRAPH` event with the same context when the pipeline has enabled hops. `PipelineStart` fires after the transform threads have started, so the run's transform `STARTED` events, and early I/O such as a Table Input read, can arrive before the graph, possibly in an earlier batch. A sink that uses the graph collects a run's events by the pipeline's log channel id (transform events carry it in the `pipelineLogChannelId` attribute) and resolves them at `FINISHED` or `FAILED`. * **`PipelineFinish`** — `LineageHubPipelineFinishXp` — phase `FINISHING` (transforms stopped, cleanup) * **`PipelineCompleted`** — `LineageHubPipelineCompletedXp` — phase `FINISHED` or `FAILED` (from pipeline error count), then `flushQuietly()` * **`WorkflowStart`** — `LineageHubWorkflowStartXp` — phase `STARTED` diff --git a/docs/hop-user-manual/modules/ROOT/pages/technology/openlineage/index.adoc b/docs/hop-user-manual/modules/ROOT/pages/technology/openlineage/index.adoc index ba55541eb9b..bf226b7874f 100644 --- a/docs/hop-user-manual/modules/ROOT/pages/technology/openlineage/index.adoc +++ b/docs/hop-user-manual/modules/ROOT/pages/technology/openlineage/index.adoc @@ -87,6 +87,7 @@ properties if needed. |`RELATIONAL_IO` |`RunEvent` (`OTHER`) with relational table datasets — see <>. Carries a `schema` facet, a `columnLineage` facet when column mappings are derivable, and a `lifecycleStateChange` facet for truncate/overwrite writes. |`TRANSFORM_SCHEMA` |Skipped (internal row shapes between transforms, not physical datasets). |`HTTP_IO` |Skipped. +|`PIPELINE_GRAPH` |Skipped. Datasets stay on the pipeline job. |=== The run id is the executing pipeline/workflow log-channel id; file I/O is correlated to the parent diff --git a/engine/src/main/java/org/apache/hop/lineage/LineageRunLifecycleEmitter.java b/engine/src/main/java/org/apache/hop/lineage/LineageRunLifecycleEmitter.java index 40a700aecdd..53d16697938 100644 --- a/engine/src/main/java/org/apache/hop/lineage/LineageRunLifecycleEmitter.java +++ b/engine/src/main/java/org/apache/hop/lineage/LineageRunLifecycleEmitter.java @@ -17,6 +17,8 @@ package org.apache.hop.lineage; +import java.util.ArrayList; +import java.util.List; import org.apache.hop.core.Const; import org.apache.hop.core.Result; import org.apache.hop.core.logging.ILogChannel; @@ -28,8 +30,11 @@ import org.apache.hop.lineage.hub.LineageHub; import org.apache.hop.lineage.model.LineageEvent; import org.apache.hop.lineage.model.LineageEventKind; +import org.apache.hop.lineage.model.PipelineGraphLineagePayload; +import org.apache.hop.lineage.model.PipelineGraphLineagePayload.Transition; import org.apache.hop.lineage.model.RunLifecycleLineagePayload; import org.apache.hop.lineage.model.RunLifecyclePhase; +import org.apache.hop.pipeline.PipelineHopMeta; import org.apache.hop.pipeline.PipelineMeta; import org.apache.hop.pipeline.engine.IPipelineEngine; import org.apache.hop.pipeline.transform.ITransform; @@ -167,12 +172,58 @@ public static void emitPipeline( if (phase == RunLifecyclePhase.FINISHED || phase == RunLifecyclePhase.FAILED) { ctx.putAttribute(OUTCOME, phase == RunLifecyclePhase.FAILED ? FAILED : SUCCESS); } + LineageContext context = ctx.build(); LineageHub.getInstance() .emit( LineageEvent.of( LineageEventKind.RUN_LIFECYCLE, - ctx.build(), + context, new RunLifecycleLineagePayload(phase, detail))); + + // The hop graph is a fact about the pipeline, not about a phase: emit it once, next to the + // STARTED event. Sinks that attribute I/O to individual transforms need it to know which + // transform fed which; see PipelineGraphLineagePayload. + // + if (phase == RunLifecyclePhase.STARTED && meta != null) { + List transitions = pipelineTransitions(meta); + if (!transitions.isEmpty()) { + LineageHub.getInstance() + .emit( + LineageEvent.of( + LineageEventKind.PIPELINE_GRAPH, + context, + new PipelineGraphLineagePayload(transitions))); + } + } + } + + /** + * The enabled hops of a pipeline, in pipeline order, as {@code from -> to} transitions. + * + *

Hops with an unknown end are skipped (a half-edited hop can exist in memory), as are hops + * without names. Informational hops stay in: data flows through them, even though they do no I/O + * of their own, so dropping them would break a chain for anything that propagates datasets along + * it. + */ + public static List pipelineTransitions(PipelineMeta meta) { + if (meta == null) { + return List.of(); + } + List transitions = new ArrayList<>(); + for (PipelineHopMeta hop : meta.getHops()) { + if (hop == null + || !hop.isEnabled() + || hop.getFromTransform() == null + || hop.getToTransform() == null) { + continue; + } + String from = hop.getFromTransform().getName(); + String to = hop.getToTransform().getName(); + if (!Utils.isEmpty(from) && !Utils.isEmpty(to)) { + transitions.add(new Transition(from, to)); + } + } + return transitions; } public static void emitWorkflow( diff --git a/engine/src/main/java/org/apache/hop/lineage/model/LineageEventKind.java b/engine/src/main/java/org/apache/hop/lineage/model/LineageEventKind.java index fc22ad7d04b..a81905eadee 100644 --- a/engine/src/main/java/org/apache/hop/lineage/model/LineageEventKind.java +++ b/engine/src/main/java/org/apache/hop/lineage/model/LineageEventKind.java @@ -17,7 +17,12 @@ package org.apache.hop.lineage.model; -/** High-level category of a lineage observation. */ +/** + * High-level category of a lineage observation. + * + *

Kinds are added over time and only ever appended. A sink must ignore a kind it does not + * handle, for example with a {@code default} branch, so a newer engine does not break it. + */ public enum LineageEventKind { RUN_LIFECYCLE, /** Transform input/output field names and Hop types (boundary schema). */ @@ -26,5 +31,11 @@ public enum LineageEventKind { /** HTTP client traffic from transforms/actions (payload: {@link HttpLineagePayload}). */ HTTP_IO, /** Relational/JDBC table read or write (payload: {@link RelationalIoLineagePayload}). */ - RELATIONAL_IO + RELATIONAL_IO, + /** + * The pipeline's hop graph, emitted once per run next to its start (payload: {@link + * PipelineGraphLineagePayload}). Sinks need it to attribute a dataset to the transform that + * touched it rather than to the whole pipeline. + */ + PIPELINE_GRAPH } diff --git a/engine/src/main/java/org/apache/hop/lineage/model/PipelineGraphLineagePayload.java b/engine/src/main/java/org/apache/hop/lineage/model/PipelineGraphLineagePayload.java new file mode 100644 index 00000000000..13a837fb75c --- /dev/null +++ b/engine/src/main/java/org/apache/hop/lineage/model/PipelineGraphLineagePayload.java @@ -0,0 +1,101 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hop.lineage.model; + +import java.util.List; +import java.util.Objects; +import lombok.Getter; + +/** + * The pipeline's hop graph: which transform feeds which, as observed when the run started. + * + *

A sink needs this to attribute lineage to the transform that actually touched a dataset. + * Dataset lineage is derived through jobs — a job's inputs are paired with its outputs — so a sink + * that files a whole pipeline's I/O under one job can only express the full input × output product, + * not the true pairs. With the graph a sink can give each transform's job the datasets it consumes + * (its own reads plus what its upstream transforms produced) and its own outputs. + * + *

Only enabled hops with two known ends are listed, in pipeline order. Informational hops are + * included: data flows through them, even though they do no I/O of their own. + * + *

The graph is emitted with the pipeline's {@code STARTED} event, which fires after the + * transform threads have started. A transform's own events, including early I/O, can therefore + * arrive before the graph, even in an earlier batch. Collect a run's events by the pipeline's log + * channel id (transform events carry it in the {@code pipelineLogChannelId} attribute) and resolve + * them when the run finishes, rather than assuming the graph comes first. + */ +@Getter +public final class PipelineGraphLineagePayload implements LineagePayload { + + /** One enabled hop: the transform named {@code from} feeds the transform named {@code to}. */ + @Getter + public static final class Transition { + private final String from; + private final String to; + + public Transition(String from, String to) { + this.from = Objects.requireNonNull(from, "from"); + this.to = Objects.requireNonNull(to, "to"); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + Transition that = (Transition) o; + return from.equals(that.from) && to.equals(that.to); + } + + @Override + public int hashCode() { + return Objects.hash(from, to); + } + + @Override + public String toString() { + return from + ">" + to; + } + } + + private final List transitions; + + public PipelineGraphLineagePayload(List transitions) { + this.transitions = List.copyOf(Objects.requireNonNull(transitions, "transitions")); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + PipelineGraphLineagePayload that = (PipelineGraphLineagePayload) o; + return transitions.equals(that.transitions); + } + + @Override + public int hashCode() { + return transitions.hashCode(); + } +} diff --git a/engine/src/main/java/org/apache/hop/lineage/spi/ILineageSink.java b/engine/src/main/java/org/apache/hop/lineage/spi/ILineageSink.java index d5bcbf05ec4..40ef2f4691e 100644 --- a/engine/src/main/java/org/apache/hop/lineage/spi/ILineageSink.java +++ b/engine/src/main/java/org/apache/hop/lineage/spi/ILineageSink.java @@ -43,6 +43,10 @@ default void init(IVariables variables, ILogChannel log) throws HopException { /** * Deliver a non-empty batch of events. Implementations should not retain references to the list * or events after returning. + * + *

Skip events of a {@link org.apache.hop.lineage.model.LineageEventKind} the sink does not + * handle rather than failing: new kinds are added over time, and an exception here loses the + * whole batch for this sink. */ void accept(List events) throws HopException; diff --git a/engine/src/test/java/org/apache/hop/lineage/LineagePipelineGraphTest.java b/engine/src/test/java/org/apache/hop/lineage/LineagePipelineGraphTest.java new file mode 100644 index 00000000000..a197d70d9ca --- /dev/null +++ b/engine/src/test/java/org/apache/hop/lineage/LineagePipelineGraphTest.java @@ -0,0 +1,199 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hop.lineage; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.atLeastOnce; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.List; +import org.apache.hop.lineage.hub.LineageHub; +import org.apache.hop.lineage.model.LineageEvent; +import org.apache.hop.lineage.model.LineageEventKind; +import org.apache.hop.lineage.model.PipelineGraphLineagePayload; +import org.apache.hop.lineage.model.PipelineGraphLineagePayload.Transition; +import org.apache.hop.lineage.model.RunLifecyclePhase; +import org.apache.hop.pipeline.PipelineHopMeta; +import org.apache.hop.pipeline.PipelineMeta; +import org.apache.hop.pipeline.engine.IPipelineEngine; +import org.apache.hop.pipeline.transform.TransformMeta; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.MockedStatic; +import org.mockito.Mockito; + +/** + * The graph a sink receives must describe the pipeline that ran: enabled hops only, in pipeline + * order, and never a transition with a missing end. + */ +class LineagePipelineGraphTest { + + private static TransformMeta transform(String name) { + TransformMeta transformMeta = new TransformMeta(); + transformMeta.setName(name); + return transformMeta; + } + + @Test + void chainIsReportedInPipelineOrder() { + PipelineMeta meta = new PipelineMeta(); + TransformMeta read = transform("Read orders"); + TransformMeta write = transform("Write copy"); + TransformMeta pause = transform("Pause"); + meta.getTransforms().add(read); + meta.getTransforms().add(write); + meta.getTransforms().add(pause); + meta.getHops().add(new PipelineHopMeta(read, write)); + meta.getHops().add(new PipelineHopMeta(write, pause)); + + assertEquals( + List.of(new Transition("Read orders", "Write copy"), new Transition("Write copy", "Pause")), + LineageRunLifecycleEmitter.pipelineTransitions(meta)); + } + + @Test + void disabledHopIsNotPartOfTheGraphTheRunExecuted() { + PipelineMeta meta = new PipelineMeta(); + TransformMeta a = transform("A"); + TransformMeta b = transform("B"); + meta.getHops().add(new PipelineHopMeta(a, b, false)); + + assertTrue(LineageRunLifecycleEmitter.pipelineTransitions(meta).isEmpty()); + } + + @Test + void hopWithoutAnEndIsSkipped() { + PipelineMeta meta = new PipelineMeta(); + TransformMeta a = transform("A"); + PipelineHopMeta noEnds = new PipelineHopMeta(); + noEnds.setEnabled(true); + PipelineHopMeta noTo = new PipelineHopMeta(a, null); + noTo.setEnabled(true); + PipelineHopMeta noFrom = new PipelineHopMeta(null, a); + noFrom.setEnabled(true); + meta.getHops().add(noEnds); + meta.getHops().add(noTo); + meta.getHops().add(noFrom); + + assertTrue(LineageRunLifecycleEmitter.pipelineTransitions(meta).isEmpty()); + } + + @Test + void hopWithAnUnnamedEndIsSkipped() { + PipelineMeta meta = new PipelineMeta(); + TransformMeta named = transform("A"); + TransformMeta unnamed = new TransformMeta(); + meta.getHops().add(new PipelineHopMeta(named, unnamed)); + + assertTrue(LineageRunLifecycleEmitter.pipelineTransitions(meta).isEmpty()); + } + + @Test + void diamondKeepsBothBranches() { + PipelineMeta meta = new PipelineMeta(); + TransformMeta read = transform("Read"); + TransformMeta upper = transform("Upper"); + TransformMeta lower = transform("Lower"); + TransformMeta join = transform("Join"); + meta.getHops().add(new PipelineHopMeta(read, upper)); + meta.getHops().add(new PipelineHopMeta(read, lower)); + meta.getHops().add(new PipelineHopMeta(upper, join)); + meta.getHops().add(new PipelineHopMeta(lower, join)); + + assertEquals( + List.of( + new Transition("Read", "Upper"), + new Transition("Read", "Lower"), + new Transition("Upper", "Join"), + new Transition("Lower", "Join")), + LineageRunLifecycleEmitter.pipelineTransitions(meta)); + } + + @Test + void noMetaMeansNoGraph() { + assertTrue(LineageRunLifecycleEmitter.pipelineTransitions(null).isEmpty()); + } + + @Test + void startEmitsTheGraphWithTheRunContext() { + List events = emitPipeline(chain(), RunLifecyclePhase.STARTED); + + assertEquals(2, events.size()); + assertEquals(LineageEventKind.RUN_LIFECYCLE, events.get(0).getKind()); + LineageEvent graph = events.get(1); + assertEquals(LineageEventKind.PIPELINE_GRAPH, graph.getKind()); + assertSame(events.get(0).getContext(), graph.getContext()); + assertEquals( + List.of(new Transition("Read", "Write")), + ((PipelineGraphLineagePayload) graph.getPayload()).getTransitions()); + } + + @Test + void otherPhasesDoNotRepeatTheGraph() { + for (RunLifecyclePhase phase : RunLifecyclePhase.values()) { + if (phase == RunLifecyclePhase.STARTED) { + continue; + } + List events = emitPipeline(chain(), phase); + assertEquals(1, events.size(), phase.name()); + assertEquals(LineageEventKind.RUN_LIFECYCLE, events.get(0).getKind(), phase.name()); + } + } + + @Test + void pipelineWithoutHopsEmitsNoGraph() { + PipelineMeta meta = new PipelineMeta(); + meta.getTransforms().add(transform("Alone")); + + List events = emitPipeline(meta, RunLifecyclePhase.STARTED); + + assertEquals(1, events.size()); + assertEquals(LineageEventKind.RUN_LIFECYCLE, events.get(0).getKind()); + } + + private static PipelineMeta chain() { + PipelineMeta meta = new PipelineMeta(); + TransformMeta read = transform("Read"); + TransformMeta write = transform("Write"); + meta.getTransforms().add(read); + meta.getTransforms().add(write); + meta.getHops().add(new PipelineHopMeta(read, write)); + return meta; + } + + @SuppressWarnings("unchecked") + private static List emitPipeline(PipelineMeta meta, RunLifecyclePhase phase) { + LineageHub hub = mock(LineageHub.class); + IPipelineEngine pipeline = mock(IPipelineEngine.class); + when(pipeline.getPipelineMeta()).thenReturn(meta); + when(pipeline.getLogChannelId()).thenReturn("pipeline-log"); + + try (MockedStatic staticHub = Mockito.mockStatic(LineageHub.class)) { + staticHub.when(LineageHub::getInstance).thenReturn(hub); + LineageRunLifecycleEmitter.emitPipeline(pipeline, phase, null); + } + + ArgumentCaptor captor = ArgumentCaptor.forClass(LineageEvent.class); + verify(hub, atLeastOnce()).emit(captor.capture()); + return captor.getAllValues(); + } +} diff --git a/plugins/tech/openlineage/src/main/java/org/apache/hop/lineage/openlineage/OpenLineageEventMapper.java b/plugins/tech/openlineage/src/main/java/org/apache/hop/lineage/openlineage/OpenLineageEventMapper.java index 6d030c6eefd..e8326cc085b 100644 --- a/plugins/tech/openlineage/src/main/java/org/apache/hop/lineage/openlineage/OpenLineageEventMapper.java +++ b/plugins/tech/openlineage/src/main/java/org/apache/hop/lineage/openlineage/OpenLineageEventMapper.java @@ -134,6 +134,7 @@ public Optional toRunEvent(LineageEvent event) { case RELATIONAL_IO -> mapRelationalIo(event); case TRANSFORM_SCHEMA -> Optional.empty(); case HTTP_IO -> Optional.empty(); + case PIPELINE_GRAPH -> Optional.empty(); }; } diff --git a/plugins/tech/openlineage/src/test/java/org/apache/hop/lineage/openlineage/OpenLineageEventMapperTest.java b/plugins/tech/openlineage/src/test/java/org/apache/hop/lineage/openlineage/OpenLineageEventMapperTest.java index b07952aaba9..37625435178 100644 --- a/plugins/tech/openlineage/src/test/java/org/apache/hop/lineage/openlineage/OpenLineageEventMapperTest.java +++ b/plugins/tech/openlineage/src/test/java/org/apache/hop/lineage/openlineage/OpenLineageEventMapperTest.java @@ -38,6 +38,8 @@ import org.apache.hop.lineage.model.LineageEvent; import org.apache.hop.lineage.model.LineageEventKind; import org.apache.hop.lineage.model.LineageFieldSchema; +import org.apache.hop.lineage.model.PipelineGraphLineagePayload; +import org.apache.hop.lineage.model.PipelineGraphLineagePayload.Transition; import org.apache.hop.lineage.model.RelationalIoLineagePayload; import org.apache.hop.lineage.model.RelationalIoOperation; import org.apache.hop.lineage.model.RelationalLifecycle; @@ -262,6 +264,23 @@ void transformSchemaIsSkippedInV1() { assertFalse(mapper.map(event).isPresent()); } + // The hop graph does not change the OpenLineage output yet: the pipeline job keeps its datasets. + @Test + void pipelineGraphIsSkipped() { + LineageEvent event = + new LineageEvent( + "evt-graph", + 1L, + LineageEventKind.PIPELINE_GRAPH, + LineageContext.builder() + .subjectType(LineageSubjectType.PIPELINE) + .logChannelId(PIPELINE_RUN_UUID) + .pipelineName("graph-pipeline") + .build(), + new PipelineGraphLineagePayload(List.of(new Transition("Read", "Write")))); + assertFalse(mapper.map(event).isPresent()); + } + @Test void fileIoReadMapsSchemaFacetFromContentSchema() throws Exception { FileIoContentSchema contentSchema =