version history: deduplicate the versions displayed

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

@@ -278,13 +278,19 @@ LEFT JOIN (
       	WHERE dv.namespace_name = :namespaceName
             AND dv.dataset_name = :datasetName
       	LIMIT :limit OFFSET :offset
+        ),
+        dataset_symlinks_names as (
+          SELECT DISTINCT dataset_uuid, name
+          FROM dataset_symlinks
+          WHERE NOT is_primary
         )
         SELECT
 	        type, name, physical_name, namespace_name, source_name, description, lifecycle_state,
             created_at, version, dataset_schema_version_uuid, fields, createdByRunUuid, schema_location,
             tags, dataset_version_uuid,
 	        JSONB_AGG(facets ORDER BY lineage_event_time ASC) AS facets
         FROM dataset_info
+        WHERE name NOT IN (SELECT name FROM dataset_symlinks_names)
         GROUP BY type, name, physical_name, namespace_name, source_name, description, lifecycle_state,
             created_at, version, dataset_schema_version_uuid, fields, createdByRunUuid, schema_location,
             tags, dataset_version_uuid

api/src/test/java/marquez/DatasetIntegrationTest.java

@@ -7,6 +7,8 @@
 
 import static marquez.db.ColumnLineageTestUtils.getDatasetA;
 import static marquez.db.ColumnLineageTestUtils.getDatasetB;
+import static marquez.db.LineageTestUtils.PRODUCER_URL;
+import static marquez.db.LineageTestUtils.SCHEMA_URL;
 import static org.assertj.core.api.Assertions.assertThat;
 
 import com.fasterxml.jackson.core.type.TypeReference;
@@ -574,4 +576,57 @@ public void testApp_doesNotShowDeletedDatasetAfterUndeleteNamespace() throws IOE
     jobs = client.listJobs(namespaceName);
     assertThat(jobs).hasSize(1);
   }
+
+  @Test
+  public void testApp_getTableVersionsWithSymlinks() {
+    client.createDataset(NAMESPACE_NAME, DB_TABLE_NAME, DB_TABLE_META);
+
+    ImmutableMap<String, Object> outputFacets =
+        ImmutableMap.of("outputFacetKey", "outputFacetValue");
+    ImmutableMap<String, Object> inputFacets = ImmutableMap.of("inputFacetKey", "inputFacetValue");
+
+    final LineageEvent.DatasetFacets datasetFacets =
+        LineageTestUtils.newDatasetFacet(
+            outputFacets,
+            LineageEvent.SchemaField.builder()
+                .name("firstname")
+                .type("string")
+                .description("the first name")
+                .build());
+    datasetFacets
+        .getDocumentation()
+        .setDescription(DB_TABLE_META.getDescription().orElse("the dataset documentation"));
+    datasetFacets.setSymlinks(
+        new LineageEvent.DatasetSymlinkFacet(
+            PRODUCER_URL,
+            SCHEMA_URL,
+            Collections.singletonList(
+                new LineageEvent.SymlinkIdentifier("symlinkNamespace", "symlinkName", "type"))));
+    final LineageEvent lineageEvent =
+        LineageEvent.builder()
+            .producer("testApp_getTableVersionsWithSymlinks")
+            .eventType("COMPLETE")
+            .run(
+                new LineageEvent.Run(
+                    UUID.randomUUID().toString(), LineageEvent.RunFacet.builder().build()))
+            .job(LineageEvent.Job.builder().namespace(NAMESPACE_NAME).name(JOB_NAME).build())
+            .eventTime(ZonedDateTime.now())
+            .inputs(
+                Collections.singletonList(
+                    LineageEvent.Dataset.builder()
+                        .namespace(NAMESPACE_NAME)
+                        .name(DB_TABLE_NAME)
+                        .facets(datasetFacets)
+                        .build()))
+            .outputs(Collections.emptyList())
+            .build();
+    final CompletableFuture<Integer> resp = sendEvent(lineageEvent);
+    assertThat(resp.join()).isEqualTo(201);
+    List<DatasetVersion> versions = client.listDatasetVersions(NAMESPACE_NAME, DB_TABLE_NAME);
+
+    versions.forEach(
+        datasetVersion -> {
+          assertThat(datasetVersion.getName()).isNotEqualTo("symlinkName");
+        });
+  }
 }