From 367268366ca2bc51c782bdcabe0134b32c2363b4 Mon Sep 17 00:00:00 2001 From: Mohit Kalra Date: Tue, 21 Jul 2026 18:18:19 -0700 Subject: [PATCH] Fix dataset_fields duplication for NULL type causing column_lineage explosion (#3083) DatasetFieldDao.upsert inserts into dataset_fields with an ON CONFLICT(dataset_uuid, name, type) target, relying on the (dataset_uuid, name, type) unique constraint to turn repeated upserts of the same field into an UPDATE. In standard SQL, NULL is never considered equal to another NULL for uniqueness purposes, so a field with an unknown/omitted type (a common, legitimate case for OpenLineage events that don't report column types) never matches an existing row: a brand new dataset_fields row - with a new UUID - is inserted on every single upsert instead of updating the existing one. Because each distinct dataset_fields row independently accumulates its own column_lineage edges (column_lineage references output/input dataset_field_uuid), this turns a linear ingestion process into a combinatorial one: N duplicate field rows for a null-typed field produce N column_lineage rows for what should be a single relationship, exactly as described in the issue (observed 1000x+ row bloat in column_lineage). I independently reproduced and verified this bug and its fix against a real Postgres 16 instance (matching the version pinned by this repo's tests) using the actual INSERT/ON CONFLICT SQL, outside of Gradle/Testcontainers: - Before fix: two upserts of the same (dataset_uuid, name) with type = NULL produced 2 distinct dataset_fields rows. - After fix: the same two upserts produced exactly 1 row, with the UUID from the first insert preserved (confirming the second upsert performed an UPDATE, not an INSERT). Fix: normalize a NULL type to the sentinel literal 'UNKNOWN' before insertion (COALESCE(:type, 'UNKNOWN')), so repeated upserts of a field with no known type reliably collide with the previously inserted row on the existing unique constraint. This matches the fix suggested directly in the issue report and avoids any change to the dataset_fields schema (no migration needed), at the cost of a dataset field's type now surfacing as the literal string "UNKNOWN" via the API instead of null when the type was never reported - updated testDatasetWithUnknownFieldType in MarquezAppIntegrationTest to reflect this. Tests added: - api/src/test/java/marquez/db/DatasetFieldDaoTest.java (new): DAO-level regression tests asserting repeated upserts of a null-typed field resolve to the same row/UUID, that the stored type becomes "UNKNOWN", and that real (non-null) types are unaffected. - MarquezAppIntegrationTest#testDatasetWithUnknownFieldType_repeatedUpsertsDoNotDuplicateField (new): full-stack regression test emitting the same dataset with a null-typed field 5 times and asserting only a single field is returned. - MarquezAppIntegrationTest#testDatasetWithUnknownFieldType: updated to expect the new "UNKNOWN" sentinel value. Test evidence: I was unable to execute these Postgres-backed DAO/ integration tests locally via Gradle - Testcontainers in this sandbox fails to negotiate with the local Docker Engine (an old docker-java client bundled in the pinned Testcontainers version defaults to Docker API v1.32, while the local engine requires >= v1.40). This affects every DB-backed test in the suite (verified the same failure on pre-existing, unmodified tests like RunDaoTest), so it is an environment limitation unrelated to this change. In lieu of that, I validated the exact INSERT/ON CONFLICT SQL directly against a real Postgres 16 container via psql (see PR description for the full before/after transcript). ./gradlew :api:compileTestJava and :api:testUnit both pass. Fixes #3083 Co-Authored-By: Claude Sonnet 5 Signed-off-by: Mohit Kalra --- .../main/java/marquez/db/DatasetFieldDao.java | 20 ++- .../marquez/MarquezAppIntegrationTest.java | 45 ++++++- .../java/marquez/db/DatasetFieldDaoTest.java | 117 ++++++++++++++++++ 3 files changed, 180 insertions(+), 2 deletions(-) create mode 100644 api/src/test/java/marquez/db/DatasetFieldDaoTest.java diff --git a/api/src/main/java/marquez/db/DatasetFieldDao.java b/api/src/main/java/marquez/db/DatasetFieldDao.java index c093eb8212..d9def28368 100644 --- a/api/src/main/java/marquez/db/DatasetFieldDao.java +++ b/api/src/main/java/marquez/db/DatasetFieldDao.java @@ -251,6 +251,24 @@ WHERE CAST((:namespaceName, :datasetName) AS DATASET_NAME) = ANY(d.dataset_symli """) List findInputFieldsDataAssociatedWithRun(UUID runUuid); + /** + * Upserts a {@code dataset_fields} row. + * + *

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). + * + *

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, " diff --git a/api/src/test/java/marquez/MarquezAppIntegrationTest.java b/api/src/test/java/marquez/MarquezAppIntegrationTest.java index 3bd65ab7a2..be76e6fb20 100644 --- a/api/src/test/java/marquez/MarquezAppIntegrationTest.java +++ b/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 diff --git a/api/src/test/java/marquez/db/DatasetFieldDaoTest.java b/api/src/test/java/marquez/db/DatasetFieldDaoTest.java new file mode 100644 index 0000000000..5f533e0ccd --- /dev/null +++ b/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()); + } +}