Delta Sharing protocol: Query Table snapshot API

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

@@ -1,11 +1,16 @@
 package io.opensharing.asset.table;
 
+import io.delta.kernel.internal.actions.Protocol;
+import io.delta.kernel.internal.tablefeatures.TableFeature;
+import io.delta.kernel.internal.tablefeatures.TableFeatures;
 import io.opensharing.http.ApiException;
 import java.util.EnumSet;
 import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /** Parses {@code delta-sharing-capabilities} and chooses parquet vs delta response actions. */
 public final class DeltaSharingCapabilities {
@@ -20,6 +25,12 @@ public enum ResponseFormat {
   private static final Set<String> KEYS =
       Set.of("responseformat", "readerfeatures", "includeendstreamaction");
 
+  private static final Set<String> READER_FEATURES =
+      TableFeatures.TABLE_FEATURES.stream()
+          .filter(TableFeature::isReaderWriterFeature)
+          .map(feature -> feature.featureName().toLowerCase(Locale.ROOT))
+          .collect(Collectors.toUnmodifiableSet());
+
   private DeltaSharingCapabilities() {}
 
   /**
@@ -39,6 +50,32 @@ public static String responded(String requestHeader) {
         : "responseformat=parquet";
   }
 
+  /**
+   * Query Table checks reader features only for delta responses on snapshots that declare them.
+   * Parquet never carries {@code readerfeatures}; a parquet query of an advanced table is rejected.
+   * Metadata does not enforce this so clients can inspect protocol before querying files.
+   */
+  public static void requireReaderFeatures(
+      String header, ResponseFormat format, Protocol protocol) {
+    Set<String> required = protocol.getReaderFeatures();
+    if (required == null || required.isEmpty()) {
+      return;
+    }
+    if (format != ResponseFormat.DELTA) {
+      throw ApiException.invalidParameter(
+          "The table requires reader features that are only available in delta format");
+    }
+    Set<String> advertised = advertisedReaderFeatures(header);
+    for (String feature : required) {
+      if (!advertised.contains(feature.toLowerCase(Locale.ROOT))) {
+        throw ApiException.invalidParameter(
+            "The table requires reader feature '"
+                + feature
+                + "' which was not specified in delta-sharing-capabilities");
+      }
+    }
+  }
+
   static Map<String, String> parse(String header) {
     if (header == null || header.isBlank()) {
       return Map.of();
@@ -97,4 +134,23 @@ static Set<ResponseFormat> responseFormats(Map<String, String> capabilities) {
     }
     return requested;
   }
+
+  private static Set<String> advertisedReaderFeatures(String header) {
+    String value = parse(header).get("readerfeatures");
+    if (value == null) {
+      return Set.of();
+    }
+    if (value.isBlank()) {
+      throw ApiException.invalidParameter("Unsupported readerfeatures: ''");
+    }
+    Set<String> advertised = new LinkedHashSet<>();
+    for (String token : value.split(",")) {
+      String feature = token.trim();
+      if (feature.isEmpty() || !READER_FEATURES.contains(feature)) {
+        throw ApiException.invalidParameter("Unsupported readerfeatures: '" + feature + "'");
+      }
+      advertised.add(feature);
+    }
+    return advertised;
+  }
 }

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

@@ -0,0 +1,53 @@
+package io.opensharing.asset.table;
+
+import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat;
+import io.opensharing.http.ApiException;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.HexFormat;
+import java.util.Locale;
+import java.util.Set;
+
+/**
+ * Parses {@code fileidhash} and hashes file paths the way the OSS server does: {@code parquet} is
+ * MD5, {@code delta} is SHA-256. When the header is omitted, the hash follows the responded format.
+ */
+public final class FileIdHash {
+
+  public static final String HEADER = "fileidhash";
+
+  private static final Set<String> VALID = Set.of("parquet", "delta");
+
+  private FileIdHash() {}
+
+  /** Lowercase {@code parquet} or {@code delta}, or {@code null} when the header is absent. */
+  public static String parse(String header) {
+    if (header == null || header.isBlank()) {
+      return null;
+    }
+    String normalized = header.trim().toLowerCase(Locale.ROOT);
+    if (!VALID.contains(normalized)) {
+      throw ApiException.invalidParameter(
+          "Unsupported fileidhash: '" + header + "'. Supported: parquet, delta");
+    }
+    return normalized;
+  }
+
+  public static String hash(String path, String fileIdHash, ResponseFormat format) {
+    boolean sha256 =
+        "delta".equals(fileIdHash)
+            || (fileIdHash == null && format == ResponseFormat.DELTA);
+    return digest(sha256 ? "SHA-256" : "MD5", path);
+  }
+
+  private static String digest(String algorithm, String path) {
+    try {
+      return HexFormat.of()
+          .formatHex(
+              MessageDigest.getInstance(algorithm).digest(path.getBytes(StandardCharsets.UTF_8)));
+    } catch (NoSuchAlgorithmException e) {
+      throw new IllegalStateException(e);
+    }
+  }
+}

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

@@ -6,6 +6,7 @@
 import io.opensharing.asset.share.ProtocolShares;
 import io.opensharing.asset.table.delta.DeltaKernel;
 import io.opensharing.asset.table.delta.DeltaTableMetadataReader;
+import io.opensharing.asset.table.delta.DeltaTableQueryReader;
 import io.opensharing.auth.AuthContext;
 import io.opensharing.auth.UserContext;
 import io.opensharing.catalog.Asset;
@@ -46,18 +47,21 @@ public class ProtocolTables {
   private final CatalogConnector catalog;
   private final DeltaKernel kernel;
   private final DeltaTableMetadataReader metadataReader;
+  private final DeltaTableQueryReader queryReader;
   private final ProtocolShares shares;
   private final SharedDataObjectStore objects;
 
   public ProtocolTables(
       CatalogConnector catalog,
       DeltaKernel kernel,
       DeltaTableMetadataReader metadataReader,
+      DeltaTableQueryReader queryReader,
       ProtocolShares shares,
       SharedDataObjectStore objects) {
     this.catalog = catalog;
     this.kernel = kernel;
     this.metadataReader = metadataReader;
+    this.queryReader = queryReader;
     this.shares = shares;
     this.objects = objects;
   }
@@ -165,6 +169,46 @@ public TemporaryTableCredentialsResponse temporaryCredentials(
     return new TemporaryTableCredentialsResponse(TemporaryCredentials.from(minted));
   }
 
+  /**
+   * One snapshot of a table in the share that supports url access, as of the latest version,
+   * {@code version}, or an ISO-8601 {@code timestamp} in {@code request}, with presigned file
+   * URLs. Pushdown hints and startingVersion queries are not implemented.
+   */
+  public DeltaTableQueryReader.Result query(
+      RecipientPrincipal recipient,
+      String share,
+      String schema,
+      String table,
+      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");
+    }
+    QueryTableRequest body = request == null ? QueryTableRequest.EMPTY : request;
+    if (body.startingVersion() != null || body.endingVersion() != null) {
+      throw ApiException.notImplemented("startingVersion queries are not supported");
+    }
+    if (hasPushdownHint(body)) {
+      throw ApiException.notImplemented("predicate and limit pushdown is not supported");
+    }
+    return queryReader.read(
+        resolved,
+        owner(granted),
+        options,
+        body.version(),
+        parseTimestamp("timestamp", body.timestamp()));
+  }
+
+  private static boolean hasPushdownHint(QueryTableRequest body) {
+    return (body.predicateHints() != null && !body.predicateHints().isEmpty())
+        || (body.jsonPredicateHints() != null && !body.jsonPredicateHints().isBlank())
+        || body.limitHint() != null;
+  }
+
   private List<SharedDataObjectEntity> inSchema(ShareEntity share, String schema) {
     List<SharedDataObjectEntity> included =
         objects.listInSchema(share, ObjectNames.normalize(schema));

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

@@ -0,0 +1,22 @@
+package io.opensharing.asset.table;
+
+import java.util.List;
+
+/**
+ * Optional body for Query Table. Snapshot queries use version or timestamp; streaming is not
+ * implemented.
+ */
+public record QueryTableRequest(
+    List<String> predicateHints,
+    String jsonPredicateHints,
+    Integer limitHint,
+    Long version,
+    String timestamp,
+    Long startingVersion,
+    Long endingVersion,
+    Boolean includeHistoricalProtocol,
+    String idempotencyKey) {
+
+  static final QueryTableRequest EMPTY =
+      new QueryTableRequest(null, null, null, null, null, null, null, null, null);
+}

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

@@ -7,18 +7,23 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializerProvider;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import io.delta.kernel.data.Row;
 import io.delta.kernel.defaults.internal.json.JsonUtils;
+import io.delta.kernel.internal.actions.AddFile;
+import io.delta.kernel.internal.actions.DeletionVectorDescriptor;
 import io.delta.kernel.internal.actions.Format;
 import io.delta.kernel.internal.actions.Metadata;
 import io.delta.kernel.internal.actions.Protocol;
+import io.delta.kernel.internal.data.DelegateRow;
 import io.delta.kernel.internal.util.VectorUtils;
 import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat;
 import io.opensharing.catalog.ResolvedAsset;
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
-/** Protocol and metadata NDJSON actions shared by Query Table Metadata and Query Table. */
+/** Protocol, metadata, and file NDJSON actions shared by Query Table Metadata and Query Table. */
 public final class TableActions {
 
   private static final ObjectMapper JSON =
@@ -72,6 +77,70 @@ public static String metadata(
                 numFiles)));
   }
 
+  public static String parquetFile(
+      String url,
+      String id,
+      Map<String, String> partitionValues,
+      long size,
+      String stats,
+      Long version,
+      Long timestamp,
+      Long expirationTimestamp) {
+    return line(
+        new FileLine(
+            new ParquetFile(
+                url,
+                id,
+                partitionValues == null ? Map.of() : partitionValues,
+                size,
+                stats,
+                version,
+                timestamp,
+                expirationTimestamp)));
+  }
+
+  public static String deltaFile(
+      String id,
+      String deletionVectorFileId,
+      Long expirationTimestamp,
+      Long version,
+      Long timestamp,
+      AddFile add) {
+    return line(
+        new FileLine(
+            new DeltaFile(
+                id,
+                deletionVectorFileId,
+                version,
+                timestamp,
+                expirationTimestamp,
+                new DeltaSingleAction(add))));
+  }
+
+  /**
+   * Replace the add path with a signed file URL. When {@code deletionVectorUrl} is set, rewrite the
+   * 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();
+    Map<Integer, Object> overrides = new HashMap<>();
+    overrides.put(row.getSchema().indexOf("path"), path);
+    if (deletionVectorUrl != null) {
+      int dvOrdinal = row.getSchema().indexOf("deletionVector");
+      Row dv = row.getStruct(dvOrdinal);
+      overrides.put(
+          dvOrdinal,
+          new DelegateRow(
+              dv,
+              Map.of(
+                  dv.getSchema().indexOf("storageType"),
+                  (Object) DeletionVectorDescriptor.PATH_DV_MARKER,
+                  dv.getSchema().indexOf("pathOrInlineDv"),
+                  deletionVectorUrl)));
+    }
+    return new AddFile(new DelegateRow(row, overrides));
+  }
+
   private static String line(Object value) {
     try {
       return JSON.writeValueAsString(value) + "\n";
@@ -99,11 +168,16 @@ public static final class KernelActionSerializer extends JsonSerializer<Object>
     @Override
     public void serialize(Object value, JsonGenerator gen, SerializerProvider serializers)
         throws IOException {
-      gen.writeRawValue(
-          JsonUtils.rowToJson(
-              value instanceof Protocol protocol
-                  ? protocol.toRow()
-                  : ((Metadata) value).toRow()));
+      Row row =
+          switch (value) {
+            case Protocol protocol -> protocol.toRow();
+            case Metadata metadata -> metadata.toRow();
+            case AddFile add -> add.toRow();
+            default ->
+                throw new IllegalArgumentException(
+                    "unsupported Kernel action: " + value.getClass().getName());
+          };
+      gen.writeRawValue(JsonUtils.rowToJson(row));
     }
   }
 
@@ -152,4 +226,33 @@ public record DeltaMetadata(
       implements MetadataBody {}
 
   public record FormatBody(String provider) {}
+
+  public record FileLine(FileBody file) {}
+
+  public sealed interface FileBody permits ParquetFile, DeltaFile {}
+
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  public record ParquetFile(
+      String url,
+      String id,
+      Map<String, String> partitionValues,
+      long size,
+      String stats,
+      Long version,
+      Long timestamp,
+      Long expirationTimestamp)
+      implements FileBody {}
+
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  public record DeltaFile(
+      String id,
+      String deletionVectorFileId,
+      Long version,
+      Long timestamp,
+      Long expirationTimestamp,
+      DeltaSingleAction deltaSingleAction)
+      implements FileBody {}
+
+  public record DeltaSingleAction(
+      @JsonSerialize(using = KernelActionSerializer.class) AddFile add) {}
 }