Remove job context.

Signed-off-by: Jakub Dardzinski <[email protected]> Add call to API in frontend. Add tests. Signed-off-by: Jakub Dardzinski <[email protected]> Update javadocs. Signed-off-by: Jakub Dardzinski <[email protected]> Apply spotless fixes. Signed-off-by: Jakub Dardzinski <[email protected]>
MarquezProject · Jan 20, 2023 · d9daa4e · d9daa4e
1 parent 3492df2
commit d9daa4e
Show file tree

Hide file tree

Showing 76 changed files with 499 additions and 481 deletions.
diff --git a/api/src/main/java/marquez/MarquezContext.java b/api/src/main/java/marquez/MarquezContext.java
@@ -15,6 +15,7 @@
 import lombok.NonNull;
 import marquez.api.ColumnLineageResource;
 import marquez.api.DatasetResource;
+import marquez.api.FacetResource;
 import marquez.api.JobResource;
 import marquez.api.NamespaceResource;
 import marquez.api.OpenLineageResource;
@@ -28,7 +29,6 @@
 import marquez.db.DatasetDao;
 import marquez.db.DatasetFieldDao;
 import marquez.db.DatasetVersionDao;
-import marquez.db.JobContextDao;
 import marquez.db.JobDao;
 import marquez.db.JobVersionDao;
 import marquez.db.LineageDao;
@@ -67,7 +67,6 @@ public final class MarquezContext {
   @Getter private final DatasetVersionDao datasetVersionDao;
   @Getter private final JobDao jobDao;
   @Getter private final JobVersionDao jobVersionDao;
-  @Getter private final JobContextDao jobContextDao;
   @Getter private final RunDao runDao;
   @Getter private final RunArgsDao runArgsDao;
   @Getter private final RunStateDao runStateDao;
@@ -91,6 +90,7 @@ public final class MarquezContext {
   @Getter private final SourceResource sourceResource;
   @Getter private final DatasetResource datasetResource;
   @Getter private final ColumnLineageResource columnLineageResource;
+  @Getter private final FacetResource facetResource;
   @Getter private final JobResource jobResource;
   @Getter private final TagResource tagResource;
   @Getter private final OpenLineageResource openLineageResource;
@@ -116,7 +116,6 @@ private MarquezContext(
     this.datasetVersionDao = jdbi.onDemand(DatasetVersionDao.class);
     this.jobDao = jdbi.onDemand(JobDao.class);
     this.jobVersionDao = jdbi.onDemand(JobVersionDao.class);
-    this.jobContextDao = jdbi.onDemand(JobContextDao.class);
     this.runDao = jdbi.onDemand(RunDao.class);
     this.runArgsDao = jdbi.onDemand(RunArgsDao.class);
     this.runStateDao = jdbi.onDemand(RunStateDao.class);
@@ -158,6 +157,7 @@ private MarquezContext(
     this.sourceResource = new SourceResource(serviceFactory);
     this.datasetResource = new DatasetResource(serviceFactory);
     this.columnLineageResource = new ColumnLineageResource(serviceFactory);
+    this.facetResource = new FacetResource(serviceFactory, openLineageDao);
     this.jobResource = new JobResource(serviceFactory, jobVersionDao);
     this.tagResource = new TagResource(serviceFactory);
     this.openLineageResource = new OpenLineageResource(serviceFactory, openLineageDao);
@@ -168,6 +168,7 @@ private MarquezContext(
             namespaceResource,
             sourceResource,
             datasetResource,
+            facetResource,
             columnLineageResource,
             jobResource,
             tagResource,

diff --git a/api/src/main/java/marquez/api/FacetResource.java b/api/src/main/java/marquez/api/FacetResource.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright 2018-2023 contributors to the Marquez project
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package marquez.api;
+
+import static javax.ws.rs.core.MediaType.APPLICATION_JSON;
+
+import com.codahale.metrics.annotation.ExceptionMetered;
+import com.codahale.metrics.annotation.ResponseMetered;
+import com.codahale.metrics.annotation.Timed;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.core.Response;
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+import marquez.common.models.RunId;
+import marquez.db.OpenLineageDao;
+import marquez.service.ServiceFactory;
+import marquez.service.models.Facets;
+
+
+@Slf4j
+@Path("/api/v1/facets")
+public class FacetResource extends BaseResource {
+
+  private final OpenLineageDao openLineageDao;
+
+  public FacetResource(
+      @NonNull final ServiceFactory serviceFactory, @NonNull final OpenLineageDao openLineageDao) {
+    super(serviceFactory);
+    this.openLineageDao = openLineageDao;
+  }
+
+  @Timed
+  @ResponseMetered
+  @ExceptionMetered
+  @GET
+  @Produces(APPLICATION_JSON)
+  @Path("/run/{id}")
+  public Response getRunFacets(
+    @PathParam("id") RunId runId) {
+    throwIfNotExists(runId);
+    Facets runFacets = openLineageDao.findRunFacetsByRunUuid(runId.getValue());
+    return Response.ok(runFacets).build();
+  }
+
+  @Timed
+  @ResponseMetered
+  @ExceptionMetered
+  @GET
+  @Produces(APPLICATION_JSON)
+  @Path("job/run/{id}")
+  public Response getJobFacets(
+    @PathParam("id") RunId runId) {
+    throwIfNotExists(runId);
+    Facets jobFacets = openLineageDao.findJobFacetsByRunUuid(runId.getValue());
+    return Response.ok(jobFacets).build();
+  }
+}
diff --git a/api/src/main/java/marquez/api/models/JobVersion.java b/api/src/main/java/marquez/api/models/JobVersion.java
@@ -5,7 +5,6 @@
 
 package marquez.api.models;
 
-import com.google.common.collect.ImmutableMap;
 import java.net.URL;
 import java.time.Instant;
 import java.util.List;
@@ -35,7 +34,6 @@ public final class JobVersion {
   @Getter private final Version version;
   @Getter private final NamespaceName namespace;
   @Nullable private final URL location;
-  @Getter private final ImmutableMap<String, String> context;
   @Getter private final List<DatasetId> inputs;
   @Getter private final List<DatasetId> outputs;
   @Getter @Nullable private final Run latestRun;
@@ -46,7 +44,6 @@ public JobVersion(
       @NonNull final Instant createdAt,
       @NonNull final Version version,
       @Nullable final URL location,
-      @Nullable final ImmutableMap<String, String> context,
       List<DatasetId> inputs,
       List<DatasetId> outputs,
       @Nullable Run latestRun) {
@@ -56,7 +53,6 @@ public JobVersion(
     this.version = version;
     this.namespace = id.getNamespace();
     this.location = location;
-    this.context = (context == null) ? ImmutableMap.of() : context;
     this.inputs = inputs;
     this.outputs = outputs;
     this.latestRun = latestRun;

diff --git a/api/src/main/java/marquez/common/Utils.java b/api/src/main/java/marquez/common/Utils.java
@@ -20,7 +20,6 @@
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.google.common.base.Joiner;
-import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.hash.Hashing;
 import io.dropwizard.jackson.Jackson;
@@ -236,7 +235,6 @@ public static Instant toInstant(@Nullable final String asIso) {
    * @param jobName The name of the job.
    * @param jobInputIds The input dataset IDs for the job.
    * @param jobOutputIds The output dataset IDs for the job.
-   * @param jobContext The context of the job.
    * @param jobLocation The source code location for the job.
    * @return A {@link Version} object based on the specified job meta.
    */
@@ -245,7 +243,6 @@ public static Version newJobVersionFor(
       @NonNull final JobName jobName,
       @NonNull final ImmutableSet<DatasetId> jobInputIds,
       @NonNull final ImmutableSet<DatasetId> jobOutputIds,
-      @NonNull final ImmutableMap<String, String> jobContext,
       @Nullable final String jobLocation) {
     final byte[] bytes =
         VERSION_JOINER
@@ -268,8 +265,7 @@ public static Version newJobVersionFor(
                                 jobOutputId.getNamespace().getValue(),
                                 jobOutputId.getName().getValue()))
                     .collect(joining(VERSION_DELIM)),
-                jobLocation,
-                KV_JOINER.join(jobContext))
+                jobLocation)
             .getBytes(UTF_8);
     return Version.of(UUID.nameUUIDFromBytes(bytes));
   }

diff --git a/api/src/main/java/marquez/common/models/DatasetId.java b/api/src/main/java/marquez/common/models/DatasetId.java
@@ -7,7 +7,6 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.ComparisonChain;
-import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import lombok.EqualsAndHashCode;
 import lombok.NonNull;
@@ -16,8 +15,8 @@
 
 /**
  * ID for {@code Dataset}. The class implements {@link Comparable} to ensure job versions generated
- * with {@link Utils#newJobVersionFor(NamespaceName, JobName, ImmutableSet, ImmutableSet,
- * ImmutableMap, String)} are consistent as jobs may contain inputs and outputs out of order.
+ * with {@link Utils#newJobVersionFor(NamespaceName, JobName, ImmutableSet, ImmutableSet, String)}
+ * are consistent as jobs may contain inputs and outputs out of order.
  */
 @EqualsAndHashCode
 @ToString

diff --git a/api/src/main/java/marquez/db/BaseDao.java b/api/src/main/java/marquez/db/BaseDao.java
@@ -18,9 +18,6 @@ public interface BaseDao extends SqlObject {
   @CreateSqlObject
   DatasetVersionDao createDatasetVersionDao();
 
-  @CreateSqlObject
-  JobContextDao createJobContextDao();
-
   @CreateSqlObject
   JobDao createJobDao();
 

diff --git a/api/src/main/java/marquez/db/Columns.java b/api/src/main/java/marquez/db/Columns.java
@@ -103,13 +103,9 @@ private Columns() {}
 
   /* JOB VERSION ROW COLUMNS */
   public static final String JOB_UUID = "job_uuid";
-  public static final String JOB_CONTEXT_UUID = "job_context_uuid";
   public static final String LOCATION = "location";
   public static final String LATEST_RUN_UUID = "latest_run_uuid";
 
-  /* JOB CONTEXT ROW COLUMNS */
-  public static final String CONTEXT = "context";
-
   /* RUN ROW COLUMNS */
   public static final String EXTERNAL_ID = "external_id";
   public static final String RUN_ARGS_UUID = "run_args_uuid";

diff --git a/api/src/main/java/marquez/db/JobContextDao.java b/api/src/main/java/marquez/db/JobContextDao.java
diff --git a/api/src/main/java/marquez/db/JobDao.java b/api/src/main/java/marquez/db/JobDao.java
@@ -15,15 +15,13 @@
 import java.util.Set;
 import java.util.UUID;
 import java.util.stream.Collectors;
-import marquez.common.Utils;
 import marquez.common.models.DatasetId;
 import marquez.common.models.DatasetName;
 import marquez.common.models.JobName;
 import marquez.common.models.JobType;
 import marquez.common.models.NamespaceName;
 import marquez.db.mappers.JobMapper;
 import marquez.db.mappers.JobRowMapper;
-import marquez.db.models.JobContextRow;
 import marquez.db.models.JobRow;
 import marquez.db.models.NamespaceRow;
 import marquez.service.models.Job;
@@ -58,10 +56,9 @@ SELECT EXISTS (
 
   @SqlQuery(
       """
-    SELECT j.*, jc.context, f.facets
+    SELECT j.*, f.facets
     FROM jobs_view j
     LEFT OUTER JOIN job_versions AS jv ON jv.uuid = j.current_version_uuid
-    LEFT OUTER JOIN job_contexts jc ON jc.uuid = j.current_job_context_uuid
     LEFT OUTER JOIN (
       SELECT run_uuid, JSON_AGG(e.facets) AS facets
       FROM (
@@ -130,10 +127,9 @@ default Optional<Job> findWithRun(String namespaceName, String jobName) {
 
   @SqlQuery(
       """
-    SELECT j.*, jc.context, f.facets
+    SELECT j.*, f.facets
       FROM jobs_view AS j
       LEFT OUTER JOIN job_versions AS jv ON jv.uuid = j.current_version_uuid
-      LEFT OUTER JOIN job_contexts jc ON jc.uuid = j.current_job_context_uuid
     LEFT OUTER JOIN (
       SELECT run_uuid, JSON_AGG(e.facets) AS facets
       FROM (
@@ -208,13 +204,6 @@ default JobRow upsertJobMeta(
         createNamespaceDao()
             .upsertNamespaceRow(
                 UUID.randomUUID(), createdAt, namespaceName.getValue(), DEFAULT_NAMESPACE_OWNER);
-    JobContextRow contextRow =
-        createJobContextDao()
-            .upsert(
-                UUID.randomUUID(),
-                createdAt,
-                Utils.toJson(jobMeta.getContext()),
-                Utils.checksumFor(jobMeta.getContext()));
     return upsertJob(
         UUID.randomUUID(),
         jobMeta.getType(),
@@ -223,7 +212,6 @@ default JobRow upsertJobMeta(
         namespace.getName(),
         jobName.getValue(),
         jobMeta.getDescription().orElse(null),
-        contextRow.getUuid(),
         toUrlString(jobMeta.getLocation().orElse(null)),
         symlinkTargetUuid,
         toJson(jobMeta.getInputs(), mapper));
@@ -276,7 +264,7 @@ INSERT INTO jobs_view AS j (
       :namespaceName,
       :name,
       :description,
-      :jobContextUuid,
+      null,
       :location,
       :inputs,
       :symlinkTargetId,
@@ -291,7 +279,6 @@ JobRow upsertJob(
       String namespaceName,
       String name,
       String description,
-      UUID jobContextUuid,
       String location,
       UUID symlinkTargetId,
       PGobject inputs);
@@ -328,7 +315,7 @@ INSERT INTO jobs_view AS j (
       :namespaceName,
       :name,
       :description,
-      :jobContextUuid,
+      null,
       :location,
       :inputs,
       :symlinkTargetId
@@ -344,7 +331,6 @@ JobRow upsertJob(
       String namespaceName,
       String name,
       String description,
-      UUID jobContextUuid,
       String location,
       UUID symlinkTargetId,
       PGobject inputs);