Remove job context.

api/src/main/java/marquez/MarquezContext.java

@@ -28,14 +28,15 @@
 import marquez.db.DatasetDao;
 import marquez.db.DatasetFieldDao;
 import marquez.db.DatasetVersionDao;
-import marquez.db.JobContextDao;
 import marquez.db.JobDao;
+import marquez.db.JobFacetsDao;
 import marquez.db.JobVersionDao;
 import marquez.db.LineageDao;
 import marquez.db.NamespaceDao;
 import marquez.db.OpenLineageDao;
 import marquez.db.RunArgsDao;
 import marquez.db.RunDao;
+import marquez.db.RunFacetsDao;
 import marquez.db.RunStateDao;
 import marquez.db.SearchDao;
 import marquez.db.SourceDao;
@@ -67,9 +68,10 @@ 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 JobFacetsDao jobFacetsDao;
   @Getter private final RunDao runDao;
   @Getter private final RunArgsDao runArgsDao;
+  @Getter private final RunFacetsDao runFacetsDao;
   @Getter private final RunStateDao runStateDao;
   @Getter private final TagDao tagDao;
   @Getter private final OpenLineageDao openLineageDao;
@@ -116,9 +118,10 @@ 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.jobFacetsDao = jdbi.onDemand(JobFacetsDao.class);
     this.runDao = jdbi.onDemand(RunDao.class);
     this.runArgsDao = jdbi.onDemand(RunArgsDao.class);
+    this.runFacetsDao = jdbi.onDemand(RunFacetsDao.class);
     this.runStateDao = jdbi.onDemand(RunStateDao.class);
     this.tagDao = jdbi.onDemand(TagDao.class);
     this.openLineageDao = jdbi.onDemand(OpenLineageDao.class);
@@ -158,7 +161,7 @@ private MarquezContext(
     this.sourceResource = new SourceResource(serviceFactory);
     this.datasetResource = new DatasetResource(serviceFactory);
     this.columnLineageResource = new ColumnLineageResource(serviceFactory);
-    this.jobResource = new JobResource(serviceFactory, jobVersionDao);
+    this.jobResource = new JobResource(serviceFactory, jobVersionDao, jobFacetsDao, runFacetsDao);
     this.tagResource = new TagResource(serviceFactory);
     this.openLineageResource = new OpenLineageResource(serviceFactory, openLineageDao);
     this.searchResource = new SearchResource(searchDao);

api/src/main/java/marquez/api/JobResource.java

@@ -15,6 +15,7 @@
 import java.util.List;
 import javax.validation.Valid;
 import javax.validation.constraints.Min;
+import javax.validation.constraints.NotNull;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.DELETE;
 import javax.ws.rs.DefaultValue;
@@ -37,11 +38,14 @@
 import marquez.api.exceptions.JobVersionNotFoundException;
 import marquez.api.models.JobVersion;
 import marquez.api.models.ResultsPage;
+import marquez.common.models.FacetType;
 import marquez.common.models.JobName;
 import marquez.common.models.NamespaceName;
 import marquez.common.models.RunId;
 import marquez.common.models.Version;
+import marquez.db.JobFacetsDao;
 import marquez.db.JobVersionDao;
+import marquez.db.RunFacetsDao;
 import marquez.db.models.JobRow;
 import marquez.service.ServiceFactory;
 import marquez.service.models.Job;
@@ -52,11 +56,18 @@
 @Path("/api/v1")
 public class JobResource extends BaseResource {
   private final JobVersionDao jobVersionDao;
+  private final JobFacetsDao jobFacetsDao;
+  private final RunFacetsDao runFacetsDao;
 
   public JobResource(
-      @NonNull final ServiceFactory serviceFactory, @NonNull final JobVersionDao jobVersionDao) {
+      @NonNull final ServiceFactory serviceFactory,
+      @NonNull final JobVersionDao jobVersionDao,
+      @NonNull JobFacetsDao jobFacetsDao,
+      @NonNull RunFacetsDao runFacetsDao) {
     super(serviceFactory);
     this.jobVersionDao = jobVersionDao;
+    this.jobFacetsDao = jobFacetsDao;
+    this.runFacetsDao = runFacetsDao;
   }
 
   /**
@@ -236,6 +247,33 @@ public RunResource runResourceRoot(@PathParam("id") RunId runId) {
     return new RunResource(runId, runService);
   }
 
+  @Timed
+  @ResponseMetered
+  @ExceptionMetered
+  @GET
+  @Produces(APPLICATION_JSON)
+  @Path("/jobs/runs/{id}/facets")
+  public Response getRunFacets(
+      @PathParam("id") RunId runId, @QueryParam("type") @NotNull FacetType type) {
+    throwIfNotExists(runId);
+    Object facets = null;
+    switch (type) {
+      case JOB:
+        facets = jobFacetsDao.findJobFacetsByRunUuid(runId.getValue());
+        break;
+      case RUN:
+        facets = runFacetsDao.findRunFacetsByRunUuid(runId.getValue());
+        break;
+      case DATASET:
+        // for future case if there's a need to add dataset facets to the endpoint
+        break;
+      default:
+        break;
+    }
+
+    return Response.ok(facets).build();
+  }
+
   @Value
   static class JobVersions {
     @NonNull

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;

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));
   }

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

api/src/main/java/marquez/common/models/FacetType.java

@@ -0,0 +1,12 @@
+/*
+ * Copyright 2018-2023 contributors to the Marquez project
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package marquez.common.models;
+
+public enum FacetType {
+  RUN,
+  JOB,
+  DATASET;
+}

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();
 

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";

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.facet) 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.facet) 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);