fix(assets): use manifest parent_map for accurate lineage (#240)

* fix(assets): use manifest parent_map for accurate lineage

Use the manifest-level parent_map (the canonical DAG) to resolve
asset dependencies instead of node-level depends_on.nodes which
can include extra edges not reflecting the real dbt DAG structure.

* feat(assets): ensure lineage from the example is correct

* fix(assets): properly set inputs & outputs

---------

Co-authored-by: François Delbrayelle <fdelbrayelle@kestra.io>

Author: fdelbrayelle

src/main/java/io/kestra/plugin/dbt/ResultParser.java

@@ -5,6 +5,7 @@
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
 import io.kestra.core.models.assets.AssetIdentifier;
 import io.kestra.core.models.assets.AssetsInOut;
+import io.kestra.core.models.assets.Asset;
 import io.kestra.core.models.assets.Custom;
 import io.kestra.core.models.executions.TaskRun;
 import io.kestra.core.models.executions.TaskRunAttempt;
@@ -150,16 +151,9 @@ private static AssetsInOut assetsFor(String uniqueId, Map<String, ModelAsset> mo
         }
 
         List<AssetIdentifier> inputs = inputIdentifiers(modelAsset, modelAssets);
+        List<Asset> outputs = outputAssets(modelAsset, modelAssets);
 
-        return new AssetsInOut(
-            inputs,
-            List.of(Custom.builder()
-                .id(modelAsset.assetId())
-                .type(TABLE_ASSET_TYPE)
-                .metadata(modelAsset.metadata())
-                .build()
-            )
-        );
+        return new AssetsInOut(inputs, outputs);
     }
 
     private static void emitAssets(RunContext runContext, Manifest manifest) throws IllegalVariableEvaluationException {
@@ -168,24 +162,32 @@ private static void emitAssets(RunContext runContext, Manifest manifest) throws
 
         for (ModelAsset asset : modelAssets.values()) {
             List<AssetIdentifier> inputs = inputIdentifiers(asset, modelAssets);
+            List<Asset> outputs = outputAssets(asset, modelAssets);
             try {
-                runContext.assets().emit(new AssetEmit(
-                    inputs,
-                    List.of(Custom.builder()
-                        .id(asset.assetId())
-                        .type(TABLE_ASSET_TYPE)
-                        .metadata(asset.metadata())
-                        .build()
-                    )
-                )
-                );
+                runContext.assets().emit(new AssetEmit(inputs, outputs));
             } catch (UnsupportedOperationException | QueueException e) {
                 // UnsupportedOperationException for OSS or tests where EE is not configured (assets are EE only)
                 runContext.logger().warn("Unable to emit dbt asset '{}'", asset.assetId(), e);
             }
         }
     }
 
+    private static List<Asset> outputAssets(ModelAsset modelAsset, Map<String, ModelAsset> modelAssets) {
+        if (modelAsset.children() == null || modelAsset.children().isEmpty()) {
+            return List.of();
+        }
+
+        return modelAsset.children().stream()
+            .map(modelAssets::get)
+            .filter(Objects::nonNull)
+            .<Asset>map(child -> Custom.builder()
+                .id(child.assetId())
+                .type(TABLE_ASSET_TYPE)
+                .metadata(child.metadata())
+                .build())
+            .toList();
+    }
+
     private static List<AssetIdentifier> inputIdentifiers(ModelAsset modelAsset, Map<String, ModelAsset> modelAssets) {
         if (modelAsset.dependsOn() == null || modelAsset.dependsOn().isEmpty()) {
             return List.of();
@@ -226,12 +228,18 @@ private static Map<String, ModelAsset> extractModelAssets(Manifest manifest) {
             if (hasValue(node.getSchema())) metadata.put("schema", node.getSchema());
             if (hasValue(name)) metadata.put("name", name);
 
-            List<String> dependsOn = List.of();
-            if (node.getDependsOn() != null) {
+            // Use parent_map from manifest (the canonical DAG) when available,
+            // falling back to node-level depends_on for older manifests.
+            List<String> dependsOn;
+            if (manifest.getParentMap() != null && manifest.getParentMap().containsKey(uniqueId)) {
+                dependsOn = manifest.getParentMap().get(uniqueId);
+            } else if (node.getDependsOn() != null) {
                 dependsOn = node.getDependsOn().getOrDefault("nodes", List.of());
+            } else {
+                dependsOn = List.of();
             }
 
-            modelAssets.put(uniqueId, new ModelAsset(assetId, metadata, dependsOn));
+            modelAssets.put(uniqueId, new ModelAsset(assetId, metadata, dependsOn, List.of()));
         }
 
         Map<String, ModelAsset> filtered = new HashMap<>(modelAssets.size());
@@ -240,10 +248,26 @@ private static Map<String, ModelAsset> extractModelAssets(Manifest manifest) {
             List<String> deps = a.dependsOn() == null ? List.of() : a.dependsOn().stream()
                 .filter(modelAssets::containsKey)
                 .toList();
-            filtered.put(e.getKey(), new ModelAsset(a.assetId(), a.metadata(), deps));
+            filtered.put(e.getKey(), new ModelAsset(a.assetId(), a.metadata(), deps, List.of()));
+        }
+
+        // Compute reverse dependencies (children: models that depend on this one)
+        Map<String, List<String>> childrenMap = new HashMap<>();
+        for (Map.Entry<String, ModelAsset> e : filtered.entrySet()) {
+            for (String dep : e.getValue().dependsOn()) {
+                childrenMap.computeIfAbsent(dep, k -> new ArrayList<>()).add(e.getKey());
+            }
+        }
+
+        // Rebuild with children populated
+        Map<String, ModelAsset> result = new HashMap<>(filtered.size());
+        for (Map.Entry<String, ModelAsset> e : filtered.entrySet()) {
+            ModelAsset a = e.getValue();
+            List<String> children = childrenMap.getOrDefault(e.getKey(), List.of());
+            result.put(e.getKey(), new ModelAsset(a.assetId(), a.metadata(), a.dependsOn(), children));
         }
 
-        return filtered;
+        return result;
     }
 
     private static String adapterType(Manifest manifest) {
@@ -287,6 +311,6 @@ private static boolean hasValue(String value) {
         return value != null && !value.trim().isEmpty();
     }
 
-    private record ModelAsset(String assetId, Map<String, Object> metadata, List<String> dependsOn) {
+    private record ModelAsset(String assetId, Map<String, Object> metadata, List<String> dependsOn, List<String> children) {
     }
 }

src/main/java/io/kestra/plugin/dbt/models/Manifest.java

@@ -15,6 +15,9 @@ public class Manifest {
     Map<String, Object> metadata;
     Map<String, Node> nodes;
 
+    @JsonProperty("parent_map")
+    Map<String, List<String>> parentMap;
+
     @Value
     @Jacksonized
     @SuperBuilder

src/test/java/io/kestra/plugin/dbt/ResultParserTest.java

@@ -10,12 +10,12 @@
 import jakarta.inject.Inject;
 import org.junit.jupiter.api.Test;
 
+import io.kestra.core.runners.AssetEmit;
+
 import java.nio.file.Files;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-import static java.util.stream.Collectors.toMap;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.*;
 
@@ -56,6 +56,12 @@ void parseManifestWithAssets_shouldEmitModelAssets() throws Exception {
                     ]
                   }
                 }
+              },
+              "parent_map": {
+                "model.analytics.stg_orders": [],
+                "model.analytics.fct_orders": [
+                  "model.analytics.stg_orders"
+                ]
               }
             }
             """);
@@ -65,22 +71,23 @@ void parseManifestWithAssets_shouldEmitModelAssets() throws Exception {
         assertThat(manifestResult.manifest(), is(notNullValue()));
         assertThat(runContext.assets().emitted(), hasSize(2));
 
-        var outputAssets = runContext.assets().emitted().stream()
-            .flatMap(assetEmit -> assetEmit.outputs().stream())
-            .toList();
-
-        assertThat(outputAssets, hasSize(2));
-        var byId = new HashMap<String, io.kestra.core.models.assets.Asset>();
-        outputAssets.forEach(asset -> byId.put(asset.getId(), asset));
+        // stg_orders has no inputs and 1 output (fct_orders, its child)
+        // fct_orders has 1 input (stg_orders) and no outputs (leaf node)
+        var stgOrdersEmit = findEmitWithOutput(runContext.assets().emitted(), "analytics.marts.fct_orders");
+        assertThat("stg_orders emission should exist", stgOrdersEmit, is(notNullValue()));
+        assertThat(stgOrdersEmit.inputs(), hasSize(0));
+        assertThat(stgOrdersEmit.outputs(), hasSize(1));
 
-        assertThat(byId.containsKey("analytics.staging.stg_orders"), is(true));
-        assertThat(byId.containsKey("analytics.marts.fct_orders"), is(true));
+        var fctOrdersOutput = stgOrdersEmit.outputs().getFirst();
+        assertThat(fctOrdersOutput.getMetadata().get("system"), is("postgres"));
+        assertThat(fctOrdersOutput.getMetadata().get("database"), is("analytics"));
+        assertThat(fctOrdersOutput.getMetadata().get("schema"), is("marts"));
+        assertThat(fctOrdersOutput.getMetadata().get("name"), is("fct_orders"));
 
-        var stgOrders = byId.get("analytics.staging.stg_orders");
-        assertThat(stgOrders.getMetadata().get("system"), is("postgres"));
-        assertThat(stgOrders.getMetadata().get("database"), is("analytics"));
-        assertThat(stgOrders.getMetadata().get("schema"), is("staging"));
-        assertThat(stgOrders.getMetadata().get("name"), is("stg_orders"));
+        var fctOrdersEmit = findEmitWithInput(runContext.assets().emitted(), "analytics.staging.stg_orders");
+        assertThat("fct_orders emission should exist", fctOrdersEmit, is(notNullValue()));
+        assertThat(fctOrdersEmit.inputs(), hasSize(1));
+        assertThat(fctOrdersEmit.outputs(), hasSize(0));
     }
 
     @Test
@@ -115,24 +122,136 @@ void parseManifestWithAssets_shouldEmitLineageInputs() throws Exception {
                     ]
                   }
                 }
+              },
+              "parent_map": {
+                "model.analytics.my_first_dbt_model": [],
+                "model.analytics.my_second_dbt_model": [
+                  "model.analytics.my_first_dbt_model"
+                ]
               }
             }
             """);
 
         ResultParser.parseManifestWithAssets(runContext, manifestFile.toFile());
 
-        var emittedByOutputId = runContext.assets().emitted().stream()
-            .collect(toMap(
-                assetEmit -> assetEmit.outputs().getFirst().getId(),
-                assetEmit -> assetEmit
-            ));
+        assertThat(runContext.assets().emitted(), hasSize(2));
 
-        assertThat(emittedByOutputId, hasKey("analytics.marts.my_first_dbt_model"));
-        assertThat(emittedByOutputId, hasKey("analytics.marts.my_second_dbt_model"));
+        // my_first_dbt_model: no inputs, 1 output (my_second_dbt_model)
+        var firstModelEmit = findEmitWithOutput(runContext.assets().emitted(), "analytics.marts.my_second_dbt_model");
+        assertThat(firstModelEmit, is(notNullValue()));
+        assertThat(firstModelEmit.inputs(), hasSize(0));
+        assertThat(firstModelEmit.outputs(), hasSize(1));
 
-        var secondModelEmit = emittedByOutputId.get("analytics.marts.my_second_dbt_model");
+        // my_second_dbt_model: 1 input (my_first_dbt_model), no outputs (leaf)
+        var secondModelEmit = findEmitWithInput(runContext.assets().emitted(), "analytics.marts.my_first_dbt_model");
+        assertThat(secondModelEmit, is(notNullValue()));
         assertThat(secondModelEmit.inputs(), hasSize(1));
         assertThat(secondModelEmit.inputs().getFirst().id(), is("analytics.marts.my_first_dbt_model"));
+        assertThat(secondModelEmit.outputs(), hasSize(0));
+    }
+
+    @Test
+    void parseManifestWithAssets_shouldUseParentMapForLineage() throws Exception {
+        // Simulate a case where depends_on.nodes includes transitive deps
+        // but parent_map only has the direct edges (the real DAG).
+        var runContext = mockRunContext();
+        var manifestFile = runContext.workingDir().path(true).resolve("manifest.json");
+        Files.writeString(manifestFile, """
+            {
+              "metadata": {
+                "adapter_type": "duckdb"
+              },
+              "nodes": {
+                "model.project.stg_orders": {
+                  "resource_type": "model",
+                  "database": "dev",
+                  "schema": "staging",
+                  "name": "stg_orders",
+                  "unique_id": "model.project.stg_orders",
+                  "depends_on": {
+                    "nodes": ["source.project.raw.orders"]
+                  }
+                },
+                "model.project.int_orders": {
+                  "resource_type": "model",
+                  "database": "dev",
+                  "schema": "intermediate",
+                  "name": "int_orders",
+                  "unique_id": "model.project.int_orders",
+                  "depends_on": {
+                    "nodes": ["model.project.stg_orders"]
+                  }
+                },
+                "model.project.fct_orders": {
+                  "resource_type": "model",
+                  "database": "dev",
+                  "schema": "marts",
+                  "name": "fct_orders",
+                  "unique_id": "model.project.fct_orders",
+                  "depends_on": {
+                    "nodes": ["model.project.stg_orders", "model.project.int_orders"]
+                  }
+                }
+              },
+              "parent_map": {
+                "model.project.stg_orders": ["source.project.raw.orders"],
+                "model.project.int_orders": ["model.project.stg_orders"],
+                "model.project.fct_orders": ["model.project.int_orders"]
+              }
+            }
+            """);
+
+        ResultParser.parseManifestWithAssets(runContext, manifestFile.toFile());
+
+        assertThat(runContext.assets().emitted(), hasSize(3));
+
+        // DAG: stg_orders → int_orders → fct_orders (parent_map, no transitive edges)
+        // Inputs = upstream deps, Outputs = downstream children
+
+        // stg_orders: no model inputs (source filtered out), 1 output (int_orders)
+        var stgOrdersEmit = findEmitWithOutput(runContext.assets().emitted(), "dev.intermediate.int_orders");
+        assertThat(stgOrdersEmit, is(notNullValue()));
+        assertThat(stgOrdersEmit.inputs(), hasSize(0));
+        assertThat(stgOrdersEmit.outputs(), hasSize(1));
+
+        // int_orders: 1 input (stg_orders), 1 output (fct_orders)
+        var intOrdersEmit = findEmitWithInputAndOutput(runContext.assets().emitted(),
+            "dev.staging.stg_orders", "dev.marts.fct_orders");
+        assertThat(intOrdersEmit, is(notNullValue()));
+        assertThat(intOrdersEmit.inputs(), hasSize(1));
+        assertThat(intOrdersEmit.inputs().getFirst().id(), is("dev.staging.stg_orders"));
+        assertThat(intOrdersEmit.outputs(), hasSize(1));
+        assertThat(intOrdersEmit.outputs().getFirst().getId(), is("dev.marts.fct_orders"));
+
+        // fct_orders: 1 input (int_orders only, from parent_map), no outputs (leaf)
+        var fctOrdersEmit = findEmitWithInput(runContext.assets().emitted(), "dev.intermediate.int_orders");
+        assertThat(fctOrdersEmit, is(notNullValue()));
+        assertThat(fctOrdersEmit.inputs(), hasSize(1));
+        assertThat(fctOrdersEmit.inputs().getFirst().id(), is("dev.intermediate.int_orders"));
+        assertThat(fctOrdersEmit.outputs(), hasSize(0));
+    }
+
+    private static AssetEmit findEmitWithOutput(List<AssetEmit> emitted, String outputId) {
+        return emitted.stream()
+            .filter(e -> e.outputs().stream().anyMatch(o -> o.getId().equals(outputId)))
+            .findFirst()
+            .orElse(null);
+    }
+
+    private static AssetEmit findEmitWithInput(List<AssetEmit> emitted, String inputId) {
+        return emitted.stream()
+            .filter(e -> e.inputs().stream().anyMatch(i -> i.id().equals(inputId)))
+            .filter(e -> e.outputs().isEmpty())
+            .findFirst()
+            .orElse(null);
+    }
+
+    private static AssetEmit findEmitWithInputAndOutput(List<AssetEmit> emitted, String inputId, String outputId) {
+        return emitted.stream()
+            .filter(e -> e.inputs().stream().anyMatch(i -> i.id().equals(inputId)))
+            .filter(e -> e.outputs().stream().anyMatch(o -> o.getId().equals(outputId)))
+            .findFirst()
+            .orElse(null);
     }
 
     private RunContext mockRunContext() {