Merge branch 'main' into feature/db-retention

Author: wslulciuc

.circleci/api-load-test.sh

@@ -14,7 +14,7 @@
 set -e
 
 # Build version of Marquez
-readonly MARQUEZ_VERSION=0.37.0-SNAPSHOT
+readonly MARQUEZ_VERSION=0.38.0-SNAPSHOT
 # Fully qualified path to marquez.jar
 readonly MARQUEZ_JAR="api/build/libs/marquez-api-${MARQUEZ_VERSION}.jar"
 

.circleci/db-migration.sh

@@ -13,7 +13,7 @@
 # Version of PostgreSQL
 readonly POSTGRES_VERSION="12.1"
 # Version of Marquez
-readonly MARQUEZ_VERSION=0.36.0
+readonly MARQUEZ_VERSION=0.37.0
 # Build version of Marquez
 readonly MARQUEZ_BUILD_VERSION="$(git log --pretty=format:'%h' -n 1)" # SHA1
 

.env.example

@@ -1,4 +1,4 @@
 API_PORT=5000
 API_ADMIN_PORT=5001
 WEB_PORT=3000
-TAG=0.36.0
+TAG=0.37.0

CHANGELOG.md

@@ -1,9 +1,17 @@
 # Changelog
 
-## [Unreleased](https://github.com/MarquezProject/marquez/compare/0.36.0...HEAD)
+## [Unreleased](https://github.com/MarquezProject/marquez/compare/0.37.0...HEAD)
+
+## [0.37.0](https://github.com/MarquezProject/marquez/compare/0.36.0...0.37.0) - 2023-07-17
+### Added
+* API: add ability to decode static metadata events [`#2495`](https://github.com/MarquezProject/marquez/pull/2495) [@pawel-big-lebowski]( https://github.com/pawel-big-lebowski)  
+    *Introduces an `EventTypeResolver` for using the `schemaURL` field to decode `POST` requests to `/lineage` with `LineageEvent`s, `DatasetEvent`s or `JobEvent`s, as the first step in implementing static lineage support.*
+
 ### Fixed
-* API: remove unnecessary DB updates [`#2531`](https://github.com/MarquezProject/marquez/pull/2531)[@pawel-big-lebowski]( https://github.com/pawel-big-lebowski)  
-  *Prevent updates that are not needed and are deadlock prone.*
+* API: remove unnecessary DB updates [`#2531`](https://github.com/MarquezProject/marquez/pull/2531) [@pawel-big-lebowski]( https://github.com/pawel-big-lebowski)  
+    *Prevent updates that are not needed and are deadlock-prone.*
+* Web: revert URL encoding when fetching lineage [`#2529`](https://github.com/MarquezProject/marquez/pull/2529) [@jlukenoff](https://github.com/jlukenoff)  
+    *Reverts the node ID from being URL-encoded and allows the backend to return lineage details successfully even when a node ID contains special characters.*
 
 ## [0.36.0](https://github.com/MarquezProject/marquez/compare/0.35.0...0.36.0) - 2023-06-27
 ### Added
@@ -28,6 +36,12 @@
 * Web: handle lineage graph cycles on the client [`#2506`](https://github.com/MarquezProject/marquez/pull/2506) [@jlukenoff](https://github.com/jlukenoff)  
     *Fixes a bug where we blow the stack on the client-side if the user selects a node that is part of a cycle in the graph.*
 
+### Added
+
+* Ability to decode static metadata events [`#2495`](https://github.com/MarquezProject/marquez/pull/2495) [@pawel-big-lebowski]( https://github.com/pawel-big-lebowski)  
+  *Adds the ability to distinguish on a bakend static metadata events introduced based on the [proposal](https://github.com/OpenLineage/OpenLineage/blob/main/proposals/1837/static_lineage.md).*
+
+
 ## [0.34.0](https://github.com/MarquezProject/marquez/compare/0.33.0...0.34.0) - 2023-05-18
 
 ### Fixed

README.md

@@ -79,8 +79,8 @@ Versions of Marquez are compatible with OpenLineage unless noted otherwise. We e
 | **Marquez**                                                                                      | **OpenLineage**                                               | **Status**    |
 |--------------------------------------------------------------------------------------------------|---------------------------------------------------------------|---------------|
 | [`UNRELEASED`](https://github.com/MarquezProject/marquez/blob/main/CHANGELOG.md#unreleased)      | [`1-0-5`](https://openlineage.io/spec/1-0-5/OpenLineage.json) | `CURRENT`     |
-| [`0.35.0`](https://github.com/MarquezProject/marquez/blob/0.35.0/CHANGELOG.md#0350---2023-06-13) | [`1-0-5`](https://openlineage.io/spec/1-0-5/OpenLineage.json) | `RECOMMENDED` |
-| [`0.34.0`](https://github.com/MarquezProject/marquez/blob/0.34.0/CHANGELOG.md#0340---2023-05-18) | [`1-0-5`](https://openlineage.io/spec/1-0-0/OpenLineage.json) | `MAINTENANCE` |
+| [`0.37.0`](https://github.com/MarquezProject/marquez/blob/0.37.0/CHANGELOG.md#0370---2023-07-17) | [`1-0-5`](https://openlineage.io/spec/1-0-5/OpenLineage.json) | `RECOMMENDED` |
+| [`0.36.0`](https://github.com/MarquezProject/marquez/blob/0.36.0/CHANGELOG.md#0360---2023-06-27) | [`1-0-5`](https://openlineage.io/spec/1-0-0/OpenLineage.json) | `MAINTENANCE` |
 
 > **Note:** The [`openlineage-python`](https://pypi.org/project/openlineage-python) and [`openlineage-java`](https://central.sonatype.com/artifact/io.openlineage/openlineage-java) libraries will a higher version than the OpenLineage [specification](https://github.com/OpenLineage/OpenLineage/tree/main/spec) as they have different version requirements.
 
@@ -160,7 +160,7 @@ Marquez listens on port `8080` for all API calls and port `8081` for the admin i
 
 * Website: https://marquezproject.ai
 * Source: https://github.com/MarquezProject/marquez
-* Chat: [https://marquezproject.slack.com](https://bit.ly/MqzSlack)
+* Chat: [MarquezProject Slack](https://bit.ly/MqzSlackInvite)
 * Twitter: [@MarquezProject](https://twitter.com/MarquezProject)
 
 ## Contributing

api/src/main/java/marquez/api/OpenLineageResource.java

@@ -38,6 +38,7 @@
 import marquez.api.models.SortDirection;
 import marquez.db.OpenLineageDao;
 import marquez.service.ServiceFactory;
+import marquez.service.models.BaseEvent;
 import marquez.service.models.LineageEvent;
 import marquez.service.models.NodeId;
 
@@ -61,20 +62,26 @@ public OpenLineageResource(
   @Consumes(APPLICATION_JSON)
   @Produces(APPLICATION_JSON)
   @Path("/lineage")
-  public void create(
-      @Valid @NotNull LineageEvent event, @Suspended final AsyncResponse asyncResponse)
+  public void create(@Valid @NotNull BaseEvent event, @Suspended final AsyncResponse asyncResponse)
       throws JsonProcessingException, SQLException {
-    openLineageService
-        .createAsync(event)
-        .whenComplete(
-            (result, err) -> {
-              if (err != null) {
-                log.error("Unexpected error while processing request", err);
-                asyncResponse.resume(Response.status(determineStatusCode(err)).build());
-              } else {
-                asyncResponse.resume(Response.status(201).build());
-              }
-            });
+    if (event instanceof LineageEvent) {
+      openLineageService
+          .createAsync((LineageEvent) event)
+          .whenComplete(
+              (result, err) -> {
+                if (err != null) {
+                  log.error("Unexpected error while processing request", err);
+                  asyncResponse.resume(Response.status(determineStatusCode(err)).build());
+                } else {
+                  asyncResponse.resume(Response.status(201).build());
+                }
+              });
+    } else {
+      log.warn("Unsupported event type {}. Skipping without error", event.getClass().getName());
+
+      // return serialized event
+      asyncResponse.resume(Response.status(200).entity(event).build());
+    }
   }
 
   private int determineStatusCode(Throwable e) {

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

@@ -71,26 +71,28 @@ void updateLastModifiedAt(
 
   @SqlQuery(
       """
-      SELECT d.*, dv.fields, dv.lifecycle_state, sv.schema_location, t.tags, facets
-      FROM datasets_view d
-      LEFT JOIN dataset_versions dv ON d.current_version_uuid = dv.uuid
-      LEFT JOIN stream_versions AS sv ON sv.dataset_version_uuid = dv.uuid
-      LEFT JOIN (
-          SELECT ARRAY_AGG(t.name) AS tags, m.dataset_uuid
-          FROM tags AS t
-                   INNER JOIN datasets_tag_mapping AS m ON m.tag_uuid = t.uuid
-          GROUP BY m.dataset_uuid
-      ) t ON t.dataset_uuid = d.uuid
-      LEFT JOIN (
-          SELECT
-              df.dataset_version_uuid,
-              JSONB_AGG(df.facet ORDER BY df.lineage_event_time ASC) AS facets
-          FROM dataset_facets_view AS df
-          WHERE df.facet IS NOT NULL AND (df.type ILIKE 'dataset' OR df.type ILIKE 'unknown')
-          GROUP BY df.dataset_version_uuid
-      ) f ON f.dataset_version_uuid = d.current_version_uuid
-      WHERE CAST((:namespaceName, :datasetName) AS DATASET_NAME) = ANY(d.dataset_symlinks)
-  """)
+          SELECT d.*, dv.fields, dv.lifecycle_state, sv.schema_location, t.tags, facets
+          FROM datasets_view d
+          LEFT JOIN dataset_versions dv ON d.current_version_uuid = dv.uuid
+          LEFT JOIN stream_versions AS sv ON sv.dataset_version_uuid = dv.uuid
+          LEFT JOIN (
+              SELECT ARRAY_AGG(t.name) AS tags, m.dataset_uuid
+              FROM tags AS t
+                       INNER JOIN datasets_tag_mapping AS m ON m.tag_uuid = t.uuid
+              GROUP BY m.dataset_uuid
+          ) t ON t.dataset_uuid = d.uuid
+          LEFT JOIN (
+              SELECT
+                  df.dataset_version_uuid,
+                  JSONB_AGG(df.facet ORDER BY df.lineage_event_time ASC) AS facets
+              FROM dataset_facets AS df
+              WHERE df.facet IS NOT NULL AND
+               (df.type ILIKE 'dataset' OR df.type ILIKE 'unknown') AND
+                df.dataset_uuid = (SELECT uuid FROM datasets WHERE name = :datasetName AND namespace_name = :namespaceName)
+              GROUP BY df.dataset_version_uuid
+          ) f ON f.dataset_version_uuid = d.current_version_uuid
+          WHERE CAST((:namespaceName, :datasetName) AS DATASET_NAME) = ANY(d.dataset_symlinks)
+      """)
   Optional<Dataset> findDatasetByName(String namespaceName, String datasetName);
 
   default Optional<Dataset> findWithTags(String namespaceName, String datasetName) {
@@ -119,28 +121,30 @@ default void setFields(Dataset ds) {
 
   @SqlQuery(
       """
-      SELECT d.*, dv.fields, dv.lifecycle_state, sv.schema_location, t.tags, facets
-      FROM datasets_view d
-      LEFT JOIN dataset_versions dv ON d.current_version_uuid = dv.uuid
-      LEFT JOIN stream_versions AS sv ON sv.dataset_version_uuid = dv.uuid
-      LEFT JOIN (
-          SELECT ARRAY_AGG(t.name) AS tags, m.dataset_uuid
-          FROM tags AS t
-          INNER JOIN datasets_tag_mapping AS m ON m.tag_uuid = t.uuid
-          GROUP BY m.dataset_uuid
-      ) t ON t.dataset_uuid = d.uuid
-      LEFT JOIN (
-          SELECT
-              df.dataset_version_uuid,
-              JSONB_AGG(df.facet ORDER BY df.lineage_event_time ASC) AS facets
-          FROM dataset_facets_view AS df
-          WHERE df.facet IS NOT NULL AND (df.type ILIKE 'dataset' OR df.type ILIKE 'unknown')
-          GROUP BY df.dataset_version_uuid
-      ) f ON f.dataset_version_uuid = d.current_version_uuid
-      WHERE d.namespace_name = :namespaceName
-      ORDER BY d.name
-      LIMIT :limit OFFSET :offset
-      """)
+          SELECT d.*, dv.fields, dv.lifecycle_state, sv.schema_location, t.tags, facets
+          FROM datasets_view d
+          LEFT JOIN dataset_versions dv ON d.current_version_uuid = dv.uuid
+          LEFT JOIN stream_versions AS sv ON sv.dataset_version_uuid = dv.uuid
+          LEFT JOIN (
+              SELECT ARRAY_AGG(t.name) AS tags, m.dataset_uuid
+              FROM tags AS t
+              INNER JOIN datasets_tag_mapping AS m ON m.tag_uuid = t.uuid
+              GROUP BY m.dataset_uuid
+          ) t ON t.dataset_uuid = d.uuid
+          LEFT JOIN (
+              SELECT
+                  df.dataset_version_uuid,
+                  JSONB_AGG(df.facet ORDER BY df.lineage_event_time ASC) AS facets
+                FROM dataset_facets AS df
+                WHERE df.facet IS NOT NULL AND
+                 (df.type ILIKE 'dataset' OR df.type ILIKE 'unknown') AND
+                  df.dataset_uuid IN (SELECT uuid FROM datasets_view WHERE namespace_name = :namespaceName ORDER BY name LIMIT :limit OFFSET :offset)
+              GROUP BY df.dataset_version_uuid
+          ) f ON f.dataset_version_uuid = d.current_version_uuid
+          WHERE d.namespace_name = :namespaceName
+          ORDER BY d.name
+          LIMIT :limit OFFSET :offset
+          """)
   List<Dataset> findAll(String namespaceName, int limit, int offset);
 
   @SqlQuery("SELECT count(*) FROM datasets_view")
@@ -249,23 +253,23 @@ DatasetRow upsert(
 
   @SqlUpdate(
       """
-    UPDATE datasets d
-    SET is_hidden = true
-    FROM namespaces n
-    WHERE n.uuid=d.namespace_uuid
-    AND n.name=:namespaceName
-  """)
+        UPDATE datasets d
+        SET is_hidden = true
+        FROM namespaces n
+        WHERE n.uuid=d.namespace_uuid
+        AND n.name=:namespaceName
+      """)
   void deleteByNamespaceName(String namespaceName);
 
   @SqlQuery(
       """
-    UPDATE datasets d
-    SET is_hidden = true
-    FROM namespaces n
-    WHERE n.uuid = d.namespace_uuid
-    AND n.name=:namespaceName AND d.name=:name
-    RETURNING *
-  """)
+        UPDATE datasets d
+        SET is_hidden = true
+        FROM namespaces n
+        WHERE n.uuid = d.namespace_uuid
+        AND n.name=:namespaceName AND d.name=:name
+        RETURNING *
+      """)
   Optional<DatasetRow> delete(String namespaceName, String name);
 
   @Transaction

api/src/main/java/marquez/service/models/BaseEvent.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2018-2023 contributors to the Marquez project
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package marquez.service.models;
+
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.fasterxml.jackson.annotation.JsonTypeInfo.As;
+import com.fasterxml.jackson.annotation.JsonTypeInfo.Id;
+import com.fasterxml.jackson.databind.annotation.JsonTypeIdResolver;
+
+@JsonTypeIdResolver(EventTypeResolver.class)
+@JsonTypeInfo(
+    use = Id.CUSTOM,
+    include = As.EXISTING_PROPERTY,
+    property = "schemaURL",
+    defaultImpl = LineageEvent.class,
+    visible = true)
+public class BaseEvent extends BaseJsonModel {}

api/src/main/java/marquez/service/models/DatasetEvent.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018-2023 contributors to the Marquez project
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package marquez.service.models;
+
+import java.net.URI;
+import java.time.ZonedDateTime;
+import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+import lombok.ToString;
+
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+@Setter
+@Getter
+@Valid
+@ToString
+public class DatasetEvent extends BaseEvent {
+  @NotNull private ZonedDateTime eventTime;
+  @Valid private LineageEvent.Dataset dataset;
+  @Valid @NotNull private String producer;
+  @Valid @NotNull private URI schemaURL;
+}