deletes: 'undelete' job on subsequent OL event

Signed-off-by: Maciej Obuchowski <obuchowski.maciej@gmail.com>

Author: mobuchowski

api/src/main/resources/marquez/db/migration/R__1_Jobs_view_and_rewrite_function.sql

@@ -35,7 +35,7 @@ BEGIN
     INSERT INTO jobs (uuid, type, created_at, updated_at, namespace_uuid, name, description,
                       current_version_uuid, namespace_name, current_job_context_uuid,
                       current_location, current_inputs, symlink_target_uuid, parent_job_uuid,
-                      parent_job_uuid_string)
+                      parent_job_uuid_string, is_hidden)
     SELECT NEW.uuid,
            NEW.type,
            NEW.created_at,
@@ -50,7 +50,8 @@ BEGIN
            NEW.current_inputs,
            NEW.symlink_target_uuid,
            NEW.parent_job_uuid,
-           COALESCE(NEW.parent_job_uuid::char(36), '')
+           COALESCE(NEW.parent_job_uuid::char(36), ''),
+           false
     ON CONFLICT (name, namespace_uuid, parent_job_uuid_string)
         DO UPDATE SET updated_at               = EXCLUDED.updated_at,
                       type                     = EXCLUDED.type,
@@ -60,8 +61,9 @@ BEGIN
                       current_inputs           = EXCLUDED.current_inputs,
                       -- update the symlink target if null. otherwise, keep the old value
                       symlink_target_uuid      = COALESCE(jobs.symlink_target_uuid,
-                              EXCLUDED.symlink_target_uuid)
-    -- the SELECT statement below will get the OLD symlink_target_uuid in case of update and the NEW
+                                                          EXCLUDED.symlink_target_uuid),
+                      is_hidden                = false
+                      -- the SELECT statement below will get the OLD symlink_target_uuid in case of update and the NEW
     -- version in case of insert
     RETURNING uuid, symlink_target_uuid, (SELECT symlink_target_uuid FROM jobs j2 WHERE j2.uuid=jobs.uuid)
         INTO job_uuid, new_symlink_target_uuid, old_symlink_target_uuid;

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

@@ -70,7 +70,11 @@ public class OpenLineageServiceIntegrationTest {
   public static final ZoneId TIMEZONE = ZoneId.of("America/Los_Angeles");
   public static final String DATASET_NAME = "theDataset";
   private RunService runService;
+
+  private JobService jobService;
   private OpenLineageDao openLineageDao;
+
+  private JobDao jobDao;
   private DatasetDao datasetDao;
   private DatasetVersionDao datasetVersionDao;
   private ArgumentCaptor<JobInputUpdate> runInputListener;
@@ -134,7 +138,9 @@ public ExpectedResults(
   public void setup(Jdbi jdbi) throws SQLException {
     openLineageDao = jdbi.onDemand(OpenLineageDao.class);
     datasetVersionDao = jdbi.onDemand(DatasetVersionDao.class);
+    jobDao = jdbi.onDemand(JobDao.class);
     runService = mock(RunService.class);
+    jobService = new JobService(jobDao, runService);
     runInputListener = ArgumentCaptor.forClass(JobInputUpdate.class);
     doNothing().when(runService).notify(runInputListener.capture());
     runOutputListener = ArgumentCaptor.forClass(JobOutputUpdate.class);
@@ -146,19 +152,18 @@ public void setup(Jdbi jdbi) throws SQLException {
         jdbi.onDemand(NamespaceDao.class)
             .upsertNamespaceRow(UUID.randomUUID(), Instant.now(), NAMESPACE, "me");
     JobRow job =
-        jdbi.onDemand(JobDao.class)
-            .upsertJob(
-                UUID.randomUUID(),
-                JobType.BATCH,
-                Instant.now(),
-                namespace.getUuid(),
-                NAMESPACE,
-                "parentJob",
-                "description",
-                null,
-                null,
-                null,
-                null);
+        jobDao.upsertJob(
+            UUID.randomUUID(),
+            JobType.BATCH,
+            Instant.now(),
+            namespace.getUuid(),
+            NAMESPACE,
+            "parentJob",
+            "description",
+            null,
+            null,
+            null,
+            null);
     Map<String, String> runArgsMap = new HashMap<>();
     RunArgsRow argsRow =
         jdbi.onDemand(RunArgsDao.class)
@@ -382,6 +387,41 @@ public void testDatasetVersionUpdatedOnRunCompletion()
         .contains(dsVersion1Id);
   }
 
+  @Test
+  void testJobIsNotHiddenAfterSubsequentOLEvent() throws ExecutionException, InterruptedException {
+    String name = "aNotHiddenJob";
+
+    LineageEvent.LineageEventBuilder builder =
+        LineageEvent.builder()
+            .eventType("COMPLETE")
+            .job(LineageEvent.Job.builder().name(name).namespace(NAMESPACE).build())
+            .eventTime(Instant.now().atZone(TIMEZONE))
+            .inputs(Collections.emptyList())
+            .outputs(Collections.emptyList());
+
+    lineageService
+        .createAsync(
+            builder
+                .run(new LineageEvent.Run(UUID.randomUUID().toString(), RunFacet.builder().build()))
+                .build())
+        .get();
+
+    assertThat(jobService.findJobByName(NAMESPACE, name)).isNotEmpty();
+
+    jobService.delete(NAMESPACE, name);
+
+    assertThat(jobService.findJobByName(NAMESPACE, name)).isEmpty();
+
+    lineageService
+        .createAsync(
+            builder
+                .run(new LineageEvent.Run(UUID.randomUUID().toString(), RunFacet.builder().build()))
+                .build())
+        .get();
+
+    assertThat(jobService.findJobByName(NAMESPACE, name)).isNotEmpty();
+  }
+
   private void checkExists(LineageEvent.Dataset ds) {
     DatasetService datasetService = new DatasetService(openLineageDao, runService);