Expose the pipeline hop graph on lineage events, fixes #9004

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`

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

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

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
 }

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

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.
+   *
+   * <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;