Feat/job to job lineage

api/src/main/java/marquez/db/Columns.java

@@ -91,6 +91,7 @@ private Columns() {}
 
   /* JOB ROW COLUMNS */
   public static final String PARENT_JOB_NAME = "parent_job_name";
+  public static final String PARENT_JOB_UUID = "parent_job_uuid";
   public static final String SIMPLE_NAME = "simple_name";
   public static final String SYMLINK_TARGET_UUID = "symlink_target_uuid";
 

api/src/main/java/marquez/db/LineageDao.java

@@ -98,6 +98,13 @@ INNER JOIN jobs_view j ON (j.uuid=l2.job_uuid OR j.uuid=l2.job_symlink_target_uu
   """)
   Set<JobData> getLineage(@BindList Set<UUID> jobIds, int depth);
 
+  @SqlQuery(
+      """
+    SELECT j.*, NULL as input_uuids, NULL AS output_uuids FROM jobs_view j
+    WHERE j.parent_job_uuid= :jobId
+    LIMIT 1""")
+  Optional<JobData> getParentJobData(UUID jobId);
+
   @SqlQuery(
       """
       SELECT ds.*, dv.fields, dv.lifecycle_state

api/src/main/java/marquez/db/mappers/JobDataMapper.java

@@ -10,6 +10,7 @@
 import static marquez.db.Columns.timestampOrThrow;
 import static marquez.db.Columns.urlOrNull;
 import static marquez.db.Columns.uuidArrayOrEmpty;
+import static marquez.db.Columns.uuidOrNull;
 import static marquez.db.Columns.uuidOrThrow;
 
 import com.google.common.collect.ImmutableSet;
@@ -39,6 +40,7 @@ public JobData map(@NonNull ResultSet results, @NonNull StatementContext context
         JobName.of(stringOrThrow(results, Columns.NAME)),
         stringOrThrow(results, Columns.SIMPLE_NAME),
         stringOrNull(results, Columns.PARENT_JOB_NAME),
+        uuidOrNull(results, Columns.PARENT_JOB_UUID),
         timestampOrThrow(results, Columns.CREATED_AT),
         timestampOrThrow(results, Columns.UPDATED_AT),
         NamespaceName.of(stringOrThrow(results, Columns.NAMESPACE_NAME)),

api/src/main/java/marquez/db/mappers/JobMapper.java

@@ -57,6 +57,7 @@ public Job map(@NonNull ResultSet results, @NonNull StatementContext context)
             JobName.of(stringOrThrow(results, Columns.NAME)),
             stringOrThrow(results, Columns.SIMPLE_NAME),
             stringOrNull(results, Columns.PARENT_JOB_NAME),
+            uuidOrNull(results, Columns.PARENT_JOB_UUID),
             timestampOrThrow(results, Columns.CREATED_AT),
             timestampOrThrow(results, Columns.UPDATED_AT),
             getDatasetFromJsonOrNull(results, "current_inputs"),

api/src/main/java/marquez/db/mappers/JobRowMapper.java

@@ -45,6 +45,7 @@ public JobRow map(@NonNull ResultSet results, @NonNull StatementContext context)
         stringOrThrow(results, Columns.NAME),
         stringOrThrow(results, Columns.SIMPLE_NAME),
         stringOrNull(results, Columns.PARENT_JOB_NAME),
+        uuidOrNull(results, Columns.PARENT_JOB_UUID),
         stringOrNull(results, Columns.DESCRIPTION),
         uuidOrNull(results, Columns.CURRENT_VERSION_UUID),
         stringOrNull(results, "current_location"),

api/src/main/java/marquez/db/models/JobRow.java

@@ -25,6 +25,7 @@ public class JobRow {
   @NonNull String name;
   @NonNull String simpleName;
   @Nullable String parentJobName;
+  @Nullable UUID parentJoUuid;
   @Nullable String description;
   @Nullable UUID currentVersionUuid;
   @Nullable String location;

api/src/main/java/marquez/service/LineageService.java

@@ -149,6 +149,17 @@ private Lineage toLineage(Set<JobData> jobData, Set<DatasetData> datasets) {
         log.error("Could not find job node for {}", jobData);
         continue;
       }
+
+      Optional<JobData> parentJobData = getParentJobData(data.getParentJobUuid());
+      parentJobData.ifPresent(
+          parent -> {
+            log.debug(
+                "child: {}, parent: {} with UUID: {}",
+                parent.getId().getName(),
+                data.getParentJobName(),
+                data);
+          });
+
       Set<DatasetData> inputs =
           data.getInputUuids().stream()
               .map(datasetById::get)

api/src/main/java/marquez/service/models/Job.java

@@ -32,6 +32,7 @@ public final class Job {
   @Getter private final JobName name;
   @Getter private final String simpleName;
   @Getter private final String parentJobName;
+  @Getter private final UUID parentJobUuid;
   @Getter private final Instant createdAt;
   @Getter private final Instant updatedAt;
   @Getter private final NamespaceName namespace;
@@ -50,6 +51,7 @@ public Job(
       @NonNull final JobName name,
       @NonNull String simpleName,
       @Nullable String parentJobName,
+      @Nullable UUID parentJobUuid,
       @NonNull final Instant createdAt,
       @NonNull final Instant updatedAt,
       @NonNull final Set<DatasetId> inputs,
@@ -65,6 +67,7 @@ public Job(
     this.name = name;
     this.simpleName = simpleName;
     this.parentJobName = parentJobName;
+    this.parentJobUuid = parentJobUuid;
     this.createdAt = createdAt;
     this.updatedAt = updatedAt;
     this.namespace = id.getNamespace();

api/src/main/java/marquez/service/models/JobData.java

@@ -34,6 +34,7 @@ public class JobData implements NodeData {
   @NonNull JobName name;
   @NonNull String simpleName;
   @Nullable String parentJobName;
+  @Nullable UUID parentJobUuid;
   @NonNull Instant createdAt;
   @NonNull Instant updatedAt;
   @NonNull NamespaceName namespace;
@@ -71,4 +72,8 @@ public Set<UUID> getInputUuids() {
   public Set<UUID> getOutputUuids() {
     return outputUuids;
   }
+
+  public UUID getParentJobUuid() {
+    return parentJobUuid;
+  }
 }

api/src/test/java/marquez/db/models/DbModelGenerator.java

@@ -211,6 +211,7 @@ public static JobRow newJobRowWith(
       @NonNull final UUID namespaceUuid,
       @NonNull final String namespaceName) {
     final String parentJobName = newJobName().getValue();
+    final UUID parentJobUuid = newRowUuid();
     final String jobName = newJobName().getValue();
     final String jobSimpleName = jobName;
     return new JobRow(
@@ -223,6 +224,7 @@ public static JobRow newJobRowWith(
         jobName,
         jobSimpleName,
         parentJobName,
+        parentJobUuid,
         newDescription(),
         null,
         newLocation().toString(),