Lineage events paging update

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

@@ -117,20 +117,25 @@ public Response getLineageEvents(
       @QueryParam("before") @DefaultValue("2030-01-01T00:00:00+00:00") ZonedDateTimeParam before,
       @QueryParam("after") @DefaultValue("1970-01-01T00:00:00+00:00") ZonedDateTimeParam after,
       @QueryParam("sortDirection") @DefaultValue("desc") SortDirection sortDirection,
-      @QueryParam("limit") @DefaultValue("100") @Min(value = 0) int limit) {
+      @QueryParam("limit") @DefaultValue("100") @Min(value = 0) int limit,
+      @QueryParam("offset") @DefaultValue("0") @Min(value = 0) int offset) {
     List<LineageEvent> events = Collections.emptyList();
     switch (sortDirection) {
       case DESC -> events =
-          openLineageDao.getAllLineageEventsDesc(before.get(), after.get(), limit);
-      case ASC -> events = openLineageDao.getAllLineageEventsAsc(before.get(), after.get(), limit);
+          openLineageDao.getAllLineageEventsDesc(before.get(), after.get(), limit, offset);
+      case ASC -> events =
+          openLineageDao.getAllLineageEventsAsc(before.get(), after.get(), limit, offset);
     }
-    return Response.ok(new Events(events)).build();
+    int totalCount = openLineageDao.getAllLineageTotalCount(before.get(), after.get());
+    return Response.ok(new Events(events, totalCount)).build();
   }
 
   @Value
   static class Events {
     @NonNull
     @JsonProperty("events")
     List<LineageEvent> value;
+
+    int totalCount;
   }
 }

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

@@ -103,8 +103,9 @@ void createLineageEvent(
   WHERE (le.event_time < :before
   AND le.event_time >= :after)
   ORDER BY le.event_time DESC
-  LIMIT :limit""")
-  List<LineageEvent> getAllLineageEventsDesc(ZonedDateTime before, ZonedDateTime after, int limit);
+  LIMIT :limit OFFSET :offset""")
+  List<LineageEvent> getAllLineageEventsDesc(
+      ZonedDateTime before, ZonedDateTime after, int limit, int offset);
 
   @SqlQuery(
       """
@@ -113,8 +114,17 @@ void createLineageEvent(
   WHERE (le.event_time < :before
   AND le.event_time >= :after)
   ORDER BY le.event_time ASC
-  LIMIT :limit""")
-  List<LineageEvent> getAllLineageEventsAsc(ZonedDateTime before, ZonedDateTime after, int limit);
+  LIMIT :limit OFFSET :offset""")
+  List<LineageEvent> getAllLineageEventsAsc(
+      ZonedDateTime before, ZonedDateTime after, int limit, int offset);
+
+  @SqlQuery(
+      """
+      SELECT count(*)
+      FROM lineage_events le
+      WHERE (le.event_time < :before
+      AND le.event_time >= :after)""")
+  int getAllLineageTotalCount(ZonedDateTime before, ZonedDateTime after);
 
   default UpdateLineageRow updateMarquezModel(LineageEvent event, ObjectMapper mapper) {
     UpdateLineageRow updateLineageRow = updateBaseMarquezModel(event, mapper);

spec/openapi.yml

@@ -91,6 +91,7 @@ paths:
         - $ref: '#/components/parameters/before'
         - $ref: '#/components/parameters/after'
         - $ref: '#/components/parameters/limit'
+        - $ref: '#/components/parameters/offset'
       summary: List all received OpenLineage events.
       description: Returns a list of OpenLineage events, sorted in direction of passed sort parameter. By default it is desc.
       tags:
@@ -945,6 +946,9 @@ components:
           type: array
           items:
             $ref: '#/components/schemas/LineageEvent'
+        totalCount:
+          type: number
+          description: The total number of events returned matching our conditions.
 
     CreatedSource:
       type: object