Add job tagging to API

api/src/main/java/marquez/api/DatasetResource.java

@@ -242,7 +242,7 @@ public Response tagField(
     throwIfNotExists(namespaceName, datasetName);
     throwIfNotExists(namespaceName, datasetName, fieldName);
     log.info(
-        "Tagging field '{}' for dataset '{}' with '{}'.",
+        "Tagging field '{}' on dataset '{}' with '{}'.",
         fieldName.getValue(),
         datasetName.getValue(),
         tagName.getValue());
@@ -270,7 +270,7 @@ public Response deleteTagField(
     throwIfNotExists(namespaceName, datasetName);
     throwIfNotExists(namespaceName, datasetName, fieldName);
     log.info(
-        "Deleting Tag '{}' from field '{}' on dataset '{}' in namepspace '{}'.",
+        "Deleting Tag '{}' from field '{}' on dataset '{}' in namespace '{}'.",
         tagName.getValue(),
         fieldName.getValue(),
         datasetName.getValue(),

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

@@ -42,6 +42,7 @@
 import marquez.common.models.JobName;
 import marquez.common.models.NamespaceName;
 import marquez.common.models.RunId;
+import marquez.common.models.TagName;
 import marquez.common.models.Version;
 import marquez.db.JobFacetsDao;
 import marquez.db.JobVersionDao;
@@ -273,6 +274,47 @@ public Response getRunFacets(
     return Response.ok(facets).build();
   }
 
+  @Timed
+  @ResponseMetered
+  @ExceptionMetered
+  @POST
+  @Path("/namespaces/{namespace}/jobs/{job}/tags/{tag}")
+  @Produces(APPLICATION_JSON)
+  public Response updatetag(
+      @PathParam("namespace") NamespaceName namespaceName,
+      @PathParam("job") JobName jobName,
+      @PathParam("tag") TagName tagName) {
+    throwIfNotExists(namespaceName);
+    throwIfNotExists(namespaceName, jobName);
+
+    jobService.updateJobTags(namespaceName.getValue(), jobName.getValue(), tagName.getValue());
+    Job job =
+        jobService
+            .findJobByName(namespaceName.getValue(), jobName.getValue())
+            .orElseThrow(() -> new JobNotFoundException(jobName));
+    return Response.ok(job).build();
+  }
+
+  @ResponseMetered
+  @ExceptionMetered
+  @DELETE
+  @Path("/namespaces/{namespace}/jobs/{job}/tags/{tag}")
+  @Produces(APPLICATION_JSON)
+  public Response deletetag(
+      @PathParam("namespace") NamespaceName namespaceName,
+      @PathParam("job") JobName jobName,
+      @PathParam("tag") TagName tagName) {
+    throwIfNotExists(namespaceName);
+    throwIfNotExists(namespaceName, jobName);
+
+    jobService.deleteJobTags(namespaceName.getValue(), jobName.getValue(), tagName.getValue());
+    Job job =
+        jobService
+            .findJobByName(namespaceName.getValue(), jobName.getValue())
+            .orElseThrow(() -> new JobNotFoundException(jobName));
+    return Response.ok(job).build();
+  }
+
   @Value
   static class JobVersions {
     @NonNull

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

@@ -61,14 +61,53 @@ SELECT EXISTS (
   @SqlQuery(
       """
     WITH job_versions_facets AS (
-        SELECT job_version_uuid, JSON_AGG(facet) as facets
-        FROM job_facets
-        GROUP BY job_version_uuid
+        SELECT
+            job_version_uuid
+        ,   JSON_AGG(facet) as facets
+        FROM
+            job_facets
+        GROUP BY
+            job_version_uuid
+    ),
+    job_tags as (
+    SELECT
+        j.uuid
+    ,   ARRAY_AGG(t.name) as tags
+    FROM
+        jobs j
+    INNER JOIN
+        jobs_tag_mapping jtm
+    ON
+        jtm.job_uuid = j.uuid
+    AND
+        j.simple_name = :jobName
+    AND
+        j.namespace_name = :namespaceName
+    INNER JOIN
+        tags t
+    ON
+        jtm.tag_uuid = t.uuid
+    GROUP BY
+      j.uuid
     )
-    SELECT j.*, facets
-    FROM jobs_view j
-    LEFT OUTER JOIN job_versions_facets f ON j.current_version_uuid = f.job_version_uuid
-    WHERE j.namespace_name=:namespaceName AND (j.name=:jobName OR :jobName = ANY(j.aliases))
+    SELECT
+        j.*
+    ,   facets
+    ,   jt.tags as tags
+    FROM
+        jobs_view j
+    LEFT OUTER JOIN
+        job_versions_facets f
+    ON
+        j.current_version_uuid = f.job_version_uuid
+    LEFT OUTER JOIN
+        job_tags jt
+    ON
+        j.uuid = jt.uuid
+    WHERE
+        j.namespace_name = :namespaceName
+    AND
+        (j.name = :jobName OR :jobName = ANY(j.aliases))
   """)
   Optional<Job> findJobByName(String namespaceName, String jobName);
 
@@ -169,16 +208,38 @@ facets_temp AS (
           lineage_event_time ASC
         ) e
     GROUP BY e.run_uuid
-    )
+    ),
+    job_tags as (
+      SELECT
+          j.uuid
+      ,   ARRAY_AGG(t.name) as tags
+      FROM
+          jobs j
+      INNER JOIN
+          jobs_tag_mapping jtm
+      ON
+          jtm.job_uuid = j.uuid
+      AND
+          j.namespace_name = :namespaceName
+      INNER JOIN
+          tags t
+      ON
+          jtm.tag_uuid = t.uuid
+      GROUP BY
+      j.uuid
+  )
     SELECT
       j.*,
-      f.facets
+      f.facets,
+      COALESCE(jt.tags, ARRAY[]::VARCHAR[]) AS tags
     FROM
       jobs_view_page AS j
     LEFT OUTER JOIN job_versions_temp AS jv
       ON jv.uuid = j.current_version_uuid
     LEFT OUTER JOIN facets_temp AS f
       ON f.run_uuid = jv.latest_run_uuid
+    LEFT OUTER JOIN job_tags jt
+      ON j.uuid  = jt.uuid
     ORDER BY
         j.name
   """)
@@ -386,4 +447,64 @@ JobRow upsertJob(
       String location,
       UUID symlinkTargetId,
       PGobject inputs);
+
+  @SqlUpdate(
+      """
+    WITH new_tag AS (
+    INSERT INTO tags (uuid, created_at, updated_at, name, description)
+    SELECT
+      gen_random_uuid(),
+      NOW(),
+      NOW(),
+      :tagName,
+      NULL
+    WHERE
+        NOT EXISTS (SELECT 1 FROM tags WHERE name = :tagName)
+    RETURNING uuid
+    ),
+    existing_tag AS (
+        SELECT uuid FROM tags WHERE name = :tagName
+    ),
+    job AS (
+      SELECT
+        uuid
+      FROM
+        jobs
+      WHERE
+        simple_name = :jobName
+      and
+        namespace_name = :namespaceName
+    )
+    INSERT INTO jobs_tag_mapping (job_uuid, tag_uuid, tagged_at)
+    SELECT
+        (SELECT uuid FROM job)
+    ,   COALESCE((SELECT uuid FROM new_tag), (SELECT uuid FROM existing_tag))
+    ,   NOW()
+    ON CONFLICT DO NOTHING
+    ;
+    """)
+  void updateJobTags(String namespaceName, String jobName, String tagName);
+
+  @SqlUpdate(
+      """
+      DELETE FROM jobs_tag_mapping jtm
+      WHERE EXISTS (
+            SELECT 1
+            FROM
+                jobs j
+            JOIN
+                tags t
+            ON
+                j.uuid = jtm.job_uuid
+            AND
+                t.uuid = jtm.tag_uuid
+            WHERE
+                t.name = :tagName
+            AND
+                j.simple_name = :jobName
+            AND
+                j.namespace_name = :namespaceName
+            );
+      """)
+  void deleteJobTags(String namespaceName, String jobName, String tagName);
 }

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

@@ -5,6 +5,7 @@
 
 package marquez.db.mappers;
 
+import static marquez.db.Columns.stringArrayOrThrow;
 import static marquez.db.Columns.stringOrNull;
 import static marquez.db.Columns.stringOrThrow;
 import static marquez.db.Columns.timestampOrThrow;
@@ -18,9 +19,11 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableList.Builder;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -32,6 +35,7 @@
 import marquez.common.models.JobName;
 import marquez.common.models.JobType;
 import marquez.common.models.NamespaceName;
+import marquez.common.models.TagName;
 import marquez.db.Columns;
 import marquez.service.models.Job;
 import org.jdbi.v3.core.mapper.RowMapper;
@@ -68,7 +72,8 @@ public Job map(@NonNull ResultSet results, @NonNull StatementContext context)
             null,
             facetsOrNull,
             uuidOrNull(results, Columns.CURRENT_VERSION_UUID),
-            getLabels(facetsOrNull));
+            getLabels(facetsOrNull),
+            toTags(results, "tags"));
     return job;
   }
 
@@ -106,6 +111,15 @@ private ImmutableList<String> getLabels(ImmutableMap<String, Object> facetsOrNul
     return builder.build();
   }
 
+  public static ImmutableSet<TagName> toTags(@NonNull ResultSet results, String column)
+      throws SQLException {
+    if (results.getObject(column) == null) {
+      return null;
+    }
+    List<String> arr = stringArrayOrThrow(results, column);
+    return arr.stream().map(TagName::of).collect(ImmutableSet.toImmutableSet());
+  }
+
   private String getJobTypeFacetField(ImmutableMap<String, Object> facetsOrNull, String field) {
     return Optional.ofNullable(facetsOrNull.get(JOB_TYPE_FACET_NAME))
         .filter(o -> o instanceof Map)

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

@@ -7,6 +7,7 @@
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
 import java.net.URL;
 import java.time.Instant;
 import java.util.Optional;
@@ -23,6 +24,7 @@
 import marquez.common.models.JobName;
 import marquez.common.models.JobType;
 import marquez.common.models.NamespaceName;
+import marquez.common.models.TagName;
 
 @EqualsAndHashCode
 @ToString
@@ -43,6 +45,7 @@ public final class Job {
   @Getter private final ImmutableMap<String, Object> facets;
   @Nullable private UUID currentVersion;
   @Getter @Nullable private ImmutableList<String> labels;
+  @Getter @Nullable private final ImmutableSet<TagName> tags;
 
   public Job(
       @NonNull final JobId id,
@@ -59,7 +62,8 @@ public Job(
       @Nullable final Run latestRun,
       @Nullable final ImmutableMap<String, Object> facets,
       @Nullable UUID currentVersion,
-      @Nullable ImmutableList<String> labels) {
+      @Nullable ImmutableList<String> labels,
+      @Nullable final ImmutableSet<TagName> tags) {
     this.id = id;
     this.type = type;
     this.name = name;
@@ -76,6 +80,7 @@ public Job(
     this.facets = (facets == null) ? ImmutableMap.of() : facets;
     this.currentVersion = currentVersion;
     this.labels = (labels == null) ? ImmutableList.of() : labels;
+    this.tags = (tags == null) ? ImmutableSet.of() : tags;
   }
 
   public Optional<URL> getLocation() {

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

@@ -16,6 +16,7 @@
 import marquez.common.models.DatasetId;
 import marquez.common.models.JobType;
 import marquez.common.models.RunId;
+import marquez.common.models.TagName;
 
 @EqualsAndHashCode
 @ToString
@@ -26,20 +27,23 @@ public final class JobMeta {
   @Nullable private final URL location;
   @Nullable private final String description;
   @Nullable private final RunId runId;
+  @Getter private final ImmutableSet<TagName> tags;
 
   public JobMeta(
       @NonNull final JobType type,
       @NonNull final ImmutableSet<DatasetId> inputs,
       @NonNull final ImmutableSet<DatasetId> outputs,
       @Nullable final URL location,
       @Nullable final String description,
-      @Nullable final RunId runId) {
+      @Nullable final RunId runId,
+      @Nullable final ImmutableSet<TagName> tags) {
     this.type = type;
     this.inputs = inputs;
     this.outputs = outputs;
     this.location = location;
     this.description = description;
     this.runId = runId;
+    this.tags = tags;
   }
 
   public Optional<URL> getLocation() {

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

@@ -0,0 +1,9 @@
+/* SPDX-License-Identifier: Apache-2.0 */
+CREATE TABLE jobs_tag_mapping (
+  job_uuid UUID REFERENCES jobs(uuid),
+  tag_uuid     UUID REFERENCES tags(uuid),
+  tagged_at    TIMESTAMP NOT NULL,
+  PRIMARY KEY (tag_uuid, job_uuid)
+);
+
+