add DELETE end point for dataset tags

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

@@ -198,6 +198,33 @@ public Response tag(
     return Response.ok(dataset).build();
   }
 
+  @Timed
+  @ResponseMetered
+  @ExceptionMetered
+  @DELETE
+  @Path("/{dataset}/tags/{tag}")
+  @Produces(APPLICATION_JSON)
+  public Response deleteDatasetTag(
+      @PathParam("namespace") NamespaceName namespaceName,
+      @PathParam("dataset") DatasetName datasetName,
+      @PathParam("tag") TagName tagName) {
+    throwIfNotExists(namespaceName);
+    throwIfNotExists(namespaceName, datasetName);
+
+    log.info(
+        "Deleted tag '{}' from dataset '{}' on namespace '{}'",
+        tagName.getValue(),
+        datasetName.getValue(),
+        namespaceName.getValue());
+    datasetService.deleteDatasetTag(
+        namespaceName.getValue(), datasetName.getValue(), tagName.getValue());
+    Dataset dataset =
+        datasetService
+            .findDatasetByName(namespaceName.getValue(), datasetName.getValue())
+            .orElseThrow(() -> new DatasetNotFoundException(datasetName));
+    return Response.ok(dataset).build();
+  }
+
   @Timed
   @ResponseMetered
   @ExceptionMetered

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

@@ -272,6 +272,33 @@ DatasetRow upsert(
       """)
   Optional<DatasetRow> delete(String namespaceName, String name);
 
+  @SqlUpdate(
+      """
+        DELETE FROM datasets_tag_mapping dtm
+        WHERE EXISTS (
+            SELECT 1
+            FROM
+              datasets d
+            JOIN
+              tags t
+            ON
+              d.uuid = dtm.dataset_uuid
+            AND
+              t.uuid = dtm.tag_uuid
+            JOIN
+              namespaces n
+            ON
+              d.namespace_uuid = n.uuid
+            WHERE
+              d.name = :datasetName
+            AND
+              t.name = :tagName
+            AND
+              n.name = :namespaceName
+        );
+      """)
+  void deleteDatasetTag(String namespaceName, String datasetName, String tagName);
+
   @Transaction
   default Dataset upsertDatasetMeta(
       NamespaceName namespaceName, DatasetName datasetName, DatasetMeta datasetMeta) {

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

@@ -5,9 +5,16 @@
 
 package marquez.api;
 
+import static marquez.common.models.CommonModelGenerator.newConnectionUrlFor;
+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.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.ImmutableSet;
 import io.dropwizard.testing.ConfigOverride;
 import io.dropwizard.testing.ResourceHelpers;
 import io.dropwizard.testing.junit5.DropwizardAppExtension;
@@ -16,11 +23,17 @@
 import io.openlineage.client.transports.HttpTransport;
 import java.net.URI;
 import java.net.URL;
+import java.util.Set;
 import marquez.MarquezApp;
 import marquez.MarquezConfig;
 import marquez.client.MarquezClient;
 import marquez.client.Utils;
+import marquez.client.models.DatasetId;
+import marquez.client.models.DbTableMeta;
+import marquez.client.models.NamespaceMeta;
+import marquez.client.models.SourceMeta;
 import marquez.client.models.Tag;
+import marquez.common.models.SourceType;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.testcontainers.containers.PostgreSQLContainer;
@@ -56,17 +69,61 @@ abstract class BaseResourceIntegrationTest {
   static final Tag PII = new Tag("PII", "Personally identifiable information");
   static final Tag SENSITIVE = new Tag("SENSITIVE", "Contains sensitive information");
 
+  // Namespace
   static String NAMESPACE_NAME;
-  static DropwizardAppExtension<MarquezConfig> MARQUEZ_APP;
+  static String NAMESPACE_DESCRIPTION;
+  static String OWNER_NAME;
+
+  // Source
+  static String SOURCE_TYPE;
+  static String SOURCE_NAME;
+  static URI CONNECTION_URL;
+  static String SOURCE_DESCRIPTION;
+
+  // Dataset
+  static String DB_TABLE_SOURCE_TYPE;
+  static String DB_TABLE_SOURCE_NAME;
+  static URI DB_TABLE_CONNECTION_URL;
+  static String DB_TABLE_SOURCE_DESCRIPTION;
+  static DatasetId DB_TABLE_ID;
+  static String DB_TABLE_NAME;
+  static String DB_TABLE_PHYSICAL_NAME;
+  static String DB_TABLE_DESCRIPTION;
+  static Set<String> DB_TABLE_TAGS;
+  static DbTableMeta DB_TABLE_META;
 
+  static DropwizardAppExtension<MarquezConfig> MARQUEZ_APP;
   static OpenLineage OL;
   static OpenLineageClient OL_CLIENT;
   static MarquezClient MARQUEZ_CLIENT;
 
   @BeforeAll
   public static void setUpOnce() throws Exception {
-    // (1) Use a randomly generated namespace.
+    // (1) Use a randomly generated namespace/dataset
     NAMESPACE_NAME = newNamespaceName().getValue();
+    NAMESPACE_DESCRIPTION = newDescription();
+    OWNER_NAME = newOwnerName().getValue();
+
+    SOURCE_TYPE = newDbSourceType().getValue();
+    SOURCE_NAME = newSourceName().getValue();
+    SOURCE_DESCRIPTION = newDescription();
+
+    DB_TABLE_SOURCE_TYPE = SourceType.of("POSTGRESQL").getValue();
+    DB_TABLE_SOURCE_NAME = SOURCE_NAME;
+    DB_TABLE_SOURCE_DESCRIPTION = newDescription();
+    DB_TABLE_ID = newDatasetIdWith(NAMESPACE_NAME);
+    DB_TABLE_NAME = DB_TABLE_ID.getName();
+    DB_TABLE_PHYSICAL_NAME = DB_TABLE_NAME;
+    DB_TABLE_DESCRIPTION = newDescription();
+    DB_TABLE_TAGS = ImmutableSet.of(PII.getName());
+    DB_TABLE_CONNECTION_URL = newConnectionUrlFor(SourceType.of("POSTGRESQL"));
+    DB_TABLE_META =
+        DbTableMeta.builder()
+            .physicalName(DB_TABLE_PHYSICAL_NAME)
+            .sourceName(DB_TABLE_SOURCE_NAME)
+            .tags(DB_TABLE_TAGS)
+            .description(DB_TABLE_DESCRIPTION)
+            .build();
 
     // (2) Configure Marquez application using test configuration and database.
     MARQUEZ_APP =
@@ -89,6 +146,28 @@ public static void setUpOnce() throws Exception {
     MARQUEZ_CLIENT = MarquezClient.builder().baseUrl(url).build();
   }
 
+  protected void createNamespace(String namespaceName) {
+    // (1) Create namespace for db table
+    final NamespaceMeta namespaceMeta =
+        NamespaceMeta.builder().ownerName(OWNER_NAME).description(NAMESPACE_DESCRIPTION).build();
+
+    MARQUEZ_CLIENT.createNamespace(namespaceName, namespaceMeta);
+  }
+
+  protected static DatasetId newDatasetIdWith(final String namespaceName) {
+    return new DatasetId(namespaceName, newDatasetName().getValue());
+  }
+
+  protected void createSource(String sourceName) {
+    final SourceMeta sourceMeta =
+        SourceMeta.builder()
+            .type(DB_TABLE_SOURCE_TYPE)
+            .connectionUrl(DB_TABLE_CONNECTION_URL)
+            .description(DB_TABLE_SOURCE_DESCRIPTION)
+            .build();
+    MARQUEZ_CLIENT.createSource(sourceName, sourceMeta);
+  }
+
   @AfterAll
   public static void cleanUp() {
     MARQUEZ_APP.after();

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

@@ -10,10 +10,12 @@
 import static org.assertj.core.api.Assertions.assertThat;
 
 import java.util.Set;
+import marquez.client.models.Dataset;
 import marquez.client.models.Tag;
 import org.junit.jupiter.api.Test;
 
 public class TagResourceIntegrationTest extends BaseResourceIntegrationTest {
+
   @Test
   public void testApp_createTag() {
     // (1) List tags.
@@ -36,4 +38,28 @@ public void testApp_listTags() {
     // (2) Ensure tags 'PII', 'SENSITIVE' defined in 'config.test.yml' are present.
     assertThat(tags).contains(PII, SENSITIVE);
   }
+
+  @Test
+  public void testApp_testDatasetTagDelete() {
+    // 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 with TESTDATASETTAG tag
+    Dataset taggedDataset =
+        MARQUEZ_CLIENT.tagDatasetWith(NAMESPACE_NAME, DB_TABLE_NAME, "TESTDATASETTAG");
+    assertThat(taggedDataset.getTags()).contains("TESTDATASETTAG");
+
+    // Test that the tag TESTDATASETTAG is deleted from the dataset
+    Dataset taggedDeleteDataset =
+        MARQUEZ_CLIENT.deleteDatasetTag(NAMESPACE_NAME, DB_TABLE_NAME, "TESTDATASETTAG");
+    assertThat(taggedDeleteDataset.getTags()).doesNotContain("TESTDATASETTAG");
+    // assert the number of tags should be 1
+    assertThat(taggedDeleteDataset.getTags()).hasSize(1);
+    // assert that only PII remains
+    assertThat(taggedDeleteDataset.getTags()).containsExactly("PII");
+  }
 }

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

@@ -240,6 +240,12 @@ public Dataset tagDatasetWith(
     return Dataset.fromJson(bodyAsJson);
   }
 
+  public Dataset deleteDatasetTag(
+      @NonNull String namespaceName, @NonNull String datasetName, @NonNull String tagName) {
+    final String bodyAsJson = http.delete(url.toDatasetTagUrl(namespaceName, datasetName, tagName));
+    return Dataset.fromJson(bodyAsJson);
+  }
+
   public Dataset tagFieldWith(
       @NonNull String namespaceName,
       @NonNull String datasetName,

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

@@ -989,6 +989,19 @@ public void testTagDataset() throws Exception {
     assertThat(dataset).isEqualTo(DB_TABLE);
   }
 
+  @Test
+  public void testDeleteDatasetTag() throws Exception {
+    final URL url =
+        buildUrlFor(
+            "/namespaces/%s/datasets/%s/tags/%s", NAMESPACE_NAME, DB_TABLE_NAME, "tag_name");
+
+    final String runAsJson = Utils.getMapper().writeValueAsString(DB_TABLE);
+    when(http.delete(url)).thenReturn(runAsJson);
+
+    final Dataset dataset = client.deleteDatasetTag(NAMESPACE_NAME, DB_TABLE_NAME, "tag_name");
+    assertThat(dataset).isEqualTo(DB_TABLE);
+  }
+
   @Test
   public void testTagField() throws Exception {
     final URL url =