Exclude runId from DatasetVersion hash to prevent version explosion

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

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<LineageEvent.SchemaField> 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();