Delta Sharing protocol: Query Table Changes (Change Data Feed)

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java

@@ -188,12 +188,7 @@ public DeltaTableQueryReader.Result query(
       QueryTableRequest request,
       DeltaTableQueryReader.ResponseOptions options) {
     ShareEntity granted = shares.requireGranted(recipient, share);
-    ResolvedAsset resolved = resolve(granted, schema, table);
-    List<String> accessModes = TableAccessModes.forTable(resolved);
-    if (accessModes == null || !accessModes.contains("url")) {
-      throw ApiException.invalidParameter(
-          "table '" + schema + "." + table + "' does not support url access");
-    }
+    ResolvedAsset resolved = requireUrlAccess(resolve(granted, schema, table), schema, table);
     QueryTableRequest body = request == null ? QueryTableRequest.EMPTY : request;
     if (body.startingVersion() != null || body.endingVersion() != null) {
       return queryChanges(body, resolved, granted, options);
@@ -223,6 +218,41 @@ public DeltaTableQueryReader.Result query(
         limitHint(body));
   }
 
+  /**
+   * The Change Data Feed of a table in the share that supports url access and records it, from
+   * {@code startingVersion} or an ISO-8601 {@code startingTimestamp} through {@code endingVersion},
+   * {@code endingTimestamp}, or the latest version, with presigned file URLs.
+   */
+  public DeltaTableQueryReader.Result changes(
+      RecipientPrincipal recipient,
+      String share,
+      String schema,
+      String table,
+      TableChangesRequest request,
+      DeltaTableQueryReader.ResponseOptions options) {
+    ShareEntity granted = shares.requireGranted(recipient, share);
+    ResolvedAsset resolved = requireUrlAccess(resolve(granted, schema, table), schema, table);
+    return queryReader.readChangeDataFeed(
+        resolved,
+        owner(granted),
+        options,
+        request.startingVersion(),
+        parseTimestamp("startingTimestamp", request.startingTimestamp()),
+        request.endingVersion(),
+        parseTimestamp("endingTimestamp", request.endingTimestamp()),
+        Boolean.TRUE.equals(request.includeHistoricalMetadata()),
+        Boolean.TRUE.equals(request.includeHistoricalProtocol()));
+  }
+
+  private static ResolvedAsset requireUrlAccess(ResolvedAsset table, String schema, String name) {
+    List<String> accessModes = TableAccessModes.forTable(table);
+    if (accessModes == null || !accessModes.contains("url")) {
+      throw ApiException.invalidParameter(
+          "table '" + schema + "." + name + "' does not support url access");
+    }
+    return table;
+  }
+
   // Predicate and limit hints apply to snapshot files only, so they are ignored here.
   private DeltaTableQueryReader.Result queryChanges(
       QueryTableRequest body,

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableActions.java

@@ -99,13 +99,14 @@ public static String parquetFile(
                 expirationTimestamp)));
   }
 
-  /** Data change file kinds, keyed {@code add} or {@code remove}. */
+  /** Data change file kinds, keyed {@code add}, {@code remove}, or {@code cdf} in parquet. */
   public enum ChangeType {
     ADD,
-    REMOVE
+    REMOVE,
+    CDC
   }
 
-  /** A data change file for a startingVersion query. */
+  /** A data change file for a startingVersion or Change Data Feed query. */
   public static String parquetChange(
       ChangeType type,
       String url,
@@ -130,6 +131,7 @@ public static String parquetChange(
         switch (type) {
           case ADD -> new AddLine(file);
           case REMOVE -> new RemoveLine(file);
+          case CDC -> new CdfLine(file);
         });
   }
 
@@ -146,10 +148,10 @@ public static String deltaFile(
         expirationTimestamp,
         version,
         timestamp,
-        new DeltaSingleAction(add.toRow(), null));
+        new DeltaSingleAction(add.toRow(), null, null));
   }
 
-  /** {@code action} is the add or remove row; {@code type} picks its action key. */
+  /** {@code action} is the add, remove, or cdc row; {@code type} picks its action key. */
   public static String deltaChange(
       String id,
       String deletionVectorFileId,
@@ -165,8 +167,9 @@ public static String deltaChange(
         version,
         timestamp,
         switch (type) {
-          case ADD -> new DeltaSingleAction(action, null);
-          case REMOVE -> new DeltaSingleAction(null, action);
+          case ADD -> new DeltaSingleAction(action, null, null);
+          case REMOVE -> new DeltaSingleAction(null, action, null);
+          case CDC -> new DeltaSingleAction(null, null, action);
         });
   }
 
@@ -331,12 +334,15 @@ public record DeltaFile(
   @JsonInclude(JsonInclude.Include.NON_NULL)
   public record DeltaSingleAction(
       @JsonSerialize(using = KernelActionSerializer.class) Row add,
-      @JsonSerialize(using = KernelActionSerializer.class) Row remove) {}
+      @JsonSerialize(using = KernelActionSerializer.class) Row remove,
+      @JsonSerialize(using = KernelActionSerializer.class) Row cdc) {}
 
   public record AddLine(ParquetFile add) {}
 
   public record RemoveLine(ParquetFile remove) {}
 
+  public record CdfLine(ParquetFile cdf) {}
+
   public record EndStreamLine(EndStreamAction endStreamAction) {}
 
   @JsonInclude(JsonInclude.Include.NON_NULL)

server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableChangesRequest.java

@@ -0,0 +1,14 @@
+package io.opensharing.asset.table;
+
+/**
+ * Query parameters of Query Table Changes. The range starts at {@code startingVersion} or {@code
+ * startingTimestamp}, and ends at {@code endingVersion}, {@code endingTimestamp}, or the latest
+ * version; timestamps are ISO-8601.
+ */
+public record TableChangesRequest(
+    Long startingVersion,
+    String startingTimestamp,
+    Long endingVersion,
+    String endingTimestamp,
+    Boolean includeHistoricalMetadata,
+    Boolean includeHistoricalProtocol) {}