Merge branch 'main' into helm-chart-fix-for-web

Author: wslulciuc

CHANGELOG.md

@@ -7,7 +7,7 @@
 ### Added
 
 * UI: add facet view enhancements [`#2336`](https://github.com/MarquezProject/marquez/pull/2336) [@tito12](https://github.com/tito12)  
-    *Creates a dynamic component with the ability to navigate and search the JSON, expand sections and click on links.*
+    *Creates a dynamic component offering the ability to navigate and search the JSON, expand sections and click on links.*
 * UI: highlight selected path on graph and display status of jobs and datasets based on last 14 runs or latest quality facets [`#2384`](https://github.com/MarquezProject/marquez/pull/2384) [@tito12](https://github.com/tito12)  
     *Adds highlighting of the visual graph based on upstream and downstream dependencies of selected nodes, makes displayed status reflect last 14 runs the case of jobs and latest quality facets in the case of datasets.*
 * UI: enable auto-accessibility feature on graph nodes [`#2388`](https://github.com/MarquezProject/marquez/pull/2400) [@merobi-hub](https://github.com/merobi-hub)  
@@ -20,11 +20,14 @@
 * API: add missing indices to `column_lineage`, `dataset_facets`, `job_facets` tables [`#2419`](https://github.com/MarquezProject/marquez/pull/2419) [@pawel-big-lebowski](https://github.com/pawel-big-lebowski)  
     *Creates missing indices on reference columns in a number of database tables.*
 * Spec: make data version and dataset types the same [`#2400`](https://github.com/MarquezProject/marquez/pull/2400) [@phixme](https://github.com/phixMe)  
-    *Makes the `fields` property the same for datasets and dataset versions, allowing type-denerating systems to treat them the same way.*
+    *Makes the `fields` property the same for datasets and dataset versions, allowing type-generating systems to treat them the same way.*
 * UI: show location button only when link to code exists [`#2409`](https://github.com/MarquezProject/marquez/pull/2409) [@tito12](https://github.com/tito12)  
     *Makes the button visible only if the link is not empty.*
 
 
+* Improve dataset facets access [`#2407`](https://github.com/MarquezProject/marquez/pull/2407) [@pawel-big-lebowski](https://github.com/pawel-big-lebowski)
+  * Improves database query performance for accessing datasets and datasets' versions.*
+
 ## [0.30.0](https://github.com/MarquezProject/marquez/compare/0.29.0...0.30.0) - 2023-01-31
 
 ### Added

api/src/main/java/marquez/MarquezContext.java

@@ -28,14 +28,15 @@
 import marquez.db.DatasetDao;
 import marquez.db.DatasetFieldDao;
 import marquez.db.DatasetVersionDao;
-import marquez.db.JobContextDao;
 import marquez.db.JobDao;
+import marquez.db.JobFacetsDao;
 import marquez.db.JobVersionDao;
 import marquez.db.LineageDao;
 import marquez.db.NamespaceDao;
 import marquez.db.OpenLineageDao;
 import marquez.db.RunArgsDao;
 import marquez.db.RunDao;
+import marquez.db.RunFacetsDao;
 import marquez.db.RunStateDao;
 import marquez.db.SearchDao;
 import marquez.db.SourceDao;
@@ -67,9 +68,10 @@ public final class MarquezContext {
   @Getter private final DatasetVersionDao datasetVersionDao;
   @Getter private final JobDao jobDao;
   @Getter private final JobVersionDao jobVersionDao;
-  @Getter private final JobContextDao jobContextDao;
+  @Getter private final JobFacetsDao jobFacetsDao;
   @Getter private final RunDao runDao;
   @Getter private final RunArgsDao runArgsDao;
+  @Getter private final RunFacetsDao runFacetsDao;
   @Getter private final RunStateDao runStateDao;
   @Getter private final TagDao tagDao;
   @Getter private final OpenLineageDao openLineageDao;
@@ -116,9 +118,10 @@ private MarquezContext(
     this.datasetVersionDao = jdbi.onDemand(DatasetVersionDao.class);
     this.jobDao = jdbi.onDemand(JobDao.class);
     this.jobVersionDao = jdbi.onDemand(JobVersionDao.class);
-    this.jobContextDao = jdbi.onDemand(JobContextDao.class);
+    this.jobFacetsDao = jdbi.onDemand(JobFacetsDao.class);
     this.runDao = jdbi.onDemand(RunDao.class);
     this.runArgsDao = jdbi.onDemand(RunArgsDao.class);
+    this.runFacetsDao = jdbi.onDemand(RunFacetsDao.class);
     this.runStateDao = jdbi.onDemand(RunStateDao.class);
     this.tagDao = jdbi.onDemand(TagDao.class);
     this.openLineageDao = jdbi.onDemand(OpenLineageDao.class);
@@ -158,7 +161,7 @@ private MarquezContext(
     this.sourceResource = new SourceResource(serviceFactory);
     this.datasetResource = new DatasetResource(serviceFactory);
     this.columnLineageResource = new ColumnLineageResource(serviceFactory);
-    this.jobResource = new JobResource(serviceFactory, jobVersionDao);
+    this.jobResource = new JobResource(serviceFactory, jobVersionDao, jobFacetsDao, runFacetsDao);
     this.tagResource = new TagResource(serviceFactory);
     this.openLineageResource = new OpenLineageResource(serviceFactory, openLineageDao);
     this.searchResource = new SearchResource(searchDao);

api/src/main/java/marquez/api/JobResource.java

@@ -15,6 +15,7 @@
 import java.util.List;
 import javax.validation.Valid;
 import javax.validation.constraints.Min;
+import javax.validation.constraints.NotNull;
 import javax.ws.rs.Consumes;
 import javax.ws.rs.DELETE;
 import javax.ws.rs.DefaultValue;
@@ -37,11 +38,14 @@
 import marquez.api.exceptions.JobVersionNotFoundException;
 import marquez.api.models.JobVersion;
 import marquez.api.models.ResultsPage;
+import marquez.common.models.FacetType;
 import marquez.common.models.JobName;
 import marquez.common.models.NamespaceName;
 import marquez.common.models.RunId;
 import marquez.common.models.Version;
+import marquez.db.JobFacetsDao;
 import marquez.db.JobVersionDao;
+import marquez.db.RunFacetsDao;
 import marquez.db.models.JobRow;
 import marquez.service.ServiceFactory;
 import marquez.service.models.Job;
@@ -52,11 +56,18 @@
 @Path("/api/v1")
 public class JobResource extends BaseResource {
   private final JobVersionDao jobVersionDao;
+  private final JobFacetsDao jobFacetsDao;
+  private final RunFacetsDao runFacetsDao;
 
   public JobResource(
-      @NonNull final ServiceFactory serviceFactory, @NonNull final JobVersionDao jobVersionDao) {
+      @NonNull final ServiceFactory serviceFactory,
+      @NonNull final JobVersionDao jobVersionDao,
+      @NonNull JobFacetsDao jobFacetsDao,
+      @NonNull RunFacetsDao runFacetsDao) {
     super(serviceFactory);
     this.jobVersionDao = jobVersionDao;
+    this.jobFacetsDao = jobFacetsDao;
+    this.runFacetsDao = runFacetsDao;
   }
 
   /**
@@ -236,6 +247,33 @@ public RunResource runResourceRoot(@PathParam("id") RunId runId) {
     return new RunResource(runId, runService);
   }
 
+  @Timed
+  @ResponseMetered
+  @ExceptionMetered
+  @GET
+  @Produces(APPLICATION_JSON)
+  @Path("/jobs/runs/{id}/facets")
+  public Response getRunFacets(
+      @PathParam("id") RunId runId, @QueryParam("type") @NotNull FacetType type) {
+    throwIfNotExists(runId);
+    Object facets = null;
+    switch (type) {
+      case JOB:
+        facets = jobFacetsDao.findJobFacetsByRunUuid(runId.getValue());
+        break;
+      case RUN:
+        facets = runFacetsDao.findRunFacetsByRunUuid(runId.getValue());
+        break;
+      case DATASET:
+        // for future case if there's a need to add dataset facets to the endpoint
+        break;
+      default:
+        break;
+    }
+
+    return Response.ok(facets).build();
+  }
+
   @Value
   static class JobVersions {
     @NonNull

api/src/main/java/marquez/api/models/JobVersion.java

@@ -5,7 +5,6 @@
 
 package marquez.api.models;
 
-import com.google.common.collect.ImmutableMap;
 import java.net.URL;
 import java.time.Instant;
 import java.util.List;
@@ -35,7 +34,6 @@ public final class JobVersion {
   @Getter private final Version version;
   @Getter private final NamespaceName namespace;
   @Nullable private final URL location;
-  @Getter private final ImmutableMap<String, String> context;
   @Getter private final List<DatasetId> inputs;
   @Getter private final List<DatasetId> outputs;
   @Getter @Nullable private final Run latestRun;
@@ -46,7 +44,6 @@ public JobVersion(
       @NonNull final Instant createdAt,
       @NonNull final Version version,
       @Nullable final URL location,
-      @Nullable final ImmutableMap<String, String> context,
       List<DatasetId> inputs,
       List<DatasetId> outputs,
       @Nullable Run latestRun) {
@@ -56,7 +53,6 @@ public JobVersion(
     this.version = version;
     this.namespace = id.getNamespace();
     this.location = location;
-    this.context = (context == null) ? ImmutableMap.of() : context;
     this.inputs = inputs;
     this.outputs = outputs;
     this.latestRun = latestRun;

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

@@ -20,7 +20,6 @@
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
 import com.google.common.base.Joiner;
-import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.hash.Hashing;
 import io.dropwizard.jackson.Jackson;
@@ -236,7 +235,6 @@ public static Instant toInstant(@Nullable final String asIso) {
    * @param jobName The name of the job.
    * @param jobInputIds The input dataset IDs for the job.
    * @param jobOutputIds The output dataset IDs for the job.
-   * @param jobContext The context of the job.
    * @param jobLocation The source code location for the job.
    * @return A {@link Version} object based on the specified job meta.
    */
@@ -245,7 +243,6 @@ public static Version newJobVersionFor(
       @NonNull final JobName jobName,
       @NonNull final ImmutableSet<DatasetId> jobInputIds,
       @NonNull final ImmutableSet<DatasetId> jobOutputIds,
-      @NonNull final ImmutableMap<String, String> jobContext,
       @Nullable final String jobLocation) {
     final byte[] bytes =
         VERSION_JOINER
@@ -268,8 +265,7 @@ public static Version newJobVersionFor(
                                 jobOutputId.getNamespace().getValue(),
                                 jobOutputId.getName().getValue()))
                     .collect(joining(VERSION_DELIM)),
-                jobLocation,
-                KV_JOINER.join(jobContext))
+                jobLocation)
             .getBytes(UTF_8);
     return Version.of(UUID.nameUUIDFromBytes(bytes));
   }

api/src/main/java/marquez/common/models/DatasetId.java

@@ -7,7 +7,6 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.collect.ComparisonChain;
-import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import lombok.EqualsAndHashCode;
 import lombok.NonNull;
@@ -16,8 +15,8 @@
 
 /**
  * ID for {@code Dataset}. The class implements {@link Comparable} to ensure job versions generated
- * with {@link Utils#newJobVersionFor(NamespaceName, JobName, ImmutableSet, ImmutableSet,
- * ImmutableMap, String)} are consistent as jobs may contain inputs and outputs out of order.
+ * with {@link Utils#newJobVersionFor(NamespaceName, JobName, ImmutableSet, ImmutableSet, String)}
+ * are consistent as jobs may contain inputs and outputs out of order.
  */
 @EqualsAndHashCode
 @ToString

api/src/main/java/marquez/common/models/FacetType.java

@@ -0,0 +1,12 @@
+/*
+ * Copyright 2018-2023 contributors to the Marquez project
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package marquez.common.models;
+
+public enum FacetType {
+  RUN,
+  JOB,
+  DATASET;
+}

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

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

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

@@ -103,13 +103,9 @@ private Columns() {}
 
   /* JOB VERSION ROW COLUMNS */
   public static final String JOB_UUID = "job_uuid";
-  public static final String JOB_CONTEXT_UUID = "job_context_uuid";
   public static final String LOCATION = "location";
   public static final String LATEST_RUN_UUID = "latest_run_uuid";
 
-  /* JOB CONTEXT ROW COLUMNS */
-  public static final String CONTEXT = "context";
-
   /* RUN ROW COLUMNS */
   public static final String EXTERNAL_ID = "external_id";
   public static final String RUN_ARGS_UUID = "run_args_uuid";

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

@@ -71,30 +71,8 @@ void updateLastModifiedAt(
 
   @SqlQuery(
       """
-      WITH selected_datasets AS (
-          SELECT d.*
-          FROM datasets_view d
-          WHERE CAST((:namespaceName, :datasetName) AS DATASET_NAME) = ANY(d.dataset_symlinks)
-      ), dataset_runs AS (
-          SELECT d.uuid, d.name, d.namespace_name, dv.run_uuid, dv.lifecycle_state, lineage_event_time, facet
-          FROM selected_datasets d
-          INNER JOIN dataset_versions AS dv ON dv.uuid = d.current_version_uuid
-          LEFT JOIN LATERAL (
-              SELECT run_uuid, lineage_event_time, facet FROM dataset_facets_view
-              WHERE dataset_uuid = dv.dataset_uuid
-          ) df ON df.run_uuid = dv.run_uuid
-          UNION
-          SELECT d.uuid, d.name, d.namespace_name, rim.run_uuid, lifecycle_state, lineage_event_time, facet
-          FROM selected_datasets d
-          INNER JOIN dataset_versions dv ON dv.uuid = d.current_version_uuid
-          LEFT JOIN runs_input_mapping rim ON dv.uuid = rim.dataset_version_uuid
-          LEFT JOIN LATERAL (
-              SELECT dataset_uuid, run_uuid, lineage_event_time, facet FROM dataset_facets_view
-              WHERE dataset_uuid = dv.dataset_uuid AND run_uuid = rim.run_uuid
-         ) df ON df.run_uuid = rim.run_uuid
-      )
       SELECT d.*, dv.fields, dv.lifecycle_state, sv.schema_location, t.tags, facets
-      FROM selected_datasets d
+      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 (
@@ -104,11 +82,15 @@ SELECT ARRAY_AGG(t.name) AS tags, m.dataset_uuid
           GROUP BY m.dataset_uuid
       ) t ON t.dataset_uuid = d.uuid
       LEFT JOIN (
-          SELECT d2.uuid AS dataset_uuid, JSONB_AGG(d2.facet ORDER BY d2.lineage_event_time ASC) AS facets
-          FROM dataset_runs AS d2
-          WHERE d2.run_uuid = d2.run_uuid AND d2.facet IS NOT NULL
-          GROUP BY d2.uuid
-      ) f ON f.dataset_uuid = d.uuid""")
+          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
+          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) {
@@ -137,32 +119,8 @@ default void setFields(Dataset ds) {
 
   @SqlQuery(
       """
-      WITH selected_datasets AS (
-          SELECT d.*
-          FROM datasets_view d
-          WHERE d.namespace_name = :namespaceName
-          ORDER BY d.name
-          LIMIT :limit OFFSET :offset
-      ), dataset_runs AS (
-          SELECT d.uuid, d.name, d.namespace_name, dv.run_uuid, dv.lifecycle_state, lineage_event_time, facet
-          FROM selected_datasets d
-          INNER JOIN dataset_versions dv ON dv.uuid = d.current_version_uuid
-          LEFT JOIN LATERAL (
-              SELECT run_uuid, lineage_event_time, facet FROM dataset_facets_view
-              WHERE dataset_uuid = dv.dataset_uuid
-          ) df ON df.run_uuid = dv.run_uuid
-          UNION
-          SELECT d.uuid, d.name, d.namespace_name, rim.run_uuid, lifecycle_state, lineage_event_time, facet
-          FROM selected_datasets d
-          INNER JOIN dataset_versions dv ON dv.uuid = d.current_version_uuid
-          LEFT JOIN runs_input_mapping rim ON dv.uuid = rim.dataset_version_uuid
-          LEFT JOIN LATERAL (
-              SELECT run_uuid, lineage_event_time, facet FROM dataset_facets_view
-              WHERE dataset_uuid = dv.dataset_uuid
-          ) df ON df.run_uuid = rim.run_uuid
-      )
       SELECT d.*, dv.fields, dv.lifecycle_state, sv.schema_location, t.tags, facets
-      FROM selected_datasets d
+      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 (
@@ -172,13 +130,17 @@ SELECT ARRAY_AGG(t.name) AS tags, m.dataset_uuid
           GROUP BY m.dataset_uuid
       ) t ON t.dataset_uuid = d.uuid
       LEFT JOIN (
-          SELECT d2.uuid AS dataset_uuid, JSONB_AGG(d2.facet ORDER BY d2.lineage_event_time ASC) AS facets
-          FROM dataset_runs AS d2
-          WHERE d2.run_uuid = d2.run_uuid
-          AND d2.facet IS NOT NULL
-          GROUP BY d2.uuid
-      ) f ON f.dataset_uuid = d.uuid
-      ORDER BY d.name""")
+          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
+          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")