Add end point to delete field level tags.

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

@@ -190,7 +190,8 @@ public Response tag(
     throwIfNotExists(namespaceName);
     throwIfNotExists(namespaceName, datasetName);
 
-    log.info("Successfully tagged dataset '{}' with '{}'.", datasetName.getValue(), tagName);
+    log.info(
+        "Successfully tagged dataset '{}' with '{}'.", datasetName.getValue(), tagName.getValue());
 
     final Dataset dataset =
         datasetService.updateTags(
@@ -242,9 +243,9 @@ public Response tagField(
     throwIfNotExists(namespaceName, datasetName, fieldName);
     log.info(
         "Tagging field '{}' for dataset '{}' with '{}'.",
-        fieldName,
+        fieldName.getValue(),
         datasetName.getValue(),
-        tagName);
+        tagName.getValue());
     final Dataset dataset =
         datasetFieldService.updateTags(
             namespaceName.getValue(),
@@ -254,6 +255,47 @@ public Response tagField(
     return Response.ok(dataset).build();
   }
 
+  @Timed
+  @ResponseMetered
+  @ExceptionMetered
+  @DELETE
+  @Path("/{dataset}/fields/{field}/tags/{tag}")
+  @Produces(APPLICATION_JSON)
+  public Response deleteTagField(
+      @PathParam("namespace") NamespaceName namespaceName,
+      @PathParam("dataset") DatasetName datasetName,
+      @PathParam("field") FieldName fieldName,
+      @PathParam("tag") TagName tagName) {
+    throwIfNotExists(namespaceName);
+    throwIfNotExists(namespaceName, datasetName);
+    throwIfNotExists(namespaceName, datasetName, fieldName);
+    log.info(
+        "Deleting Tag '{}' from field '{}' on dataset '{}' in namepspace '{}'.",
+        tagName.getValue(),
+        fieldName.getValue(),
+        datasetName.getValue(),
+        namespaceName.getValue());
+
+    // delete tag from field
+    datasetFieldService.deleteDatasetFieldTag(
+        namespaceName.getValue(),
+        datasetName.getValue(),
+        fieldName.getValue(),
+        tagName.getValue().toUpperCase(Locale.getDefault()));
+    // delete tag from dataset_versions
+    datasetFieldService.deleteDatasetVersionFieldTag(
+        namespaceName.getValue(),
+        datasetName.getValue(),
+        fieldName.getValue(),
+        tagName.getValue().toUpperCase(Locale.getDefault()));
+    // return entire dataset
+    Dataset dataset =
+        datasetService
+            .findDatasetByName(namespaceName.getValue(), datasetName.getValue())
+            .orElseThrow(() -> new DatasetNotFoundException(datasetName));
+    return Response.ok(dataset).build();
+  }
+
   @Value
   static class DatasetVersions {
     @NonNull

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

@@ -78,11 +78,55 @@ default Dataset updateTags(
                       .collect(Collectors.toList());
               datasetVersionDao.updateFields(ver, datasetVersionDao.toPgObjectFields(fields));
             });
-
     updateTags(fieldUuid, tag.getUuid(), now);
     return createDatasetDao().findDatasetByName(namespaceName, datasetName).get();
   }
 
+  @SqlUpdate(
+      """
+      DELETE FROM dataset_fields_tag_mapping dftm
+      WHERE EXISTS
+      (
+          SELECT 1
+          FROM
+              dataset_fields df
+          JOIN datasets d ON df.dataset_uuid = d.uuid AND df.uuid = dftm.dataset_field_uuid
+          JOIN tags t ON dftm.tag_uuid = t.uuid
+          JOIN namespaces n ON d.namespace_uuid = n.uuid
+          WHERE d.name = :datasetName
+          AND t.name = :tagName
+          AND n.name = :namespaceName
+          AND df.name = :fieldName
+      );
+    """)
+  void deleteDatasetFieldTag(
+      String namespaceName, String datasetName, String fieldName, String tagName);
+
+  @SqlUpdate(
+      """
+        UPDATE dataset_versions
+        SET fields = (
+            SELECT jsonb_agg(
+                CASE
+                    WHEN elem->>'name' = :fieldName AND elem->'tags' @> jsonb_build_array(:tagName)
+                    THEN jsonb_set(elem, '{tags}', (elem->'tags') - :tagName)
+                    ELSE elem
+                END
+            )
+            FROM jsonb_array_elements(fields) AS elem
+        )
+        WHERE dataset_name = :datasetName AND namespace_name = :namespaceName
+        AND
+        EXISTS (
+            SELECT 1
+            FROM jsonb_array_elements(fields) AS elem
+            WHERE elem->>'name' = :fieldName
+            AND elem->'tags' @> jsonb_build_array(:tagName)
+        )
+    """)
+  void deleteDatasetVersionFieldTag(
+      String namespaceName, String datasetName, String fieldName, String tagName);
+
   @SqlUpdate(
       "INSERT INTO dataset_fields_tag_mapping (dataset_field_uuid, tag_uuid, tagged_at) "
           + "VALUES (:rowUuid, :tagUuid, :taggedAt)")

api/src/test/java/marquez/api/BaseResourceIntegrationTest.java

@@ -9,11 +9,14 @@
 import static marquez.common.models.CommonModelGenerator.newDatasetName;
 import static marquez.common.models.CommonModelGenerator.newDbSourceType;
 import static marquez.common.models.CommonModelGenerator.newDescription;
+import static marquez.common.models.CommonModelGenerator.newFieldName;
+import static marquez.common.models.CommonModelGenerator.newFieldType;
 import static marquez.common.models.CommonModelGenerator.newNamespaceName;
 import static marquez.common.models.CommonModelGenerator.newOwnerName;
 import static marquez.common.models.CommonModelGenerator.newSourceName;
 import static marquez.db.DbTest.POSTGRES_14;
 
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
 import io.dropwizard.testing.ConfigOverride;
 import io.dropwizard.testing.ResourceHelpers;
@@ -30,6 +33,7 @@
 import marquez.client.Utils;
 import marquez.client.models.DatasetId;
 import marquez.client.models.DbTableMeta;
+import marquez.client.models.Field;
 import marquez.client.models.NamespaceMeta;
 import marquez.client.models.SourceMeta;
 import marquez.client.models.Tag;
@@ -91,6 +95,7 @@ abstract class BaseResourceIntegrationTest {
   static String DB_TABLE_DESCRIPTION;
   static Set<String> DB_TABLE_TAGS;
   static DbTableMeta DB_TABLE_META;
+  static ImmutableList<Field> DB_TABLE_FIELDS;
 
   static DropwizardAppExtension<MarquezConfig> MARQUEZ_APP;
   static OpenLineage OL;
@@ -117,10 +122,12 @@ public static void setUpOnce() throws Exception {
     DB_TABLE_DESCRIPTION = newDescription();
     DB_TABLE_TAGS = ImmutableSet.of(PII.getName());
     DB_TABLE_CONNECTION_URL = newConnectionUrlFor(SourceType.of("POSTGRESQL"));
+    DB_TABLE_FIELDS = ImmutableList.of(newFieldWith(SENSITIVE.getName()), newField());
     DB_TABLE_META =
         DbTableMeta.builder()
             .physicalName(DB_TABLE_PHYSICAL_NAME)
             .sourceName(DB_TABLE_SOURCE_NAME)
+            .fields(DB_TABLE_FIELDS)
             .tags(DB_TABLE_TAGS)
             .description(DB_TABLE_DESCRIPTION)
             .build();
@@ -158,6 +165,18 @@ protected static DatasetId newDatasetIdWith(final String namespaceName) {
     return new DatasetId(namespaceName, newDatasetName().getValue());
   }
 
+  protected static Field newField() {
+    return newFieldWith(ImmutableSet.of());
+  }
+
+  protected static Field newFieldWith(final String tag) {
+    return newFieldWith(ImmutableSet.of(tag));
+  }
+
+  protected static Field newFieldWith(final ImmutableSet<String> tags) {
+    return new Field(newFieldName().getValue(), newFieldType(), tags, newDescription());
+  }
+
   protected void createSource(String sourceName) {
     final SourceMeta sourceMeta =
         SourceMeta.builder()

api/src/test/java/marquez/api/TagResourceIntegrationTest.java

@@ -62,4 +62,42 @@ public void testApp_testDatasetTagDelete() {
     // assert that only PII remains
     assertThat(taggedDeleteDataset.getTags()).containsExactly("PII");
   }
+
+  @Test
+  public void testApp_testDatasetTagFieldDelete() {
+    // Create Namespace
+    createNamespace(NAMESPACE_NAME);
+    // create a source
+    createSource(DB_TABLE_SOURCE_NAME);
+    // Create Dataset
+    MARQUEZ_CLIENT.createDataset(NAMESPACE_NAME, DB_TABLE_NAME, DB_TABLE_META);
+
+    // tag dataset field
+    Dataset taggedDatasetField =
+        MARQUEZ_CLIENT.tagFieldWith(
+            NAMESPACE_NAME,
+            DB_TABLE_NAME,
+            DB_TABLE_META.getFields().get(0).getName(),
+            "TESTFIELDTAG");
+    // assert the tag TESTFIELDTAG has been added to field at position 0
+    assertThat(taggedDatasetField.getFields().get(0).getTags()).contains("TESTFIELDTAG");
+    // assert a total of two tags exist on the field
+    assertThat(taggedDatasetField.getFields().get(0).getTags()).hasSize(2);
+
+    // delete the field tag TESTFIELDTAG from the dataset field at position 0
+    Dataset taggedDatasetFieldDelete =
+        MARQUEZ_CLIENT.deleteDatasetFieldTag(
+            NAMESPACE_NAME,
+            DB_TABLE_NAME,
+            DB_TABLE_META.getFields().get(0).getName(),
+            "TESTFIELDTAG");
+
+    // Test that the tag TESTDATASETTAG is deleted from the dataset field at position 0
+    assertThat(taggedDatasetFieldDelete.getFields().get(0).getTags())
+        .doesNotContain("TESTFIELDTAG");
+    // assert the number of tags on the field should be 1
+    assertThat(taggedDatasetFieldDelete.getFields().get(0).getTags()).hasSize(1);
+    // assert that only SENSITIVE remains
+    assertThat(taggedDatasetFieldDelete.getFields().get(0).getTags()).containsExactly("SENSITIVE");
+  }
 }

clients/java/src/main/java/marquez/client/MarquezClient.java

@@ -256,6 +256,16 @@ public Dataset tagFieldWith(
     return Dataset.fromJson(bodyAsJson);
   }
 
+  public Dataset deleteDatasetFieldTag(
+      @NonNull String namespaceName,
+      @NonNull String datasetName,
+      @NonNull String fieldName,
+      @NonNull String tagName) {
+    final String bodyAsJson =
+        http.delete(url.toFieldTagURL(namespaceName, datasetName, fieldName, tagName));
+    return Dataset.fromJson(bodyAsJson);
+  }
+
   /**
    * @deprecated Prefer OpenLineage, see <a
    *     href="https://openlineage.io">https://openlineage.io</a>. This method is scheduled to be

clients/java/src/test/java/marquez/client/MarquezClientTest.java

@@ -1016,6 +1016,21 @@ public void testTagField() throws Exception {
     assertThat(dataset).isEqualTo(DB_TABLE);
   }
 
+  @Test
+  public void testDeleteTagField() throws Exception {
+    final URL url =
+        buildUrlFor(
+            "/namespaces/%s/datasets/%s/fields/%s/tags/%s",
+            NAMESPACE_NAME, DB_TABLE_NAME, "field", "tag_name");
+
+    final String runAsJson = Utils.getMapper().writeValueAsString(DB_TABLE);
+    when(http.delete(url)).thenReturn(runAsJson);
+
+    final Dataset dataset =
+        client.deleteDatasetFieldTag(NAMESPACE_NAME, DB_TABLE_NAME, "field", "tag_name");
+    assertThat(dataset).isEqualTo(DB_TABLE);
+  }
+
   @Test
   public void testListTags() throws Exception {
     ImmutableSet<Tag> expectedTags =