Merge branch 'main' into fix/remove-limit

Author: wslulciuc

api/src/main/java/marquez/common/Utils.java

@@ -33,6 +33,8 @@
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.ZonedDateTime;
+import java.util.Collection;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -59,6 +61,7 @@
 import marquez.service.models.LineageEvent;
 import marquez.service.models.LineageEvent.ParentRunFacet;
 import marquez.service.models.StreamMeta;
+import org.apache.commons.lang3.tuple.Pair;
 import org.apache.commons.lang3.tuple.Triple;
 
 public final class Utils {
@@ -351,6 +354,36 @@ private static Version newDatasetVersionFor(DatasetVersionData data) {
     return Version.of(UUID.nameUUIDFromBytes(bytes));
   }
 
+  /**
+   * Returns a new {@link Version} object based on the dataset id and collection of fields
+   * information. A {@link Version} is generated by concatenating the provided metadata together
+   * (delimited by a colon). For fields, only the name and type contribute. The resulting string is
+   * then converted to a {@code byte} array and passed to {@link UUID#nameUUIDFromBytes(byte[])}.
+   *
+   * @param namespaceName The namespace of the dataset.
+   * @param datasetName The dataset name.
+   * @param fields The fields of the dataset.
+   * @return A {@link Version} object based on the specified dataset metadata.
+   */
+  public static Version newDatasetSchemaVersionFor(
+      String namespaceName, String datasetName, Collection<Pair<String, String>> fields) {
+    final byte[] bytes =
+        VERSION_JOINER
+            .join(
+                namespaceName,
+                datasetName,
+                Optional.ofNullable(fields).orElse(Collections.emptyList()).stream()
+                    .sorted()
+                    .map(Utils::joinField)
+                    .collect(joining(VERSION_DELIM)))
+            .getBytes(UTF_8);
+    return Version.of(UUID.nameUUIDFromBytes(bytes));
+  }
+
+  private static String joinField(Pair<String, String> field) {
+    return VERSION_JOINER.join(field.getLeft(), field.getRight());
+  }
+
   private static String joinField(Triple<String, String, String> field) {
     return VERSION_JOINER.join(field.getLeft(), field.getMiddle(), field.getRight());
   }

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

@@ -18,6 +18,9 @@ public interface BaseDao extends SqlObject {
   @CreateSqlObject
   DatasetVersionDao createDatasetVersionDao();
 
+  @CreateSqlObject
+  DatasetSchemaVersionDao createDatasetSchemaVersionDao();
+
   @CreateSqlObject
   JobDao createJobDao();
 

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

@@ -49,6 +49,7 @@ private Columns() {}
   public static final String NAMESPACE_UUID = "namespace_uuid";
   public static final String DATASET_UUID = "dataset_uuid";
   public static final String DATASET_VERSION_UUID = "dataset_version_uuid";
+  public static final String DATASET_SCHEMA_VERSION_UUID = "dataset_schema_version_uuid";
   public static final String JOB_VERSION_UUID = "job_version_uuid";
   public static final String CURRENT_VERSION_UUID = "current_version_uuid";
   public static final String CHECKSUM = "checksum";

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

@@ -107,7 +107,7 @@ default void setFields(Dataset ds) {
     ds.getCurrentVersion()
         .ifPresent(
             dsv -> {
-              ds.setFields(datasetFieldDao.find(dsv));
+              ds.setFields(datasetFieldDao.findByDatasetVersion(dsv));
             });
   }
 
@@ -363,7 +363,7 @@ default Dataset upsertDatasetMeta(
     DatasetVersionRow dvRow =
         createDatasetVersionDao()
             .upsertDatasetVersion(
-                datasetRow.getUuid(),
+                datasetRow,
                 now,
                 namespaceName.getValue(),
                 datasetName.getValue(),

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

@@ -220,7 +220,18 @@ WHERE CAST((:namespaceName, :datasetName) AS DATASET_NAME) = ANY(d.dataset_symli
           + "FROM dataset_fields f "
           + "INNER JOIN dataset_versions_field_mapping fm on fm.dataset_field_uuid = f.uuid "
           + "WHERE fm.dataset_version_uuid = :datasetVersionUuid")
-  List<Field> find(UUID datasetVersionUuid);
+  List<Field> findByDatasetVersion(UUID datasetVersionUuid);
+
+  @SqlQuery(
+      "SELECT f.*, "
+          + "ARRAY(SELECT t.name "
+          + "      FROM dataset_fields_tag_mapping m "
+          + "      INNER JOIN tags t on t.uuid = m.tag_uuid "
+          + "      WHERE m.dataset_field_uuid = f.uuid) AS tags "
+          + "FROM dataset_fields f "
+          + "INNER JOIN dataset_schema_versions_field_mapping fm on fm.dataset_field_uuid = f.uuid "
+          + "WHERE fm.dataset_schema_version_uuid = :datasetSchemaVersionUuid")
+  List<Field> findByDatasetSchemaVersion(UUID datasetSchemaVersionUuid);
 
   @SqlQuery(
       """

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

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2018-2024 contributors to the Marquez project
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package marquez.db;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Optional;
+import java.util.UUID;
+import java.util.stream.Collectors;
+import marquez.common.Utils;
+import marquez.common.models.Version;
+import marquez.db.mappers.DatasetSchemaVersionRowMapper;
+import marquez.db.models.DatasetFieldRow;
+import marquez.db.models.DatasetRow;
+import marquez.db.models.DatasetSchemaVersionRow;
+import org.apache.commons.lang3.tuple.Pair;
+import org.jdbi.v3.sqlobject.config.RegisterRowMapper;
+import org.jdbi.v3.sqlobject.statement.SqlBatch;
+import org.jdbi.v3.sqlobject.statement.SqlQuery;
+
+@RegisterRowMapper(DatasetSchemaVersionRowMapper.class)
+public interface DatasetSchemaVersionDao extends BaseDao {
+  default Version upsertSchemaVersion(
+      DatasetRow datasetRow, List<DatasetFieldRow> datasetFields, Instant now) {
+    final Version computedVersion =
+        Utils.newDatasetSchemaVersionFor(
+            datasetRow.getNamespaceName(),
+            datasetRow.getName(),
+            datasetFields.stream()
+                .map(field -> Pair.of(field.getName(), field.getType()))
+                .collect(Collectors.toSet()));
+    upsertSchemaVersion(computedVersion.getValue(), datasetRow.getUuid(), now)
+        .ifPresent(
+            newRow -> {
+              // if not null it means a new insert, so we have to do the fields as well
+              // if null then it means the version already exists, and so the fields must already
+              // exist
+              upsertFieldMappings(
+                  newRow.getUuid(),
+                  datasetFields.stream()
+                      .map(DatasetFieldRow::getUuid)
+                      .collect(Collectors.toList()));
+            });
+    return computedVersion;
+  }
+
+  @SqlQuery(
+      "INSERT INTO dataset_schema_versions "
+          + "(uuid, dataset_uuid, created_at) "
+          + "VALUES (:uuid, :datasetUuid, :now) "
+          + "ON CONFLICT DO NOTHING "
+          + "RETURNING *")
+  Optional<DatasetSchemaVersionRow> upsertSchemaVersion(UUID uuid, UUID datasetUuid, Instant now);
+
+  @SqlBatch(
+      "INSERT INTO dataset_schema_versions_field_mapping "
+          + "(dataset_schema_version_uuid, dataset_field_uuid) "
+          + "VALUES (:schemaVersionUuid, :fieldUuid) "
+          + "ON CONFLICT DO NOTHING")
+  void upsertFieldMappings(UUID schemaVersionUuid, Iterable<UUID> fieldUuid);
+}

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

@@ -25,6 +25,7 @@
 import marquez.db.mappers.DatasetVersionRowMapper;
 import marquez.db.mappers.ExtendedDatasetVersionRowMapper;
 import marquez.db.models.DatasetFieldRow;
+import marquez.db.models.DatasetRow;
 import marquez.db.models.DatasetVersionRow;
 import marquez.db.models.ExtendedDatasetVersionRow;
 import marquez.db.models.TagRow;
@@ -47,7 +48,7 @@ public interface DatasetVersionDao extends BaseDao {
 
   @Transaction
   default DatasetVersionRow upsertDatasetVersion(
-      UUID datasetUuid,
+      DatasetRow datasetRow,
       Instant now,
       String namespaceName,
       String datasetName,
@@ -56,14 +57,38 @@ default DatasetVersionRow upsertDatasetVersion(
     TagDao tagDao = createTagDao();
     DatasetFieldDao datasetFieldDao = createDatasetFieldDao();
 
+    List<DatasetFieldRow> datasetFields = new ArrayList<>();
+    List<DatasetFieldTag> datasetFieldTags = new ArrayList<>();
+    for (Field field : datasetMeta.getFields()) {
+      DatasetFieldRow datasetFieldRow =
+          datasetFieldDao.upsert(
+              UUID.randomUUID(),
+              now,
+              field.getName().getValue(),
+              field.getType(),
+              field.getDescription().orElse(null),
+              datasetRow.getUuid());
+      datasetFields.add(datasetFieldRow);
+      for (TagName tagName : field.getTags()) {
+        TagRow tag = tagDao.upsert(UUID.randomUUID(), now, tagName.getValue());
+        datasetFieldTags.add(new DatasetFieldTag(datasetFieldRow.getUuid(), tag.getUuid(), now));
+      }
+    }
+    datasetFieldDao.updateTags(datasetFieldTags);
+
     Version version = Utils.newDatasetVersionFor(namespaceName, datasetName, datasetMeta);
     UUID newDatasetVersionUuid = UUID.randomUUID();
+    UUID datasetSchemaVersionUuid =
+        createDatasetSchemaVersionDao()
+            .upsertSchemaVersion(datasetRow, datasetFields, now)
+            .getValue();
     DatasetVersionRow datasetVersionRow =
         upsert(
             newDatasetVersionUuid,
             now,
-            datasetUuid,
+            datasetRow.getUuid(),
             version.getValue(),
+            datasetSchemaVersionUuid,
             datasetMeta.getRunId().map(RunId::getValue).orElse(null),
             toPgObjectFields(datasetMeta.getFields()),
             namespaceName,
@@ -84,29 +109,13 @@ default DatasetVersionRow upsertDatasetVersion(
     }
 
     List<DatasetFieldMapping> datasetFieldMappings = new ArrayList<>();
-    List<DatasetFieldTag> datasetFieldTag = new ArrayList<>();
-
-    for (Field field : datasetMeta.getFields()) {
-      DatasetFieldRow datasetFieldRow =
-          datasetFieldDao.upsert(
-              UUID.randomUUID(),
-              now,
-              field.getName().getValue(),
-              field.getType(),
-              field.getDescription().orElse(null),
-              datasetUuid);
-      for (TagName tagName : field.getTags()) {
-        TagRow tag = tagDao.upsert(UUID.randomUUID(), now, tagName.getValue());
-        datasetFieldTag.add(new DatasetFieldTag(datasetFieldRow.getUuid(), tag.getUuid(), now));
-      }
+    for (DatasetFieldRow datasetFieldRow : datasetFields) {
       datasetFieldMappings.add(
           new DatasetFieldMapping(datasetVersionRow.getUuid(), datasetFieldRow.getUuid()));
     }
-
     datasetFieldDao.updateFieldMapping(datasetFieldMappings);
-    datasetFieldDao.updateTags(datasetFieldTag);
 
-    createDatasetDao().updateVersion(datasetUuid, now, datasetVersionRow.getUuid());
+    createDatasetDao().updateVersion(datasetRow.getUuid(), now, datasetVersionRow.getUuid());
     return datasetVersionRow;
   }
 
@@ -166,8 +175,8 @@ WITH selected_dataset_versions AS (
           LEFT JOIN dataset_facets_view df ON df.dataset_version_uuid = dv.uuid
       )
       SELECT d.type, d.name, d.physical_name, d.namespace_name, d.source_name, d.description, dv.lifecycle_state,\s
-          dv.created_at, dv.version, dv.fields, dv.run_uuid AS createdByRunUuid, sv.schema_location,
-          t.tags, f.facets
+          dv.created_at, dv.version, dv.dataset_schema_version_uuid, dv.fields, dv.run_uuid AS createdByRunUuid,
+          sv.schema_location, t.tags, f.facets
       FROM selected_dataset_versions dv
       LEFT JOIN datasets_view d ON d.uuid = dv.dataset_uuid
       LEFT JOIN stream_versions AS sv ON sv.dataset_version_uuid = dv.uuid
@@ -197,8 +206,8 @@ WITH selected_dataset_versions AS (
           LEFT JOIN dataset_facets_view df ON df.dataset_version_uuid = dv.uuid  AND (df.type ILIKE 'dataset' OR df.type ILIKE 'unknown' OR df.type ILIKE 'input')
       )
       SELECT d.type, d.name, d.physical_name, d.namespace_name, d.source_name, d.description, dv.lifecycle_state,\s
-          dv.created_at, dv.version, dv.fields, dv.run_uuid AS createdByRunUuid, sv.schema_location,
-          t.tags, f.facets
+          dv.created_at, dv.version, dv.dataset_schema_version_uuid, dv.fields, dv.run_uuid AS createdByRunUuid,
+          sv.schema_location, t.tags, f.facets
       FROM selected_dataset_versions dv
       LEFT JOIN datasets_view d ON d.uuid = dv.dataset_uuid
       LEFT JOIN stream_versions AS sv ON sv.dataset_version_uuid = dv.uuid
@@ -246,8 +255,8 @@ default Optional<DatasetVersion> findByWithRun(UUID version) {
       """
       WITH dataset_info AS (
 	    SELECT d.type, d.name, d.physical_name, d.namespace_name, d.source_name, d.description, dv.lifecycle_state,
-		dv.created_at, dv.version, dv.fields, dv.run_uuid AS createdByRunUuid, sv.schema_location,
-		t.tags, f.facets, f.lineage_event_time, f.dataset_version_uuid, facet_name
+		dv.created_at, dv.version, dv.dataset_schema_version_uuid, dv.fields, dv.run_uuid AS createdByRunUuid,
+		sv.schema_location, t.tags, f.facets, f.lineage_event_time, f.dataset_version_uuid, facet_name
 		FROM dataset_versions dv
 		LEFT JOIN datasets_view d ON d.uuid = dv.dataset_uuid
 		LEFT JOIN stream_versions AS sv ON sv.dataset_version_uuid = dv.uuid
@@ -272,12 +281,12 @@ LEFT JOIN (
         )
         SELECT
 	        type, name, physical_name, namespace_name, source_name, description, lifecycle_state,
-            created_at, version, fields, createdByRunUuid, schema_location,
+            created_at, version, dataset_schema_version_uuid, fields, createdByRunUuid, schema_location,
             tags, dataset_version_uuid,
 	        JSONB_AGG(facets ORDER BY lineage_event_time ASC) AS facets
         FROM dataset_info
         GROUP BY type, name, physical_name, namespace_name, source_name, description, lifecycle_state,
-            created_at, version, fields, createdByRunUuid, schema_location,
+            created_at, version, dataset_schema_version_uuid, fields, createdByRunUuid, schema_location,
             tags, dataset_version_uuid
         ORDER BY created_at DESC
   """)
@@ -302,9 +311,9 @@ default List<DatasetVersion> findAllWithRun(
 
   @SqlQuery(
       "INSERT INTO dataset_versions "
-          + "(uuid, created_at, dataset_uuid, version, run_uuid, fields, namespace_name, dataset_name, lifecycle_state) "
+          + "(uuid, created_at, dataset_uuid, version, dataset_schema_version_uuid, run_uuid, fields, namespace_name, dataset_name, lifecycle_state) "
           + "VALUES "
-          + "(:uuid, :now, :datasetUuid, :version, :runUuid, :fields, :namespaceName, :datasetName, :lifecycleState) "
+          + "(:uuid, :now, :datasetUuid, :version, :schemaVersionUuid, :runUuid, :fields, :namespaceName, :datasetName, :lifecycleState) "
           + "ON CONFLICT(version) "
           + "DO UPDATE SET "
           + "run_uuid = EXCLUDED.run_uuid "
@@ -314,6 +323,7 @@ DatasetVersionRow upsert(
       Instant now,
       UUID datasetUuid,
       UUID version,
+      UUID schemaVersionUuid,
       UUID runUuid,
       PGobject fields,
       String namespaceName,

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

@@ -442,6 +442,7 @@ private static ExtendedDatasetVersionRow toExtendedDatasetVersionRow(DatasetReco
         d.getDatasetRow().getCreatedAt(),
         d.getDatasetVersionRow().getDatasetUuid(),
         d.getDatasetVersionRow().getVersion(),
+        d.getDatasetVersionRow().getSchemaVersionUuid().orElse(null),
         null,
         null,
         d.getDatasetRow().getNamespaceName(),