This is an automated email from the ASF dual-hosted git repository.
mattcasters pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 0c7da4dce7 Expose the pipeline hop graph on lineage events, fixes
#9004 (#9042)
0c7da4dce7 is described below
commit 0c7da4dce709b1d640688b431cf3ddfb8e12a99c
Author: Bart Maertens <[email protected]>
AuthorDate: Wed Oct 21 09:49:08 2026 +0200
Expose the pipeline hop graph on lineage events, fixes #9004 (#9042)
- 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
---
.../hop-dev-manual/modules/ROOT/pages/lineage.adoc | 5 +-
.../ROOT/pages/technology/openlineage/index.adoc | 1 +
.../hop/lineage/LineageRunLifecycleEmitter.java | 53 +++++-
.../apache/hop/lineage/model/LineageEventKind.java | 15 +-
.../lineage/model/PipelineGraphLineagePayload.java | 101 +++++++++++
.../org/apache/hop/lineage/spi/ILineageSink.java | 4 +
.../hop/lineage/LineagePipelineGraphTest.java | 199 +++++++++++++++++++++
.../openlineage/OpenLineageEventMapper.java | 1 +
.../openlineage/OpenLineageEventMapperTest.java | 19 ++
9 files changed, 393 insertions(+), 5 deletions(-)
diff --git a/docs/hop-dev-manual/modules/ROOT/pages/lineage.adoc
b/docs/hop-dev-manual/modules/ROOT/pages/lineage.adoc
index f5fdf37a0c..4f13aac749 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 **`RelationalLi [...]
* **`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 [...]
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 [...]
* **`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 ba55541eb9..bf226b7874 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
<<Database lineage>>. 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 40a700aecd..53d1669793 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.context.LineageSubjectType;
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 final class LineageRunLifecycleEmitter {
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<Transition> 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.
+ *
+ * <p>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<Transition> pipelineTransitions(PipelineMeta meta) {
+ if (meta == null) {
+ return List.of();
+ }
+ List<Transition> 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 fc22ad7d04..a81905eade 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.
+ *
+ * <p>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 0000000000..13a837fb75
--- /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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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<Transition> transitions;
+
+ public PipelineGraphLineagePayload(List<Transition> 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 d5bcbf05ec..40ef2f4691 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 @@ public interface ILineageSink {
/**
* Deliver a non-empty batch of events. Implementations should not retain
references to the list
* or events after returning.
+ *
+ * <p>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<LineageEvent> 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 0000000000..a197d70d9c
--- /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<LineageEvent> 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<LineageEvent> 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<LineageEvent> 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<LineageEvent> emitPipeline(PipelineMeta meta,
RunLifecyclePhase phase) {
+ LineageHub hub = mock(LineageHub.class);
+ IPipelineEngine<PipelineMeta> pipeline = mock(IPipelineEngine.class);
+ when(pipeline.getPipelineMeta()).thenReturn(meta);
+ when(pipeline.getLogChannelId()).thenReturn("pipeline-log");
+
+ try (MockedStatic<LineageHub> staticHub =
Mockito.mockStatic(LineageHub.class)) {
+ staticHub.when(LineageHub::getInstance).thenReturn(hub);
+ LineageRunLifecycleEmitter.emitPipeline(pipeline, phase, null);
+ }
+
+ ArgumentCaptor<LineageEvent> 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 6d030c6eef..e8326cc085 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 final class OpenLineageEventMapper {
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 b07952aaba..3762543517 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.FileIoTabularColumn;
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 @@ class OpenLineageEventMapperTest {
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 =
