Delta Sharing protocol: Query Table startingVersion, endingVersion, and historical protocol

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

@@ -177,7 +177,8 @@ public TemporaryTableCredentialsResponse temporaryCredentials(
    * {@code version}, or an ISO-8601 {@code timestamp} in {@code request}, with presigned file
    * URLs. A latest-snapshot query can ask for a refresh token, and a later query can pass it back
    * to re-sign the same version. jsonPredicateHints and limitHint narrow the files returned.
-   * startingVersion queries are not implemented.
+   * {@code startingVersion}, with an optional {@code endingVersion}, returns the data change files
+   * committed in that range instead.
    */
   public DeltaTableQueryReader.Result query(
       RecipientPrincipal recipient,
@@ -195,7 +196,7 @@ public DeltaTableQueryReader.Result query(
     }
     QueryTableRequest body = request == null ? QueryTableRequest.EMPTY : request;
     if (body.startingVersion() != null || body.endingVersion() != null) {
-      throw ApiException.notImplemented("startingVersion queries are not supported");
+      return queryChanges(body, resolved, granted, options);
     }
     boolean historical = body.version() != null || body.timestamp() != null;
     boolean includeRefreshToken = Boolean.TRUE.equals(body.includeRefreshToken());
@@ -222,6 +223,41 @@ public DeltaTableQueryReader.Result query(
         limitHint(body));
   }
 
+  // Predicate and limit hints apply to snapshot files only, so they are ignored here.
+  private DeltaTableQueryReader.Result queryChanges(
+      QueryTableRequest body,
+      ResolvedAsset table,
+      ShareEntity share,
+      DeltaTableQueryReader.ResponseOptions options) {
+    Long startingVersion = body.startingVersion();
+    Long endingVersion = body.endingVersion();
+    if (startingVersion == null) {
+      throw ApiException.invalidParameter("endingVersion requires startingVersion");
+    }
+    if (body.version() != null || body.timestamp() != null) {
+      throw ApiException.invalidParameter(
+          "Only one of version, timestamp, and startingVersion can be set");
+    }
+    if (Boolean.TRUE.equals(body.includeRefreshToken()) || body.refreshToken() != null) {
+      throw ApiException.invalidParameter(
+          "includeRefreshToken and refreshToken cannot be used with startingVersion");
+    }
+    if (startingVersion < 0) {
+      throw ApiException.invalidParameter("startingVersion must be non-negative");
+    }
+    if (endingVersion != null && endingVersion < startingVersion) {
+      throw ApiException.invalidParameter(
+          "endingVersion must be greater than or equal to startingVersion");
+    }
+    return queryReader.readChanges(
+        table,
+        owner(share),
+        options,
+        startingVersion,
+        endingVersion,
+        Boolean.TRUE.equals(body.includeHistoricalProtocol()));
+  }
+
   // limitHint counts unfiltered file records, so it is dropped when any predicate hint is present.
   private static Long limitHint(QueryTableRequest body) {
     boolean hasPredicate =

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

@@ -3,8 +3,8 @@
 import java.util.List;
 
 /**
- * Optional body for Query Table. Snapshot queries use version or timestamp; streaming is not
- * implemented.
+ * Optional body for Query Table. Snapshot queries use version or timestamp; startingVersion, with
+ * an optional endingVersion, returns data change files instead.
  */
 public record QueryTableRequest(
     // Deprecated by the protocol in favor of jsonPredicateHints; only logged, never used to filter.

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

@@ -99,22 +99,88 @@ public static String parquetFile(
                 expirationTimestamp)));
   }
 
+  /** Data change file kinds, keyed {@code add} or {@code remove}. */
+  public enum ChangeType {
+    ADD,
+    REMOVE
+  }
+
+  /** A data change file for a startingVersion query. */
+  public static String parquetChange(
+      ChangeType type,
+      String url,
+      String id,
+      Map<String, String> partitionValues,
+      long size,
+      String stats,
+      long version,
+      long timestamp,
+      Long expirationTimestamp) {
+    ParquetFile file =
+        new ParquetFile(
+            url,
+            id,
+            partitionValues == null ? Map.of() : partitionValues,
+            size,
+            stats,
+            version,
+            timestamp,
+            expirationTimestamp);
+    return line(
+        switch (type) {
+          case ADD -> new AddLine(file);
+          case REMOVE -> new RemoveLine(file);
+        });
+  }
+
   public static String deltaFile(
       String id,
       String deletionVectorFileId,
       Long expirationTimestamp,
       Long version,
       Long timestamp,
       AddFile add) {
+    return deltaFile(
+        id,
+        deletionVectorFileId,
+        expirationTimestamp,
+        version,
+        timestamp,
+        new DeltaSingleAction(add.toRow(), null));
+  }
+
+  /** {@code action} is the add or remove row; {@code type} picks its action key. */
+  public static String deltaChange(
+      String id,
+      String deletionVectorFileId,
+      Long expirationTimestamp,
+      long version,
+      long timestamp,
+      ChangeType type,
+      Row action) {
+    return deltaFile(
+        id,
+        deletionVectorFileId,
+        expirationTimestamp,
+        version,
+        timestamp,
+        switch (type) {
+          case ADD -> new DeltaSingleAction(action, null);
+          case REMOVE -> new DeltaSingleAction(null, action);
+        });
+  }
+
+  private static String deltaFile(
+      String id,
+      String deletionVectorFileId,
+      Long expirationTimestamp,
+      Long version,
+      Long timestamp,
+      DeltaSingleAction action) {
     return line(
         new FileLine(
             new DeltaFile(
-                id,
-                deletionVectorFileId,
-                version,
-                timestamp,
-                expirationTimestamp,
-                new DeltaSingleAction(add))));
+                id, deletionVectorFileId, version, timestamp, expirationTimestamp, action)));
   }
 
   public static String endStreamAction(String refreshToken, Long minUrlExpirationTimestamp) {
@@ -126,7 +192,11 @@ public static String endStreamAction(String refreshToken, Long minUrlExpirationT
    * on-disk deletion vector as a path-type descriptor so clients fetch the signed object.
    */
   public static AddFile withPath(AddFile add, String path, String deletionVectorUrl) {
-    Row row = add.toRow();
+    return new AddFile(withPath(add.toRow(), path, deletionVectorUrl));
+  }
+
+  /** {@link #withPath(AddFile, String, String)} for an add or remove action row. */
+  public static Row withPath(Row row, String path, String deletionVectorUrl) {
     Map<Integer, Object> overrides = new HashMap<>();
     overrides.put(row.getSchema().indexOf("path"), path);
     if (deletionVectorUrl != null) {
@@ -142,7 +212,7 @@ public static AddFile withPath(AddFile add, String path, String deletionVectorUr
                   dv.getSchema().indexOf("pathOrInlineDv"),
                   deletionVectorUrl)));
     }
-    return new AddFile(new DelegateRow(row, overrides));
+    return new DelegateRow(row, overrides);
   }
 
   private static String line(Object value) {
@@ -177,6 +247,7 @@ public void serialize(Object value, JsonGenerator gen, SerializerProvider serial
             case Protocol protocol -> protocol.toRow();
             case Metadata metadata -> metadata.toRow();
             case AddFile add -> add.toRow();
+            case Row action -> action;
             default ->
                 throw new IllegalArgumentException(
                     "unsupported Kernel action: " + value.getClass().getName());
@@ -257,8 +328,14 @@ public record DeltaFile(
       DeltaSingleAction deltaSingleAction)
       implements FileBody {}
 
+  @JsonInclude(JsonInclude.Include.NON_NULL)
   public record DeltaSingleAction(
-      @JsonSerialize(using = KernelActionSerializer.class) AddFile add) {}
+      @JsonSerialize(using = KernelActionSerializer.class) Row add,
+      @JsonSerialize(using = KernelActionSerializer.class) Row remove) {}
+
+  public record AddLine(ParquetFile add) {}
+
+  public record RemoveLine(ParquetFile remove) {}
 
   public record EndStreamLine(EndStreamAction endStreamAction) {}