From c890432c4fd76213e3c733b1df4b52657891977e Mon Sep 17 00:00:00 2001 From: Mohit Kalra Date: Tue, 21 Jul 2026 18:10:15 -0700 Subject: [PATCH] Exclude runId from DatasetVersion hash to prevent version explosion (#3082) Utils.newDatasetVersionFor(DatasetVersionData) included the run's UUID as an input to the SHA/UUID-based hash used to compute a dataset's Version. Since a DatasetVersion is meant to identify the *content* of a dataset (namespace, source, physical name, schema fields, lifecycle state), including runId meant a brand new DatasetVersion (and corresponding dataset_versions / dataset_versions_field_mapping rows) was created on every single run that wrote to a dataset - even when the dataset's schema/content was completely unchanged from the previous run. For datasets written by high-frequency jobs this causes unbounded growth of dataset version rows and the query timeouts described in the issue. This method is the single choke point used by all three DatasetVersion-hashing call sites (OpenLineageDao, RunDao, and DatasetVersionDao), so removing runId from the hash here fixes the issue everywhere a DatasetVersion is minted, without needing to touch any call sites or public method signatures. Adds a regression test, testDatasetVersionEqualAcrossDifferentRunsWithSameDatasetContent, which asserts that two DatasetVersions computed from identical dataset content but different runIds are equal. Verified this test fails before the fix (two different UUIDs) and passes after (same UUID). Fixes #3082 Co-Authored-By: Claude Sonnet 5 Signed-off-by: Mohit Kalra --- api/src/main/java/marquez/common/Utils.java | 9 +++-- .../test/java/marquez/common/UtilsTest.java | 36 +++++++++++++++++++ 2 files changed, 43 insertions(+), 2 deletions(-) diff --git a/api/src/main/java/marquez/common/Utils.java b/api/src/main/java/marquez/common/Utils.java index 6cfd7d1765..442be8ab75 100644 --- a/api/src/main/java/marquez/common/Utils.java +++ b/api/src/main/java/marquez/common/Utils.java @@ -339,6 +339,12 @@ public static Version newDatasetVersionFor( } private static Version newDatasetVersionFor(DatasetVersionData data) { + // Note: runId is intentionally excluded from the hash. A DatasetVersion identifies the + // *content* of a dataset (its namespace, source, physical name, schema, and lifecycle + // state) - not the run that produced it. Including runId here would mean a brand new + // DatasetVersion is minted on every single run even when the dataset itself hasn't + // changed, causing unbounded growth of dataset_versions/dataset_versions_field_mapping + // rows for datasets written by high-frequency jobs (see #3082). final byte[] bytes = VERSION_JOINER .join( @@ -348,8 +354,7 @@ private static Version newDatasetVersionFor(DatasetVersionData data) { data.getPhysicalName(), data.getSchemaLocation(), data.getFields().stream().map(Utils::joinField).collect(joining(VERSION_DELIM)), - data.getLifecycleState(), - data.getRunId()) + data.getLifecycleState()) .getBytes(UTF_8); return Version.of(UUID.nameUUIDFromBytes(bytes)); } diff --git a/api/src/test/java/marquez/common/UtilsTest.java b/api/src/test/java/marquez/common/UtilsTest.java index d77c0684df..af84236484 100644 --- a/api/src/test/java/marquez/common/UtilsTest.java +++ b/api/src/test/java/marquez/common/UtilsTest.java @@ -286,6 +286,42 @@ public void testDatasetVersionEqualOnSameData() { assertThat(first).isEqualTo(second); } + @Test + public void testDatasetVersionEqualAcrossDifferentRunsWithSameDatasetContent() { + // A DatasetVersion identifies the *content* (schema, physical name, lifecycle state, etc.) of + // a dataset. Two runs writing the exact same dataset content should therefore produce the + // *same* DatasetVersion, even though the runs themselves have different runIds. Including + // runId in the hash causes a new DatasetVersion to be created on every run, even when nothing + // about the dataset changed (see: https://github.com/MarquezProject/marquez/issues/3082). + final NamespaceName namespaceName = newNamespaceName(); + DatasetName datasetName = newDatasetName(); + DatasetName physicalName = newDatasetName(); + SourceName sourceName = newSourceName(); + String lifecycleState = newLifecycleState(); + List schemaFields = newSchemaFields(2); + + Version first = + Utils.newDatasetVersionFor( + namespaceName.getValue(), + sourceName.getValue(), + physicalName.getValue(), + datasetName.getValue(), + lifecycleState, + schemaFields, + newRunId().getValue()); + Version second = + Utils.newDatasetVersionFor( + namespaceName.getValue(), + sourceName.getValue(), + physicalName.getValue(), + datasetName.getValue(), + lifecycleState, + schemaFields, + newRunId().getValue()); + + assertThat(first).isEqualTo(second); + } + @Test public void testDatasetVersionForDBTableMetaDataIsNotEqualOnDifferentData() { DbTableMeta dbTableMeta = newDbTableMeta();