Added parent job name to job models

Signed-off-by: Michael Collado <collado.mike@gmail.com>

Author: collado-mike

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

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

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

@@ -38,6 +38,7 @@ public JobData map(@NonNull ResultSet results, @NonNull StatementContext context
         JobType.valueOf(stringOrThrow(results, Columns.TYPE)),
         JobName.of(stringOrThrow(results, Columns.NAME)),
         stringOrThrow(results, Columns.SIMPLE_NAME),
+        stringOrNull(results, Columns.PARENT_JOB_NAME),
         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

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

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

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

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

@@ -32,6 +32,7 @@ public class JobData implements NodeData {
   @NonNull JobType type;
   @NonNull JobName name;
   @NonNull String simpleName;
+  @Nullable String parentJobName;
   @NonNull Instant createdAt;
   @NonNull Instant updatedAt;
   @NonNull NamespaceName namespace;

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

@@ -20,6 +20,7 @@ public class JobRow {
   @NonNull String namespaceName;
   @NonNull String name;
   @NonNull String simpleName;
+  @Nullable String parentJobName;
   @Nullable String description;
   @Nullable UUID currentVersionUuid;
   @Nullable UUID jobContextUuid;

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

@@ -27,6 +27,7 @@ public final class Job {
   @Getter private final JobType type;
   @Getter private final JobName name;
   @Getter private final String simpleName;
+  @Getter private final String parentJobName;
   @Getter private final Instant createdAt;
   @Getter private final Instant updatedAt;
   @Getter private final NamespaceName namespace;
@@ -44,6 +45,7 @@ public Job(
       @NonNull final JobType type,
       @NonNull final JobName name,
       @NonNull String simpleName,
+      @Nullable String parentJobName,
       @NonNull final Instant createdAt,
       @NonNull final Instant updatedAt,
       @NonNull final Set<DatasetId> inputs,
@@ -58,6 +60,7 @@ public Job(
     this.type = type;
     this.name = name;
     this.simpleName = simpleName;
+    this.parentJobName = parentJobName;
     this.createdAt = createdAt;
     this.updatedAt = updatedAt;
     this.namespace = id.getNamespace();

api/src/main/resources/marquez/db/migration/V43__alter_jobs_add_job_parent_id.sql

@@ -14,12 +14,13 @@ CREATE OR REPLACE VIEW jobs_view
 AS
     with recursive
      job_fqn AS (
-        SELECT uuid, name, namespace_name, parent_job_uuid
+        SELECT uuid, name, namespace_name, NULL::text AS parent_job_name, parent_job_uuid
         FROM jobs
-        UNION ALL
+        UNION
         SELECT j1.uuid,
                CASE WHEN j2.name IS NOT NULL THEN j2.name || '.' || j1.name ELSE j1.name END AS name,
                CASE WHEN j2.namespace_name IS NOT NULL THEN j2.namespace_name ELSE j1.namespace_name END AS namespace_name,
+               j2.name AS parent_job_name,
                j2.parent_job_uuid
         FROM jobs j1
         INNER JOIN job_fqn j2 ON j2.uuid=j1.parent_job_uuid
@@ -29,6 +30,7 @@ AS
            f.namespace_name,
            j.name AS simple_name,
            j.parent_job_uuid,
+           f.parent_job_name,
            j.type,
            j.created_at,
            j.updated_at,

api/src/test/java/marquez/OpenLineageIntegrationTest.java

@@ -202,12 +202,14 @@ public void testOpenLineageJobHierarchy()
     Job job = client.getJob(NAMESPACE_NAME, dagName + "." + task1Name);
     assertThat(job)
         .isNotNull()
-        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName + "." + task1Name));
+        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName + "." + task1Name))
+        .hasFieldOrPropertyWithValue("parentJobName", dagName);
 
     Job parentJob = client.getJob(NAMESPACE_NAME, dagName);
     assertThat(parentJob)
         .isNotNull()
-        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName));
+        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName))
+        .hasFieldOrPropertyWithValue("parentJobName", null);
     List<Run> runsList = client.listRuns(NAMESPACE_NAME, dagName);
     assertThat(runsList).isNotEmpty().hasSize(1);
   }
@@ -244,12 +246,14 @@ public void testOpenLineageJobHierarchyOldAirflowIntegration()
     assertThat(job)
         .isNotNull()
         .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName + "." + task1Name))
-        .hasFieldOrPropertyWithValue("simpleName", task1Name);
+        .hasFieldOrPropertyWithValue("simpleName", task1Name)
+        .hasFieldOrPropertyWithValue("parentJobName", dagName);
 
     Job parentJob = client.getJob(NAMESPACE_NAME, dagName);
     assertThat(parentJob)
         .isNotNull()
-        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName));
+        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName))
+        .hasFieldOrPropertyWithValue("parentJobName", null);
     List<Run> runsList = client.listRuns(NAMESPACE_NAME, dagName);
     assertThat(runsList).isNotEmpty().hasSize(1);
     UUID parentRunUuid = Utils.toNameBasedUuid(dagName, airflowParentRunId);
@@ -290,19 +294,22 @@ public void testOpenLineageJobHierarchySparkAndAirflow()
     assertThat(airflowTask)
         .isNotNull()
         .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName + "." + task1Name))
-        .hasFieldOrPropertyWithValue("simpleName", task1Name);
+        .hasFieldOrPropertyWithValue("simpleName", task1Name)
+        .hasFieldOrPropertyWithValue("parentJobName", dagName);
 
     Job sparkJob = client.getJob(NAMESPACE_NAME, dagName + "." + task1Name + "." + sparkTaskName);
     assertThat(sparkJob)
         .isNotNull()
         .hasFieldOrPropertyWithValue(
             "id", new JobId(NAMESPACE_NAME, dagName + "." + task1Name + "." + sparkTaskName))
-        .hasFieldOrPropertyWithValue("simpleName", sparkTaskName);
+        .hasFieldOrPropertyWithValue("simpleName", sparkTaskName)
+        .hasFieldOrPropertyWithValue("parentJobName", dagName + "." + task1Name);
 
     Job parentJob = client.getJob(NAMESPACE_NAME, dagName);
     assertThat(parentJob)
         .isNotNull()
-        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName));
+        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName))
+        .hasFieldOrPropertyWithValue("parentJobName", null);
     List<Run> runsList = client.listRuns(NAMESPACE_NAME, dagName);
     assertThat(runsList).isNotEmpty().hasSize(1);
   }

clients/java/src/main/java/marquez/client/models/Job.java

@@ -23,6 +23,7 @@ public final class Job extends JobMeta {
   @Getter private final JobId id;
   @Getter private final String name;
   @Getter private final String simpleName;
+  @Getter private final String parentJobName;
   @Getter private final Instant createdAt;
   @Getter private final Instant updatedAt;
   @Getter private final String namespace;
@@ -35,6 +36,7 @@ public Job(
       final JobType type,
       @NonNull final String name,
       @NonNull final String simpleName,
+      final String parentJobName,
       @NonNull final Instant createdAt,
       @NonNull final Instant updatedAt,
       @NonNull final String namespace,
@@ -50,6 +52,7 @@ public Job(
     this.id = id;
     this.name = name;
     this.simpleName = simpleName;
+    this.parentJobName = parentJobName;
     this.createdAt = createdAt;
     this.updatedAt = updatedAt;
     this.namespace = namespace;