Latest runs for jobs (#2901)

* Latest runs coming down. * Adding query param for job listing. * Editing tests. * Bringing in jobs without a run state for tests. * Spotless * Moving to list for lastRunStates. Signed-off-by: phixMe <[email protected]> --------- Signed-off-by: phixMe <[email protected]>
MarquezProject · Oct 8, 2024 · 65d57f7 · 65d57f7
1 parent 1d90f18
commit 65d57f7
Show file tree

Hide file tree

Showing 9 changed files with 84 additions and 33 deletions.
diff --git a/api/src/main/java/marquez/api/JobResource.java b/api/src/main/java/marquez/api/JobResource.java
@@ -12,6 +12,8 @@
 import com.codahale.metrics.annotation.Timed;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import javax.validation.Valid;
 import javax.validation.constraints.Min;
@@ -42,6 +44,7 @@
 import marquez.common.models.JobName;
 import marquez.common.models.NamespaceName;
 import marquez.common.models.RunId;
+import marquez.common.models.RunState;
 import marquez.common.models.TagName;
 import marquez.common.models.Version;
 import marquez.db.JobFacetsDao;
@@ -164,11 +167,19 @@ public Response listJobVersions(
   @Produces(APPLICATION_JSON)
   public Response list(
       @PathParam("namespace") NamespaceName namespaceName,
+      @QueryParam("lastRunStates") List<RunState> lastRunStates,
       @QueryParam("limit") @DefaultValue("100") @Min(value = 0) int limit,
       @QueryParam("offset") @DefaultValue("0") @Min(value = 0) int offset) {
     throwIfNotExists(namespaceName);
 
-    final List<Job> jobs = jobService.findAllWithRun(namespaceName.getValue(), limit, offset);
+    // default to all run states if not specified
+    if (lastRunStates.isEmpty()) {
+      lastRunStates = new ArrayList<>();
+      Collections.addAll(lastRunStates, RunState.values());
+    }
+
+    final List<Job> jobs =
+        jobService.findAllWithRun(namespaceName.getValue(), lastRunStates, limit, offset);
     final int totalCount = jobService.countFor(namespaceName.getValue());
     return Response.ok(new ResultsPage<>("jobs", jobs, totalCount)).build();
   }

diff --git a/api/src/main/java/marquez/db/JobDao.java b/api/src/main/java/marquez/db/JobDao.java
@@ -20,24 +20,28 @@
 import marquez.common.models.JobName;
 import marquez.common.models.JobType;
 import marquez.common.models.NamespaceName;
+import marquez.common.models.RunState;
 import marquez.db.JobVersionDao.IoType;
 import marquez.db.JobVersionDao.JobDataset;
 import marquez.db.JobVersionDao.JobDatasetMapper;
 import marquez.db.mappers.JobMapper;
 import marquez.db.mappers.JobRowMapper;
+import marquez.db.mappers.RunMapper;
 import marquez.db.models.JobRow;
 import marquez.db.models.NamespaceRow;
 import marquez.service.models.Job;
 import marquez.service.models.JobMeta;
 import marquez.service.models.Run;
 import org.jdbi.v3.sqlobject.config.RegisterRowMapper;
+import org.jdbi.v3.sqlobject.customizer.BindList;
 import org.jdbi.v3.sqlobject.statement.SqlQuery;
 import org.jdbi.v3.sqlobject.statement.SqlUpdate;
 import org.postgresql.util.PGobject;
 
 @RegisterRowMapper(JobRowMapper.class)
 @RegisterRowMapper(JobMapper.class)
 @RegisterRowMapper(JobDatasetMapper.class)
+@RegisterRowMapper(RunMapper.class)
 public interface JobDao extends BaseDao {
 
   @SqlQuery(
@@ -138,14 +142,10 @@ default Optional<Job> findWithDatasetsAndRun(String namespaceName, String jobNam
     Optional<Job> job = findJobByName(namespaceName, jobName);
     job.ifPresent(
         j -> {
-          Optional<Run> run = createRunDao().findByLatestJob(namespaceName, jobName);
-          run.ifPresentOrElse(
-              r -> this.setJobData(r, j),
-              () ->
-                  this.setJobData(
-                      createJobVersionDao()
-                          .findCurrentInputOutputDatasetsFor(namespaceName, jobName),
-                      j));
+          List<Run> runs = createRunDao().findByLatestJob(namespaceName, jobName, 1, 0);
+          this.setJobData(runs, j);
+          this.setJobDataset(
+              createJobVersionDao().findCurrentInputOutputDatasetsFor(namespaceName, jobName), j);
         });
     return job;
   }
@@ -244,10 +244,18 @@ job_tags as (
           ON f.run_uuid = jv.latest_run_uuid
         LEFT OUTER JOIN job_tags jt
           ON j.uuid  = jt.uuid
+        LEFT JOIN runs r
+          ON r.uuid = jv.latest_run_uuid
+        WHERE
+         (r.current_run_state IN (<lastRunStates>) OR r.uuid IS NULL)
         ORDER BY
             j.name
       """)
-  List<Job> findAll(String namespaceName, int limit, int offset);
+  List<Job> findAll(
+      String namespaceName,
+      @BindList("lastRunStates") List<RunState> lastRunStates,
+      int limit,
+      int offset);
 
   @SqlQuery("SELECT count(*) FROM jobs_view AS j WHERE symlink_target_uuid IS NULL")
   int count();
@@ -271,18 +279,19 @@ job_tags as (
           + "AND symlink_target_uuid IS NULL")
   int countFor(String namespaceName);
 
-  default List<Job> findAllWithRun(String namespaceName, int limit, int offset) {
+  default List<Job> findAllWithRun(
+      String namespaceName, List<RunState> lastRunStates, int limit, int offset) {
     RunDao runDao = createRunDao();
-    return findAll(namespaceName, limit, offset).stream()
+    return findAll(namespaceName, lastRunStates, limit, offset).stream()
         .peek(
-            j ->
-                runDao
-                    .findByLatestJob(namespaceName, j.getName().getValue())
-                    .ifPresent(run -> this.setJobData(run, j)))
-        .collect(Collectors.toList());
+            j -> {
+              List<Run> runs = runDao.findByLatestJob(namespaceName, j.getName().getValue(), 10, 0);
+              this.setJobData(runs, j);
+            })
+        .toList();
   }
 
-  default void setJobData(List<JobDataset> datasets, Job j) {
+  default void setJobDataset(List<JobDataset> datasets, Job j) {
     Optional.of(
             datasets.stream()
                 .filter(d -> d.ioType().equals(IoType.INPUT))
@@ -304,19 +313,25 @@ default void setJobData(List<JobDataset> datasets, Job j) {
         .ifPresent(s -> j.setOutputs(s));
   }
 
-  default void setJobData(Run run, Job j) {
-    j.setLatestRun(run);
+  default void setJobData(List<Run> runs, Job j) {
+    if (runs.isEmpty()) {
+      return;
+    }
+
+    Run latestRun = runs.get(0);
+    j.setLatestRun(latestRun);
+    j.setLatestRuns(runs);
     DatasetVersionDao datasetVersionDao = createDatasetVersionDao();
     j.setInputs(
-        datasetVersionDao.findInputDatasetVersionsFor(run.getId().getValue()).stream()
+        datasetVersionDao.findInputDatasetVersionsFor(latestRun.getId().getValue()).stream()
             .map(
                 ds ->
                     new DatasetId(
                         NamespaceName.of(ds.getNamespaceName()),
                         DatasetName.of(ds.getDatasetName())))
             .collect(Collectors.toSet()));
     j.setOutputs(
-        datasetVersionDao.findOutputDatasetVersionsFor(run.getId().getValue()).stream()
+        datasetVersionDao.findOutputDatasetVersionsFor(latestRun.getId().getValue()).stream()
             .map(
                 ds ->
                     new DatasetId(

diff --git a/api/src/main/java/marquez/db/RunDao.java b/api/src/main/java/marquez/db/RunDao.java
@@ -510,15 +510,15 @@ default RunRow upsertRunMeta(
   @SqlQuery(
       BASE_FIND_RUN_SQL
           + """
-      WHERE r.uuid=(
+      WHERE r.uuid IN (
         SELECT r.uuid FROM runs_view r
         INNER JOIN jobs_view j ON j.namespace_name=r.namespace_name AND j.name=r.job_name
         WHERE j.namespace_name=:namespace AND (j.name=:jobName OR j.name=ANY(j.aliases))
         ORDER BY transitioned_at DESC
-        LIMIT 1
+        LIMIT :limit OFFSET :offset
       )
       """)
-  Optional<Run> findByLatestJob(String namespace, String jobName);
+  List<Run> findByLatestJob(String namespace, String jobName, int limit, int offset);
 
   @Builder
   record RunUpsert(

diff --git a/api/src/main/java/marquez/db/mappers/JobMapper.java b/api/src/main/java/marquez/db/mappers/JobMapper.java
@@ -71,6 +71,7 @@ public Job map(@NonNull ResultSet results, @NonNull StatementContext context)
             // Latest Run is resolved in the JobDao. This can be brought in via a join and
             //  and a jsonb but custom deserializers will need to be introduced
             null,
+            null,
             facetsOrNull,
             uuidOrNull(results, Columns.CURRENT_VERSION_UUID),
             getLabels(facetsOrNull),

diff --git a/api/src/main/java/marquez/service/models/Job.java b/api/src/main/java/marquez/service/models/Job.java
@@ -10,6 +10,7 @@
 import com.google.common.collect.ImmutableSet;
 import java.net.URL;
 import java.time.Instant;
+import java.util.List;
 import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
@@ -43,6 +44,7 @@ public final class Job {
   @Nullable private final URL location;
   @Nullable private final String description;
   @Nullable @Setter private Run latestRun;
+  @Nullable @Setter private List<Run> latestRuns;
   @Getter private final ImmutableMap<String, Object> facets;
   @Nullable private UUID currentVersion;
   @Getter @Nullable private ImmutableList<String> labels;
@@ -62,6 +64,7 @@ public Job(
       @Nullable final URL location,
       @Nullable final String description,
       @Nullable final Run latestRun,
+      @Nullable final List<Run> latestRuns,
       @Nullable final ImmutableMap<String, Object> facets,
       @Nullable UUID currentVersion,
       @Nullable ImmutableList<String> labels,
@@ -80,6 +83,7 @@ public Job(
     this.location = location;
     this.description = description;
     this.latestRun = latestRun;
+    this.latestRuns = latestRuns;
     this.facets = (facets == null) ? ImmutableMap.of() : facets;
     this.currentVersion = currentVersion;
     this.labels = (labels == null) ? ImmutableList.of() : labels;
@@ -98,6 +102,10 @@ public Optional<Run> getLatestRun() {
     return Optional.ofNullable(latestRun);
   }
 
+  public Optional<List<Run>> getLatestRuns() {
+    return Optional.ofNullable(latestRuns);
+  }
+
   public Optional<UUID> getCurrentVersion() {
     return Optional.ofNullable(currentVersion);
   }

diff --git a/api/src/test/java/marquez/db/JobDaoTest.java b/api/src/test/java/marquez/db/JobDaoTest.java
@@ -17,10 +17,13 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.sql.SQLException;
 import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
 import marquez.common.models.JobType;
+import marquez.common.models.RunState;
 import marquez.db.models.DbModelGenerator;
 import marquez.db.models.JobRow;
 import marquez.db.models.NamespaceRow;
@@ -106,7 +109,10 @@ public void testFindAll() {
     JobRow anotherJobSameNamespace =
         createJobWithoutSymlinkTarget(jdbi, namespace, "anotherJob", "a random other job");
 
-    List<Job> jobs = jobDao.findAll(namespace.getName(), 10, 0);
+    List<RunState> runStates = new ArrayList<>();
+    Collections.addAll(runStates, RunState.values());
+
+    List<Job> jobs = jobDao.findAll(namespace.getName(), runStates, 10, 0);
 
     // the symlinked job isn't present in the response - only the symlink target and the job with
     // no symlink

diff --git a/api/src/test/java/marquez/db/RunDaoTest.java b/api/src/test/java/marquez/db/RunDaoTest.java
@@ -165,10 +165,9 @@ public void testFindByLatestJob() {
     TreeSet<RunRow> sortedRuns =
         new TreeSet<>(Comparator.comparing(RunRow::getUpdatedAt).reversed());
     sortedRuns.addAll(runs);
-    Optional<Run> byLatestJob = runDao.findByLatestJob(jobRow.getNamespaceName(), jobRow.getName());
+    Run byLatestJob =
+        runDao.findByLatestJob(jobRow.getNamespaceName(), jobRow.getName(), 1, 0).get(0);
     assertThat(byLatestJob)
-        .isPresent()
-        .get()
         .hasFieldOrPropertyWithValue("id", new RunId(sortedRuns.first().getUuid()));
 
     JobRow newTargetJob =
@@ -183,10 +182,11 @@ public void testFindByLatestJob() {
         jobMeta.getDescription().orElse(null));
 
     // get the latest run for the *newTargetJob*. It should be the same as the old job's latest run
-    byLatestJob = runDao.findByLatestJob(newTargetJob.getNamespaceName(), newTargetJob.getName());
+    byLatestJob =
+        runDao
+            .findByLatestJob(newTargetJob.getNamespaceName(), newTargetJob.getName(), 1, 0)
+            .get(0);
     assertThat(byLatestJob)
-        .isPresent()
-        .get()
         .hasFieldOrPropertyWithValue("id", new RunId(sortedRuns.first().getUuid()));
   }
 

diff --git a/api/src/test/java/marquez/service/LineageServiceTest.java b/api/src/test/java/marquez/service/LineageServiceTest.java
@@ -11,6 +11,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertTrue;
 
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -28,6 +29,7 @@
 import marquez.common.models.JobName;
 import marquez.common.models.NamespaceName;
 import marquez.common.models.OutputDatasetVersion;
+import marquez.common.models.RunState;
 import marquez.db.DatasetDao;
 import marquez.db.JobDao;
 import marquez.db.LineageDao;
@@ -205,7 +207,10 @@ && jobNameEquals(n, "downstreamJob0<-outputData<-readJob0<-commonDataset"))
                 new NamespaceName(NAMESPACE),
                 new DatasetName("outputData<-readJob0<-commonDataset")));
 
-    List<Job> jobs = jobDao.findAllWithRun(NAMESPACE, 1000, 0);
+    List<RunState> runStates = new ArrayList<>();
+    Collections.addAll(runStates, RunState.values());
+
+    List<Job> jobs = jobDao.findAllWithRun(NAMESPACE, runStates, 1000, 0);
     jobs =
         jobs.stream()
             .filter(j -> j.getName().getValue().contains("newDownstreamJob"))

diff --git a/spec/openapi.yml b/spec/openapi.yml
@@ -1596,6 +1596,10 @@ components:
           type: string
         latestRun:
           $ref: '#/components/schemas/Run'
+        latestRuns:
+          type: array
+          items:
+            $ref: '#/components/schemas/Run'
         facets:
           $ref: '#/components/schemas/JobFacets'
         currentVersion:
@@ -1617,6 +1621,7 @@ components:
         context: {'SQL': "SELECT * FROM mytable;"}
         description: My first job!
         latestRun: null
+        latestRuns: []
         facets: {}
         currentVersion: "b1d626a2-6d3a-475e-9ecf-943176d4a8c6"