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()); + } +}