Fix dataset_fields duplication for NULL type causing column_lineage explosion

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

@@ -251,6 +251,24 @@ WHERE CAST((:namespaceName, :datasetName) AS DATASET_NAME) = ANY(d.dataset_symli
           """)
   List<InputFieldData> findInputFieldsDataAssociatedWithRun(UUID runUuid);
 
+  /**
+   * Upserts a {@code dataset_fields} row.
+   *
+   * <p>The {@code type} column participates in the {@code (dataset_uuid, name, type)} unique
+   * constraint used as the {@code ON CONFLICT} target below. In standard SQL, {@code NULL} is
+   * never considered equal to another {@code NULL} for uniqueness purposes, so a field with an
+   * unknown/omitted {@code type} (a legitimate, common case for OpenLineage events that don't
+   * report column types) would never match an existing row and a brand new {@code dataset_fields}
+   * row - with a new {@code uuid} - would be inserted on every single upsert. Since each distinct
+   * {@code dataset_fields} row independently accumulates its own {@code column_lineage} edges,
+   * this caused an unbounded, combinatorial explosion of duplicate rows in both {@code
+   * dataset_fields} and {@code column_lineage} for any field with a null type (see #3083).
+   *
+   * <p>To fix this without weakening the constraint, a null {@code type} is normalized to the
+   * literal sentinel value {@code 'UNKNOWN'} before insertion, so that repeated upserts of a
+   * field with no known type reliably collide with the previously-inserted row (which also stored
+   * {@code 'UNKNOWN'}) and are treated as an update instead of a fresh insert.
+   */
   @SqlQuery(
       "INSERT INTO dataset_fields ("
           + "uuid, "
@@ -262,7 +280,7 @@ WHERE CAST((:namespaceName, :datasetName) AS DATASET_NAME) = ANY(d.dataset_symli
           + "description"
           + ") VALUES ("
           + ":uuid, "
-          + ":type, "
+          + "COALESCE(:type, 'UNKNOWN'), "
           + ":now, "
           + ":now, "
           + ":datasetUuid, "

api/src/test/java/marquez/MarquezAppIntegrationTest.java

@@ -191,6 +191,11 @@ public void testDatasetWithUnknownFieldType() {
     // (3) Create db table with invalid field type
     final Field field0 = Field.builder().name("field0").type(newFieldType()).build();
     final Field field1 = Field.builder().name("field1").type(null).build();
+    // A null/unknown field type is normalized to "UNKNOWN" on write so that repeated upserts of
+    // the same field reliably collide on the (dataset_uuid, name, type) unique constraint instead
+    // of creating a new dataset_fields row (and therefore a new column_lineage edge) every time
+    // (see #3083).
+    final Field expectedField1 = Field.builder().name("field1").type("UNKNOWN").build();
     final DatasetName datasetName = newDatasetName();
     final DbTableMeta dbTableMeta =
         DbTableMeta.builder()
@@ -200,7 +205,45 @@ public void testDatasetWithUnknownFieldType() {
             .build();
     final Dataset dataset =
         client.createDataset(NAMESPACE_NAME, datasetName.getValue(), dbTableMeta);
-    assertThat(dataset.getFields()).containsExactly(field0, field1);
+    assertThat(dataset.getFields()).containsExactly(field0, expectedField1);
+  }
+
+  @Test
+  public void testDatasetWithUnknownFieldType_repeatedUpsertsDoNotDuplicateField() {
+    // Regression test for https://github.com/MarquezProject/marquez/issues/3083: upserting the
+    // same dataset with a field that has a null/unknown type multiple times must not create
+    // multiple dataset_fields rows for that field.
+    final NamespaceMeta namespaceMeta =
+        NamespaceMeta.builder().ownerName(OWNER_NAME).description(NAMESPACE_DESCRIPTION).build();
+    client.createNamespace(NAMESPACE_NAME, namespaceMeta);
+
+    final SourceMeta sourceMeta =
+        SourceMeta.builder()
+            .type(STREAM_SOURCE_TYPE)
+            .connectionUrl(STREAM_CONNECTION_URL)
+            .description(STREAM_SOURCE_DESCRIPTION)
+            .build();
+    client.createSource(DB_TABLE_SOURCE_NAME, sourceMeta);
+
+    final Field unknownTypeField = Field.builder().name("modified_at").type(null).build();
+    final DatasetName datasetName = newDatasetName();
+    final DbTableMeta dbTableMeta =
+        DbTableMeta.builder()
+            .physicalName(datasetName.getValue())
+            .sourceName(DB_TABLE_SOURCE_NAME)
+            .fields(ImmutableList.of(unknownTypeField))
+            .build();
+
+    // Emit the same dataset (and therefore the same null-type field) multiple times, as described
+    // in the issue's reproduction steps.
+    Dataset dataset = null;
+    for (int i = 0; i < 5; i++) {
+      dataset = client.createDataset(NAMESPACE_NAME, datasetName.getValue(), dbTableMeta);
+    }
+
+    // Only a single field should exist for 'modified_at', not five duplicates.
+    assertThat(dataset.getFields()).hasSize(1);
+    assertThat(dataset.getFields().get(0).getName()).isEqualTo("modified_at");
   }
 
   @Test

api/src/test/java/marquez/db/DatasetFieldDaoTest.java

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2018-2023 contributors to the Marquez project
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package marquez.db;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Instant;
+import java.util.UUID;
+import marquez.db.models.DatasetFieldRow;
+import marquez.db.models.DatasetRow;
+import marquez.jdbi.MarquezJdbiExternalPostgresExtension;
+import marquez.service.models.Dataset;
+import org.jdbi.v3.core.Jdbi;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Regression tests for https://github.com/MarquezProject/marquez/issues/3083: a {@code
+ * dataset_fields} row whose {@code type} is {@code null} must not be duplicated on repeated
+ * upserts, since duplicate {@code dataset_fields} rows each independently accumulate their own
+ * {@code column_lineage} edges, causing a combinatorial explosion.
+ */
+@ExtendWith(MarquezJdbiExternalPostgresExtension.class)
+class DatasetFieldDaoTest {
+
+  private static DatasetFieldDao datasetFieldDao;
+  private static DatasetDao datasetDao;
+  private static Jdbi jdbi;
+
+  @BeforeAll
+  public static void setUpOnce(Jdbi jdbi) {
+    DatasetFieldDaoTest.jdbi = jdbi;
+    datasetFieldDao = jdbi.onDemand(DatasetFieldDao.class);
+    datasetDao = jdbi.onDemand(DatasetDao.class);
+  }
+
+  @AfterEach
+  public void tearDown(Jdbi jdbi) {
+    marquez.api.JdbiUtils.cleanDatabase(jdbi);
+  }
+
+  @Test
+  public void testUpsert_withNullType_doesNotDuplicateOnRepeatedUpserts() {
+    final Dataset dataset = DbTestUtils.newDataset(jdbi);
+    final DatasetRow datasetRow =
+        datasetDao.getUuid(dataset.getNamespace().getValue(), dataset.getName().getValue()).get();
+
+    final String fieldName = "modified_at";
+    final Instant now = Instant.now();
+
+    // Simulate the same field, with an unknown/null type, being reported across multiple
+    // OpenLineage events for the same dataset - each upsert uses a freshly-generated candidate
+    // UUID, exactly as OpenLineageDao.upsertFields(...) does for every incoming event.
+    DatasetFieldRow first =
+        datasetFieldDao.upsert(
+            UUID.randomUUID(), now, fieldName, null, null, datasetRow.getUuid());
+    DatasetFieldRow second =
+        datasetFieldDao.upsert(
+            UUID.randomUUID(), now, fieldName, null, null, datasetRow.getUuid());
+    DatasetFieldRow third =
+        datasetFieldDao.upsert(
+            UUID.randomUUID(), now, fieldName, null, null, datasetRow.getUuid());
+
+    // All three upserts must resolve to the *same* underlying row.
+    assertThat(second.getUuid()).isEqualTo(first.getUuid());
+    assertThat(third.getUuid()).isEqualTo(first.getUuid());
+
+    // And only a single dataset_fields row should exist for this field.
+    Integer rowCount =
+        jdbi.withHandle(
+            handle ->
+                handle
+                    .createQuery(
+                        "SELECT count(*) FROM dataset_fields WHERE dataset_uuid = :datasetUuid AND name = :name")
+                    .bind("datasetUuid", datasetRow.getUuid())
+                    .bind("name", fieldName)
+                    .mapTo(Integer.class)
+                    .one());
+    assertThat(rowCount).isEqualTo(1);
+  }
+
+  @Test
+  public void testUpsert_withNullType_isStoredAsUnknownSentinel() {
+    final Dataset dataset = DbTestUtils.newDataset(jdbi);
+    final DatasetRow datasetRow =
+        datasetDao.getUuid(dataset.getNamespace().getValue(), dataset.getName().getValue()).get();
+
+    DatasetFieldRow row =
+        datasetFieldDao.upsert(
+            UUID.randomUUID(), Instant.now(), "unknown_type_field", null, null, datasetRow.getUuid());
+
+    assertThat(row.getType()).isEqualTo("UNKNOWN");
+  }
+
+  @Test
+  public void testUpsert_withRealType_isUnaffected() {
+    final Dataset dataset = DbTestUtils.newDataset(jdbi);
+    final DatasetRow datasetRow =
+        datasetDao.getUuid(dataset.getNamespace().getValue(), dataset.getName().getValue()).get();
+
+    final Instant now = Instant.now();
+    DatasetFieldRow first =
+        datasetFieldDao.upsert(
+            UUID.randomUUID(), now, "typed_field", "VARCHAR", null, datasetRow.getUuid());
+    DatasetFieldRow second =
+        datasetFieldDao.upsert(
+            UUID.randomUUID(), now, "typed_field", "VARCHAR", null, datasetRow.getUuid());
+
+    assertThat(first.getType()).isEqualTo("VARCHAR");
+    assertThat(second.getUuid()).isEqualTo(first.getUuid());
+  }
+}