Merge branch 'MarquezProject:main' into feature/add-delete-end-point-…

…for-column-tags

.circleci/db-migration.sh

@@ -13,7 +13,7 @@
 # Version of PostgreSQL
 readonly POSTGRES_VERSION="14"
 # Version of Marquez
-readonly MARQUEZ_VERSION=0.43.0
+readonly MARQUEZ_VERSION=0.43.1
 # 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.43.0
+TAG=0.43.1

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

@@ -204,7 +204,7 @@ ExtendedJobVersionRow upsertJobVersion(
     INSERT INTO job_versions_io_mapping (
       job_version_uuid, dataset_uuid, io_type, job_uuid, job_symlink_target_uuid, is_current_job_version, made_current_at)
     VALUES (:jobVersionUuid, :datasetUuid, :ioType, :jobUuid, :symlinkTargetJobUuid, TRUE, NOW())
-    ON CONFLICT (job_version_uuid, dataset_uuid, io_type, job_uuid) DO NOTHING
+    ON CONFLICT (job_version_uuid, dataset_uuid, io_type, job_uuid) DO UPDATE SET is_current_job_version = TRUE
   """)
   void upsertCurrentInputOrOutputDatasetFor(
       UUID jobVersionUuid,

api/src/test/java/marquez/db/LineageTestUtils.java

@@ -116,13 +116,48 @@ public static UpdateLineageRow createLineageRow(
       List<Dataset> outputs,
       @Valid LineageEvent.ParentRunFacet parentRunFacet,
       ImmutableMap<String, Object> runFacets) {
+    return createLineageRow(
+        dao,
+        jobName,
+        UUID.randomUUID(),
+        status,
+        jobFacet,
+        inputs,
+        outputs,
+        parentRunFacet,
+        runFacets);
+  }
+
+  /**
+   * Create an {@link UpdateLineageRow} from the input job details and datasets.
+   *
+   * @param dao
+   * @param jobName
+   * @param runId
+   * @param status
+   * @param jobFacet
+   * @param inputs
+   * @param outputs
+   * @param parentRunFacet
+   * @param runFacets
+   * @return
+   */
+  public static UpdateLineageRow createLineageRow(
+      OpenLineageDao dao,
+      String jobName,
+      UUID runId,
+      String status,
+      JobFacet jobFacet,
+      List<Dataset> inputs,
+      List<Dataset> outputs,
+      @Valid LineageEvent.ParentRunFacet parentRunFacet,
+      ImmutableMap<String, Object> runFacets) {
     NominalTimeRunFacet nominalTimeRunFacet = new NominalTimeRunFacet();
     nominalTimeRunFacet.setNominalStartTime(
         Instant.now().atZone(LOCAL_ZONE).truncatedTo(ChronoUnit.HOURS));
     nominalTimeRunFacet.setNominalEndTime(
         nominalTimeRunFacet.getNominalStartTime().plus(1, ChronoUnit.HOURS));
 
-    UUID runId = UUID.randomUUID();
     LineageEvent event =
         LineageEvent.builder()
             .eventType(status)

api/src/test/java/marquez/service/LineageServiceTest.java

@@ -16,6 +16,7 @@
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Optional;
+import java.util.UUID;
 import java.util.stream.Collectors;
 import marquez.api.JdbiUtils;
 import marquez.common.models.DatasetId;
@@ -56,6 +57,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
 
 @ExtendWith(MarquezJdbiExternalPostgresExtension.class)
 public class LineageServiceTest {
@@ -427,6 +429,77 @@ public void testLineageWithWithCycle() {
         .matches(n -> n.isJobType() && n.asJobId().getName().getValue().equals("writeJob"));
   }
 
+  @Test
+  public void testGetLineageJobRunTwice() {
+    Dataset input = Dataset.builder().name("input-dataset").namespace(NAMESPACE).build();
+    Dataset output = Dataset.builder().name("output-dataset").namespace(NAMESPACE).build();
+    UUID runId = UUID.randomUUID();
+
+    // (1) Run batch job which outputs input-dataset
+    LineageTestUtils.createLineageRow(
+        openLineageDao,
+        "someJob",
+        runId,
+        "START",
+        jobFacet,
+        Arrays.asList(input),
+        Collections.emptyList(),
+        null,
+        ImmutableMap.of());
+
+    LineageTestUtils.createLineageRow(
+        openLineageDao,
+        "someJob",
+        runId,
+        "COMPLETE",
+        jobFacet,
+        Collections.emptyList(),
+        Arrays.asList(output),
+        null,
+        ImmutableMap.of());
+
+    // (2) Rerun it
+    LineageTestUtils.createLineageRow(
+        openLineageDao,
+        "someJob",
+        runId,
+        "START",
+        jobFacet,
+        Arrays.asList(input),
+        Collections.emptyList(),
+        null,
+        ImmutableMap.of());
+
+    LineageTestUtils.createLineageRow(
+        openLineageDao,
+        "someJob",
+        runId,
+        "COMPLETE",
+        jobFacet,
+        Collections.emptyList(),
+        Arrays.asList(output),
+        null,
+        ImmutableMap.of());
+
+    // (4) lineage on output dataset shall be same as lineage on input dataset
+    Lineage lineageFromInput =
+        lineageService.lineage(
+            NodeId.of(
+                new DatasetId(new NamespaceName(NAMESPACE), new DatasetName("input-dataset"))),
+            5,
+            true);
+
+    Lineage lineageFromOutput =
+        lineageService.lineage(
+            NodeId.of(
+                new DatasetId(new NamespaceName(NAMESPACE), new DatasetName("output-dataset"))),
+            5,
+            true);
+
+    assertThat(lineageFromInput.getGraph()).hasSize(3); // 2 datasets + 1 job
+    assertThat(lineageFromInput.getGraph()).isEqualTo(lineageFromOutput.getGraph());
+  }
+
   @Test
   public void testGetLineageForRunningStreamingJob() {
     Dataset input = Dataset.builder().name("input-dataset").namespace(NAMESPACE).build();

chart/Chart.yaml

@@ -29,4 +29,4 @@ name: marquez
 sources:
   - https://github.com/MarquezProject/marquez
   - https://marquezproject.github.io/marquez/
-version: 0.43.0
+version: 0.43.1

chart/values.yaml

@@ -17,7 +17,7 @@ marquez:
   image:
     registry: docker.io
     repository: marquezproject/marquez
-    tag: 0.43.0
+    tag: 0.43.1
     pullPolicy: IfNotPresent
   ## Name of the existing secret containing credentials for the Marquez installation.
   ## When this is specified, it will take precedence over the values configured in the 'db' section.
@@ -75,7 +75,7 @@ web:
   image:
     registry: docker.io
     repository: marquezproject/marquez-web
-    tag: 0.43.0
+    tag: 0.43.1
     pullPolicy: IfNotPresent
   ## Marquez website will run on this port
   ##

clients/java/README.md

@@ -10,14 +10,14 @@ Maven:
 <dependency>
     <groupId>io.github.marquezproject</groupId>
     <artifactId>marquez-java</artifactId>
-    <version>0.43.0</version>
+    <version>0.43.1</version>
 </dependency>
 ```
 
 or Gradle:
 
 ```groovy
-implementation 'io.github.marquezproject:marquez-java:0.43.0
+implementation 'io.github.marquezproject:marquez-java:0.43.1
 ```
 
 ## Usage

clients/python/marquez_client/__init__.py

@@ -4,7 +4,7 @@
 # -*- coding: utf-8 -*-
 
 __author__ = """Marquez Project"""
-__version__ = "0.44.0"
+__version__ = "0.43.1"
 
 from marquez_client.client import MarquezClient              # noqa: F401
 from marquez_client.clients import Clients                   # noqa: F401

clients/python/setup.cfg

@@ -1,9 +1,9 @@
 [bumpversion]
-current_version = 0.44.0
+current_version = 0.43.1
 commit = False
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(?P<rc>.*)
-serialize = 
+serialize =
 	{major}.{minor}.{patch}{rc}
 	{major}.{minor}.{patch}
 

clients/python/setup.py

@@ -24,7 +24,7 @@
 
 setup(
     name="marquez-python",
-    version="0.44.0",
+    version="0.43.1",
     description="Marquez Python Client",
     long_description=readme,
     long_description_content_type="text/markdown",

docker/up.sh

@@ -8,9 +8,9 @@
 set -e
 
 # Version of Marquez
-readonly VERSION=0.43.0
+readonly VERSION=0.43.1
 # Build version of Marquez
-readonly BUILD_VERSION=0.43.0
+readonly BUILD_VERSION=0.43.1
 
 title() {
   echo -e "\033[1m${1}\033[0m"