Skip to content

Commit c9271b8

Browse files
mobuchowskimobuchowski
committed
remove namespace endpoint
Signed-off-by: Maciej Obuchowski <obuchowski.maciej@gmail.com>
1 parent b31503c commit c9271b8

12 files changed

Lines changed: 202 additions & 143 deletions

File tree

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

Lines changed: 12 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,6 @@
2626
import javax.ws.rs.GET;
2727
import javax.ws.rs.POST;
2828
import javax.ws.rs.Path;
29-
import javax.ws.rs.PathParam;
3029
import javax.ws.rs.Produces;
3130
import javax.ws.rs.QueryParam;
3231
import javax.ws.rs.container.AsyncResponse;
@@ -35,6 +34,7 @@
3534
import lombok.NonNull;
3635
import lombok.Value;
3736
import lombok.extern.slf4j.Slf4j;
37+
import marquez.api.models.SortDirection;
3838
import marquez.db.OpenLineageDao;
3939
import marquez.service.ServiceFactory;
4040
import marquez.service.models.LineageEvent;
@@ -107,28 +107,21 @@ public Response getLineage(
107107
public Response getLineageEvents(
108108
@QueryParam("before") @DefaultValue("2030-01-01T00:00:00+00:00") ZonedDateTimeParam before,
109109
@QueryParam("after") @DefaultValue("1970-01-01T00:00:00+00:00") ZonedDateTimeParam after,
110+
@QueryParam("sortDirection") @DefaultValue("desc") SortDirection sortDirection,
110111
@QueryParam("limit") @DefaultValue("100") @Min(value = 0) int limit) {
111-
final List<LineageEvent> events =
112-
openLineageDao.getAllLineageEvents(before.get(), after.get(), limit);
112+
final List<LineageEvent> events;
113+
if (sortDirection.getValue().equalsIgnoreCase("desc")) {
114+
events = openLineageDao.getAllLineageEventsDesc(before.get(), after.get(), limit);
115+
} else if (sortDirection.getValue().equalsIgnoreCase("asc")) {
116+
events = openLineageDao.getAllLineageEventsAsc(before.get(), after.get(), limit);
117+
} else {
118+
return Response.status(BAD_REQUEST)
119+
.entity(String.format("%s should be either 'asc' or 'desc", sortDirection.getValue()))
120+
.build();
121+
}
113122
return Response.ok(new Events(events)).build();
114123
}
115124

116-
@Timed
117-
@ResponseMetered
118-
@ExceptionMetered
119-
@GET
120-
@Path("/namespace/{namespace}/events/lineage")
121-
@Produces(APPLICATION_JSON)
122-
public Response getLineageEventsByNamespace(
123-
@PathParam("namespace") String namespace,
124-
@QueryParam("before") @DefaultValue("2030-01-01T00:00:00+00:00") ZonedDateTimeParam before,
125-
@QueryParam("after") @DefaultValue("1970-01-01T00:00:00+00:00") ZonedDateTimeParam after,
126-
@QueryParam("limit") @DefaultValue("100") @Min(value = 0) int limit) {
127-
final List<LineageEvent> event =
128-
openLineageDao.getLineageEventsByNamespace(namespace, before.get(), after.get(), limit);
129-
return Response.ok(new Events(event)).build();
130-
}
131-
132125
@Value
133126
static class Events {
134127
@NonNull

‎api/src/main/java/marquez/api/models/SortDirection.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
package marquez.api.models;
2+
3+
import lombok.AllArgsConstructor;
4+
import lombok.Getter;
5+
6+
@AllArgsConstructor
7+
public enum SortDirection {
8+
DESC("desc"),
9+
ASC("asc");
10+
11+
@Getter public final String value;
12+
}

‎api/src/main/java/marquez/db/OpenLineageDao.java‎

Lines changed: 8 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,6 @@
5656
import marquez.service.models.LineageEvent.SchemaDatasetFacet;
5757
import marquez.service.models.LineageEvent.SchemaField;
5858
import org.jdbi.v3.sqlobject.config.RegisterRowMapper;
59-
import org.jdbi.v3.sqlobject.customizer.Bind;
6059
import org.jdbi.v3.sqlobject.statement.SqlQuery;
6160
import org.jdbi.v3.sqlobject.statement.SqlUpdate;
6261
import org.postgresql.util.PGobject;
@@ -98,46 +97,17 @@ void createLineageEvent(
9897
AND le.event_time >= :after)
9998
ORDER BY le.event_time DESC
10099
LIMIT :limit""")
101-
List<LineageEvent> getAllLineageEvents(ZonedDateTime before, ZonedDateTime after, int limit);
100+
List<LineageEvent> getAllLineageEventsDesc(ZonedDateTime before, ZonedDateTime after, int limit);
102101

103-
/**
104-
* This is a "hack" to get inputs/outputs namespace from jsonb column: <a
105-
* href="https://github.com/jdbi/jdbi/issues/1510#issuecomment-485423083">explanation</a>
106-
*/
107102
@SqlQuery(
108103
"""
109-
WITH job_events AS (
110-
SELECT le.event
111-
FROM lineage_events le
112-
WHERE le.job_namespace = :namespace
113-
AND (le.event_time < :before
114-
AND le.event_time >= :after)
115-
ORDER BY le.event_time DESC
116-
), dataset_events AS (
117-
SELECT le.event, le.event_time
118-
FROM lineage_events le
119-
JOIN dataset_versions dv on le.run_uuid = dv.run_uuid
120-
JOIN datasets ds on dv.dataset_uuid = ds.uuid
121-
JOIN namespaces n on ds.namespace_uuid = n.uuid
122-
WHERE n.name = :namespace
123-
AND (le.event_time < :before
124-
AND le.event_time >= :after)
125-
ORDER BY le.event_time DESC
126-
)
127-
SELECT le.event
128-
FROM (
129-
SELECT * FROM dataset_events
130-
UNION ALL
131-
SELECT * FROM job_events
132-
) le
133-
ORDER BY le.event_time
134-
LIMIT :limit
135-
""")
136-
List<LineageEvent> getLineageEventsByNamespace(
137-
@Bind("namespace") String namespace,
138-
@Bind("before") ZonedDateTime before,
139-
@Bind("after") ZonedDateTime after,
140-
@Bind("limit") int limit);
104+
SELECT event
105+
FROM lineage_events le
106+
WHERE (le.event_time < :before
107+
AND le.event_time >= :after)
108+
ORDER BY le.event_time ASC
109+
LIMIT :limit""")
110+
List<LineageEvent> getAllLineageEventsAsc(ZonedDateTime before, ZonedDateTime after, int limit);
141111

142112
default UpdateLineageRow updateMarquezModel(LineageEvent event, ObjectMapper mapper) {
143113
UpdateLineageRow updateLineageRow = updateBaseMarquezModel(event, mapper);

‎api/src/main/resources/marquez/db/migration/V47__add_lineage_event_indexes.sql‎

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,2 @@
11
CREATE INDEX CONCURRENTLY lineage_events_event_time
22
on lineage_events(event_time DESC);
3-
4-
CREATE INDEX CONCURRENTLY lineage_events_namespace_event_time
5-
on lineage_events(job_namespace, event_time DESC);

‎api/src/test/java/marquez/OpenLineageIntegrationTest.java‎

Lines changed: 79 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@
2727
import java.time.ZoneId;
2828
import java.time.ZonedDateTime;
2929
import java.time.temporal.ChronoField;
30+
import java.time.temporal.ChronoUnit;
3031
import java.util.Arrays;
3132
import java.util.Collections;
3233
import java.util.List;
@@ -36,6 +37,7 @@
3637
import java.util.concurrent.ExecutionException;
3738
import java.util.concurrent.TimeUnit;
3839
import java.util.concurrent.TimeoutException;
40+
import marquez.client.MarquezClient;
3941
import marquez.client.models.Dataset;
4042
import marquez.client.models.DatasetVersion;
4143
import marquez.client.models.Job;
@@ -414,7 +416,7 @@ public void testSendEventAndGetItBack() {
414416
}
415417

416418
@Test
417-
public void testFindEventByDatasetNamespace() {
419+
public void testFindEventIsSortedByTime() {
418420
marquez.service.models.LineageEvent.Run run =
419421
new marquez.service.models.LineageEvent.Run(
420422
UUID.randomUUID().toString(),
@@ -426,50 +428,44 @@ public void testFindEventByDatasetNamespace() {
426428
.build();
427429

428430
ZonedDateTime time = ZonedDateTime.now(ZoneId.of("UTC"));
431+
marquez.service.models.LineageEvent.Dataset dataset =
432+
marquez.service.models.LineageEvent.Dataset.builder()
433+
.namespace(NAMESPACE_NAME)
434+
.name(DB_TABLE_NAME)
435+
.build();
429436

430437
marquez.service.models.LineageEvent.LineageEventBuilder builder =
431438
marquez.service.models.LineageEvent.builder()
432-
.producer("testFindEventByDatasetNamespace")
433-
.eventType("COMPLETE")
439+
.producer("testFindEventIsSortedByTime")
434440
.run(run)
435441
.job(job)
436-
.eventTime(time)
437-
.inputs(Collections.emptyList());
442+
.inputs(Collections.emptyList())
443+
.outputs(Collections.singletonList(dataset));
438444

439-
for (int i = 0; i < 10; i++) {
440-
marquez.service.models.LineageEvent.Dataset dataset =
441-
marquez.service.models.LineageEvent.Dataset.builder()
442-
.namespace(String.format("namespace%d", i))
443-
.name(DB_TABLE_NAME)
444-
.build();
445+
marquez.service.models.LineageEvent firstEvent =
446+
builder.eventTime(time).eventType("START").build();
445447

446-
marquez.service.models.LineageEvent event =
447-
builder.outputs(Collections.singletonList(dataset)).build();
448+
CompletableFuture<Integer> resp = sendEvent(firstEvent);
449+
assertThat(resp.join()).isEqualTo(201);
448450

449-
final CompletableFuture<Integer> resp = sendEvent(event);
450-
assertThat(resp.join()).isEqualTo(201);
451-
}
451+
marquez.service.models.LineageEvent secondEvent =
452+
builder.eventTime(time.plusSeconds(10)).eventType("COMPLETE").build();
452453

453-
List<LineageEvent> rawEvents = client.listLineageEvents("namespace3");
454+
resp = sendEvent(secondEvent);
455+
assertThat(resp.join()).isEqualTo(201);
454456

455-
marquez.service.models.LineageEvent thirdEvent =
456-
builder
457-
.outputs(
458-
Collections.singletonList(
459-
marquez.service.models.LineageEvent.Dataset.builder()
460-
.namespace(String.format("namespace3"))
461-
.name(DB_TABLE_NAME)
462-
.build()))
463-
.build();
457+
List<LineageEvent> rawEvents = client.listLineageEvents();
464458

465-
assertThat(rawEvents.size()).isEqualTo(1);
459+
assertThat(rawEvents.size()).isEqualTo(2);
466460
ObjectMapper mapper = Utils.getMapper();
467-
assertThat((JsonNode) mapper.valueToTree(thirdEvent))
461+
assertThat((JsonNode) mapper.valueToTree(firstEvent))
462+
.isEqualTo(mapper.valueToTree(rawEvents.get(1)));
463+
assertThat((JsonNode) mapper.valueToTree(secondEvent))
468464
.isEqualTo(mapper.valueToTree(rawEvents.get(0)));
469465
}
470466

471467
@Test
472-
public void testFindEventIsSortedByTime() {
468+
public void testFindEventIsSortedByTimeAsc() {
473469
marquez.service.models.LineageEvent.Run run =
474470
new marquez.service.models.LineageEvent.Run(
475471
UUID.randomUUID().toString(),
@@ -507,12 +503,65 @@ public void testFindEventIsSortedByTime() {
507503
resp = sendEvent(secondEvent);
508504
assertThat(resp.join()).isEqualTo(201);
509505

510-
List<LineageEvent> rawEvents = client.listLineageEvents(NAMESPACE_NAME);
506+
List<LineageEvent> rawEvents = client.listLineageEvents(MarquezClient.SortDirection.ASC, 10);
511507

512508
assertThat(rawEvents.size()).isEqualTo(2);
513509
ObjectMapper mapper = Utils.getMapper();
514510
assertThat((JsonNode) mapper.valueToTree(firstEvent))
511+
.isEqualTo(mapper.valueToTree(rawEvents.get(0)));
512+
assertThat((JsonNode) mapper.valueToTree(secondEvent))
515513
.isEqualTo(mapper.valueToTree(rawEvents.get(1)));
514+
}
515+
516+
@Test
517+
public void testFindEventBeforeAfterTime() {
518+
marquez.service.models.LineageEvent.Run run =
519+
new marquez.service.models.LineageEvent.Run(
520+
UUID.randomUUID().toString(),
521+
marquez.service.models.LineageEvent.RunFacet.builder().build());
522+
marquez.service.models.LineageEvent.Job job =
523+
marquez.service.models.LineageEvent.Job.builder()
524+
.namespace(NAMESPACE_NAME)
525+
.name(JOB_NAME)
526+
.build();
527+
528+
ZonedDateTime after = ZonedDateTime.of(2021, 1, 1, 0, 0, 0, 0, ZoneId.of("UTC"));
529+
ZonedDateTime before = ZonedDateTime.of(2022, 1, 1, 0, 0, 0, 0, ZoneId.of("UTC"));
530+
531+
marquez.service.models.LineageEvent.Dataset dataset =
532+
marquez.service.models.LineageEvent.Dataset.builder()
533+
.namespace(NAMESPACE_NAME)
534+
.name(DB_TABLE_NAME)
535+
.build();
536+
537+
marquez.service.models.LineageEvent.LineageEventBuilder builder =
538+
marquez.service.models.LineageEvent.builder()
539+
.producer("testFindEventIsSortedByTime")
540+
.run(run)
541+
.job(job)
542+
.inputs(Collections.emptyList())
543+
.outputs(Collections.singletonList(dataset));
544+
545+
marquez.service.models.LineageEvent firstEvent =
546+
builder.eventTime(after.minus(1, ChronoUnit.YEARS)).eventType("START").build();
547+
548+
CompletableFuture<Integer> resp = sendEvent(firstEvent);
549+
assertThat(resp.join()).isEqualTo(201);
550+
551+
marquez.service.models.LineageEvent secondEvent =
552+
builder.eventTime(after.plusSeconds(10)).eventType("COMPLETE").build();
553+
554+
resp = sendEvent(secondEvent);
555+
assertThat(resp.join()).isEqualTo(201);
556+
557+
marquez.service.models.LineageEvent thirdEvent =
558+
builder.eventTime(before.plusSeconds(10)).eventType("COMPLETE").build();
559+
560+
List<LineageEvent> rawEvents =
561+
client.listLineageEvents(MarquezClient.SortDirection.ASC, before, after, 10);
562+
563+
assertThat(rawEvents.size()).isEqualTo(1);
564+
ObjectMapper mapper = Utils.getMapper();
516565
assertThat((JsonNode) mapper.valueToTree(secondEvent))
517566
.isEqualTo(mapper.valueToTree(rawEvents.get(0)));
518567
}

‎api/src/test/java/marquez/api/OpenLineageResourceTest.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
import io.dropwizard.testing.junit5.DropwizardExtensionsSupport;
1717
import io.dropwizard.testing.junit5.ResourceExtension;
1818
import java.util.Map;
19+
import javax.ws.rs.core.Response;
1920
import marquez.common.Utils;
2021
import marquez.db.OpenLineageDao;
2122
import marquez.service.LineageService;
@@ -63,4 +64,16 @@ public void testGetLineage() {
6364

6465
assertEquals(lineage, LINEAGE);
6566
}
67+
68+
@Test
69+
public void testGetLineageEventsBadSort() {
70+
final Response response =
71+
UNDER_TEST
72+
.target("/api/v1/events/lineage")
73+
.queryParam("sortDirection", "asdf")
74+
.request()
75+
.get();
76+
77+
assertEquals(response.getStatus(), 400);
78+
}
6679
}

‎clients/java/src/main/java/marquez/client/MarquezClient.java‎

Lines changed: 15 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -20,11 +20,13 @@
2020
import java.io.InputStream;
2121
import java.net.URL;
2222
import java.time.Instant;
23+
import java.time.ZonedDateTime;
2324
import java.util.List;
2425
import java.util.Properties;
2526
import java.util.Set;
2627
import javax.annotation.Nullable;
2728
import javax.net.ssl.SSLContext;
29+
import lombok.AllArgsConstructor;
2830
import lombok.EqualsAndHashCode;
2931
import lombok.Getter;
3032
import lombok.NoArgsConstructor;
@@ -89,21 +91,26 @@ public MarquezClient(final URL baseUrl, @Nullable final String apiKey) {
8991
}
9092

9193
public List<LineageEvent> listLineageEvents() {
92-
return listLineageEvents(DEFAULT_LIMIT, DEFAULT_OFFSET);
94+
return listLineageEvents(SortDirection.DESC, DEFAULT_LIMIT);
9395
}
9496

95-
public List<LineageEvent> listLineageEvents(int limit, int offset) {
96-
final String bodyAsJson = http.get(url.toEventUrl(limit, offset));
97+
public List<LineageEvent> listLineageEvents(MarquezClient.SortDirection sort, int limit) {
98+
final String bodyAsJson = http.get(url.toEventUrl(sort, limit));
9799
return Events.fromJson(bodyAsJson).getValue();
98100
}
99101

100-
public List<LineageEvent> listLineageEvents(String namespaceName) {
101-
return listLineageEvents(namespaceName, DEFAULT_LIMIT, DEFAULT_OFFSET);
102+
public List<LineageEvent> listLineageEvents(
103+
MarquezClient.SortDirection sort, ZonedDateTime before, ZonedDateTime after, int limit) {
104+
final String bodyAsJson = http.get(url.toEventUrl(sort, before, after, limit));
105+
return Events.fromJson(bodyAsJson).getValue();
102106
}
103107

104-
public List<LineageEvent> listLineageEvents(String namespaceName, int limit, int offset) {
105-
final String bodyAsJson = http.get(url.toEventUrl(namespaceName, limit, offset));
106-
return Events.fromJson(bodyAsJson).getValue();
108+
@AllArgsConstructor
109+
public enum SortDirection {
110+
DESC("desc"),
111+
ASC("asc");
112+
113+
@Getter public final String value;
107114
}
108115

109116
public Namespace createNamespace(

‎clients/java/src/main/java/marquez/client/MarquezPathV1.java‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -76,10 +76,6 @@ static String lineageEventPath() {
7676
return path("/events/lineage");
7777
}
7878

79-
static String lineageEventPath(String namespaceName) {
80-
return path("/namespace/%s/events/lineage", namespaceName);
81-
}
82-
8379
static String sourcePath(String sourceName) {
8480
return path("/sources/%s", sourceName);
8581
}

0 commit comments

Comments
 (0)