point-in-timea for column-level lineage (#2265)

Signed-off-by: Pawel Leszczynski <[email protected]>

Signed-off-by: Pawel Leszczynski <[email protected]>

CHANGELOG.md

@@ -1,6 +1,8 @@
 # Changelog
 
 ## [Unreleased](https://github.com/MarquezProject/marquez/compare/0.28.0...HEAD)
+* Column-lineage endpoints supports point-in-time requests [`#2265`](https://github.com/MarquezProject/marquez/pull/2265) [@pawel-big-lebowski](https://github.com/pawel-big-lebowski)
+ *Enable requesting `column-lineage` endpoint by a dataset version, job version or dataset field of a specific dataset version.*
 
 ## [0.28.0](https://github.com/MarquezProject/marquez/compare/0.27.0...0.28.0) - 2022-11-21
 

api/src/main/java/marquez/api/ColumnLineageResource.java

@@ -10,7 +10,6 @@
 import com.codahale.metrics.annotation.ExceptionMetered;
 import com.codahale.metrics.annotation.ResponseMetered;
 import com.codahale.metrics.annotation.Timed;
-import java.time.Instant;
 import java.util.concurrent.ExecutionException;
 import javax.validation.constraints.NotNull;
 import javax.ws.rs.DefaultValue;
@@ -44,7 +43,10 @@ public Response getLineage(
  @QueryParam("depth") @DefaultValue(DEFAULT_DEPTH) int depth,
  @QueryParam("withDownstream") @DefaultValue("false") boolean withDownstream)
  throws ExecutionException, InterruptedException {
- return Response.ok(columnLineageService.lineage(nodeId, depth, withDownstream, Instant.now()))
- .build();
+ if (nodeId.hasVersion() && withDownstream) {
+ return Response.status(400, "Node version cannot be specified when withDownstream is true")
+ .build();
+ }
+ return Response.ok(columnLineageService.lineage(nodeId, depth, withDownstream)).build();
  }
 }

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

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018-2022 contributors to the Marquez project
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package marquez.common.models;
+
+import java.util.UUID;
+import lombok.AllArgsConstructor;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.ToString;
+
+/** ID for {@code DatasetField} with a version of {@code Dataset}. */
+@EqualsAndHashCode
+@AllArgsConstructor
+@ToString
+public class DatasetFieldVersionId {
+
+ @Getter private final DatasetId datasetId;
+ @Getter private final FieldName fieldName;
+ @Getter private final UUID version;
+
+ public static DatasetFieldVersionId of(
+ String namespace, String datasetName, String field, UUID version) {
+ return new DatasetFieldVersionId(
+ new DatasetId(NamespaceName.of(namespace), DatasetName.of(datasetName)),
+ FieldName.of(field),
+ version);
+ }
+}

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

@@ -19,4 +19,9 @@ public class JobVersionId {
  @NonNull NamespaceName namespace;
  @NonNull JobName name;
  @NonNull UUID version;
+
+ public static JobVersionId of(
+ final NamespaceName namespaceName, final JobName jobName, final UUID version) {
+ return new JobVersionId(namespaceName, jobName, version);
+ }
 }

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

@@ -146,7 +146,8 @@ WHERE output_dataset_field_uuid IN (<datasetFieldUuids>)
  output_fields.dataset_name,
  output_fields.field_name,
  output_fields.type,
- ARRAY_AGG(DISTINCT ARRAY[input_fields.namespace_name, input_fields.dataset_name, input_fields.field_name]) AS inputFields,
+ ARRAY_AGG(DISTINCT ARRAY[input_fields.namespace_name, input_fields.dataset_name, CAST(clr.input_dataset_version_uuid AS VARCHAR), input_fields.field_name]) AS inputFields,
+ clr.output_dataset_version_uuid as dataset_version_uuid,
  clr.transformation_description,
  clr.transformation_type,
  clr.created_at,
@@ -160,6 +161,7 @@ WHERE output_dataset_field_uuid IN (<datasetFieldUuids>)
  output_fields.dataset_name,
  output_fields.field_name,
  output_fields.type,
+ clr.output_dataset_version_uuid,
  clr.transformation_description,
  clr.transformation_type,
  clr.created_at,
@@ -191,7 +193,8 @@ dataset_fields_view AS (
  output_fields.dataset_name,
  output_fields.field_name,
  output_fields.type,
- ARRAY_AGG(ARRAY[input_fields.namespace_name, input_fields.dataset_name, input_fields.field_name]) AS inputFields,
+ ARRAY_AGG(DISTINCT ARRAY[input_fields.namespace_name, input_fields.dataset_name, CAST(c.input_dataset_version_uuid AS VARCHAR), input_fields.field_name]) AS inputFields,
+ c.output_dataset_version_uuid as dataset_version_uuid,
  c.transformation_description,
  c.transformation_type,
  c.created_at,
@@ -204,6 +207,7 @@ dataset_fields_view AS (
  output_fields.dataset_name,
  output_fields.field_name,
  output_fields.type,
+ c.output_dataset_version_uuid,
  c.transformation_description,
  c.transformation_type,
  c.created_at,

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

@@ -17,12 +17,14 @@
 import marquez.db.mappers.DatasetFieldMapper;
 import marquez.db.mappers.DatasetFieldRowMapper;
 import marquez.db.mappers.FieldDataMapper;
+import marquez.db.mappers.PairUuidInstantMapper;
 import marquez.db.models.DatasetFieldRow;
 import marquez.db.models.DatasetRow;
 import marquez.db.models.InputFieldData;
 import marquez.db.models.TagRow;
 import marquez.service.models.Dataset;
 import marquez.service.models.DatasetVersion;
+import org.apache.commons.lang3.tuple.Pair;
 import org.jdbi.v3.sqlobject.config.RegisterRowMapper;
 import org.jdbi.v3.sqlobject.customizer.BindBean;
 import org.jdbi.v3.sqlobject.statement.SqlBatch;
@@ -32,6 +34,7 @@
 @RegisterRowMapper(DatasetFieldRowMapper.class)
 @RegisterRowMapper(DatasetFieldMapper.class)
 @RegisterRowMapper(FieldDataMapper.class)
+@RegisterRowMapper(PairUuidInstantMapper.class)
 public interface DatasetFieldDao extends BaseDao {
  @SqlQuery(
  """
@@ -98,13 +101,26 @@ default Dataset updateTags(
 
  @SqlQuery(
  """
- SELECT df.uuid
- FROM dataset_fields  df
+ SELECT df.uuid, max(dv.created_at)
+ FROM dataset_fields df
  JOIN datasets_view AS d ON d.uuid = df.dataset_uuid
+ JOIN dataset_versions_field_mapping AS fm ON fm.dataset_field_uuid = df.uuid
+ JOIN dataset_versions AS dv ON dv.uuid = fm.dataset_version_uuid
  WHERE CAST((:namespaceName, :datasetName) AS DATASET_NAME) = ANY(d.dataset_symlinks)
+ GROUP BY df.uuid
  """)
  List<UUID> findDatasetFieldsUuids(String namespaceName, String datasetName);
 
+ @SqlQuery(
+ """
+ SELECT df.uuid, dv.created_at
+ FROM dataset_fields df
+ JOIN dataset_versions_field_mapping AS fm ON fm.dataset_field_uuid = df.uuid
+ JOIN dataset_versions AS dv ON dv.uuid = :datasetVersion
+ WHERE fm.dataset_version_uuid = :datasetVersion
+ """)
+ List<Pair<UUID, Instant>> findDatasetVersionFieldsUuids(UUID datasetVersion);
+
  @SqlQuery(
  """
  WITH latest_run AS (
@@ -121,6 +137,15 @@ WITH latest_run AS (
  """)
  List<UUID> findFieldsUuidsByJob(String namespaceName, String jobName);
 
+ @SqlQuery(
+ """
+ SELECT dataset_fields.uuid, r.created_at
+ FROM dataset_fields
+ JOIN dataset_versions ON dataset_versions.dataset_uuid = dataset_fields.dataset_uuid
+ JOIN runs_view r ON r.job_version_uuid = :jobVersion
+ """)
+ List<Pair<UUID, Instant>> findFieldsUuidsByJobVersion(UUID jobVersion);
+
  @SqlQuery(
  """
  SELECT df.uuid
@@ -131,6 +156,17 @@ WHERE CAST((:namespaceName, :datasetName) AS DATASET_NAME) = ANY(d.dataset_symli
  """)
  Optional<UUID> findUuid(String namespaceName, String datasetName, String name);
 
+ @SqlQuery(
+ """
+ SELECT df.uuid, dv.created_at
+ FROM dataset_fields df
+ JOIN datasets_view AS d ON d.uuid = df.dataset_uuid
+ JOIN dataset_versions AS dv ON dv.uuid = :datasetVersion
+ JOIN dataset_versions_field_mapping AS fm ON fm.dataset_field_uuid = df.uuid
+ WHERE fm.dataset_version_uuid = :datasetVersion AND df.name = :fieldName
+ """)
+ List<Pair<UUID, Instant>> findDatasetVersionFieldsUuids(String fieldName, UUID datasetVersion);
+
  @SqlQuery(
  "SELECT f.*, "
  + "ARRAY(SELECT t.name "

api/src/main/java/marquez/db/mappers/ColumnLineageNodeDataMapper.java

@@ -9,12 +9,14 @@
 import static marquez.db.Columns.TRANSFORMATION_TYPE;
 import static marquez.db.Columns.stringOrNull;
 import static marquez.db.Columns.stringOrThrow;
+import static marquez.db.Columns.uuidOrThrow;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.collect.ImmutableList;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.Arrays;
+import java.util.UUID;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import marquez.common.Utils;
@@ -35,6 +37,7 @@ public ColumnLineageNodeData map(ResultSet results, StatementContext ctx) throws
  return new ColumnLineageNodeData(
  stringOrThrow(results, Columns.NAMESPACE_NAME),
  stringOrThrow(results, Columns.DATASET_NAME),
+ uuidOrThrow(results, Columns.DATASET_VERSION_UUID),
  stringOrThrow(results, Columns.FIELD_NAME),
  stringOrThrow(results, Columns.TYPE),
  stringOrNull(results, TRANSFORMATION_DESCRIPTION),
@@ -54,7 +57,7 @@ public static ImmutableList<InputFieldNodeData> toInputFields(ResultSet results,
  return ImmutableList.copyOf(
  Arrays.asList(deserializedArray).stream()
  .map(o -> (String[]) o)
- .map(arr -> new InputFieldNodeData(arr[0], arr[1], arr[2]))
+ .map(arr -> new InputFieldNodeData(arr[0], arr[1], UUID.fromString(arr[2]), arr[3]))
  .collect(Collectors.toList()));
  }
 }

api/src/main/java/marquez/db/mappers/PairUuidInstantMapper.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2018-2022 contributors to the Marquez project
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package marquez.db.mappers;
+
+import static marquez.db.Columns.timestampOrNull;
+import static marquez.db.Columns.uuidOrThrow;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.time.Instant;
+import java.util.UUID;
+import lombok.NonNull;
+import marquez.db.Columns;
+import org.apache.commons.lang3.tuple.Pair;
+import org.jdbi.v3.core.mapper.RowMapper;
+import org.jdbi.v3.core.statement.StatementContext;
+
+public final class PairUuidInstantMapper implements RowMapper<Pair<UUID, Instant>> {
+ @Override
+ public Pair<UUID, Instant> map(@NonNull ResultSet results, @NonNull StatementContext context)
+ throws SQLException {
+ return Pair.of(
+ uuidOrThrow(results, Columns.ROW_UUID), timestampOrNull(results, Columns.CREATED_AT));
+ }
+}

api/src/main/java/marquez/db/models/ColumnLineageNodeData.java

@@ -6,6 +6,8 @@
 package marquez.db.models;
 
 import java.util.List;
+import java.util.UUID;
+import javax.annotation.Nullable;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NonNull;
@@ -15,6 +17,7 @@
 public class ColumnLineageNodeData implements NodeData {
  @NonNull String namespace;
  @NonNull String dataset;
+ @Nullable UUID datasetVersion;
  @NonNull String field;
  @NonNull String fieldType;
  String transformationDescription;

api/src/main/java/marquez/db/models/InputFieldNodeData.java

@@ -5,14 +5,19 @@
 
 package marquez.db.models;
 
+import java.util.UUID;
+import javax.annotation.Nullable;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.NonNull;
+import lombok.ToString;
 
 @Getter
 @AllArgsConstructor
+@ToString
 public class InputFieldNodeData {
  @NonNull String namespace;
  @NonNull String dataset;
+ @Nullable UUID datasetVersion;
  @NonNull String field;
 }