Delta Sharing protocol: query shared Delta table metadata

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

@@ -0,0 +1,100 @@
+package io.opensharing.asset.table;
+
+import io.opensharing.http.ApiException;
+import java.util.EnumSet;
+import java.util.LinkedHashMap;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+
+/** Parses {@code delta-sharing-capabilities} and chooses parquet vs delta response actions. */
+public final class DeltaSharingCapabilities {
+
+  public enum ResponseFormat {
+    PARQUET,
+    DELTA
+  }
+
+  public static final String HEADER = "delta-sharing-capabilities";
+
+  private static final Set<String> KEYS =
+      Set.of("responseformat", "readerfeatures", "includeendstreamaction");
+
+  private DeltaSharingCapabilities() {}
+
+  /**
+   * No header, or parquet only, defaults to parquet. Prefer delta whenever it is listed, including
+   * {@code responseformat=delta,parquet}.
+   */
+  public static ResponseFormat choose(String header) {
+    return responseFormats(parse(header)).contains(ResponseFormat.DELTA)
+        ? ResponseFormat.DELTA
+        : ResponseFormat.PARQUET;
+  }
+
+  /** Value for the response {@code delta-sharing-capabilities} header: the format actually used. */
+  public static String responded(String requestHeader) {
+    return choose(requestHeader) == ResponseFormat.DELTA
+        ? "responseformat=delta"
+        : "responseformat=parquet";
+  }
+
+  static Map<String, String> parse(String header) {
+    if (header == null || header.isBlank()) {
+      return Map.of();
+    }
+    Map<String, String> capabilities = new LinkedHashMap<>();
+    for (String part : header.split(";")) {
+      String trimmed = part.trim();
+      if (trimmed.isEmpty()) {
+        continue;
+      }
+      int eq = trimmed.indexOf('=');
+      if (eq <= 0) {
+        throw ApiException.invalidParameter(
+            "delta-sharing-capabilities must be semicolon-separated key=value pairs");
+      }
+      String key = trimmed.substring(0, eq).trim().toLowerCase(Locale.ROOT);
+      String value = trimmed.substring(eq + 1).trim().toLowerCase(Locale.ROOT);
+      // Async query is unsupported; ignore it so those clients still get a synchronous response.
+      if (key.equals("asyncquery")) {
+        continue;
+      }
+      if (!KEYS.contains(key)) {
+        throw ApiException.invalidParameter(
+            "Unsupported delta-sharing-capabilities key: '" + key + "'");
+      }
+      if (key.equals("includeendstreamaction")
+          && !value.equals("true")
+          && !value.equals("false")) {
+        throw ApiException.invalidParameter(
+            "Unsupported " + key + ": '" + value + "'. Supported: true, false");
+      }
+      capabilities.put(key, value);
+    }
+    return capabilities;
+  }
+
+  static Set<ResponseFormat> responseFormats(Map<String, String> capabilities) {
+    EnumSet<ResponseFormat> requested = EnumSet.noneOf(ResponseFormat.class);
+    String formats = capabilities.get("responseformat");
+    if (formats == null) {
+      return requested;
+    }
+    if (formats.isBlank()) {
+      throw ApiException.invalidParameter(
+          "Unsupported responseformat: ''. Supported: parquet, delta");
+    }
+    for (String token : formats.split(",")) {
+      String format = token.trim();
+      switch (format) {
+        case "parquet" -> requested.add(ResponseFormat.PARQUET);
+        case "delta" -> requested.add(ResponseFormat.DELTA);
+        default ->
+            throw ApiException.invalidParameter(
+                "Unsupported responseformat: '" + format + "'. Supported: parquet, delta");
+      }
+    }
+    return requested;
+  }
+}

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

@@ -5,6 +5,7 @@
 import io.opensharing.asset.SharedDataObjectStore;
 import io.opensharing.asset.share.ProtocolShares;
 import io.opensharing.asset.table.delta.DeltaKernel;
+import io.opensharing.asset.table.delta.DeltaTableMetadataReader;
 import io.opensharing.auth.AuthContext;
 import io.opensharing.auth.UserContext;
 import io.opensharing.catalog.Asset;
@@ -40,16 +41,19 @@ public class ProtocolTables {
 
   private final CatalogConnector catalog;
   private final DeltaKernel kernel;
+  private final DeltaTableMetadataReader metadataReader;
   private final ProtocolShares shares;
   private final SharedDataObjectStore objects;
 
   public ProtocolTables(
       CatalogConnector catalog,
       DeltaKernel kernel,
+      DeltaTableMetadataReader metadataReader,
       ProtocolShares shares,
       SharedDataObjectStore objects) {
     this.catalog = catalog;
     this.kernel = kernel;
+    this.metadataReader = metadataReader;
     this.shares = shares;
     this.objects = objects;
   }
@@ -97,10 +101,29 @@ public long version(
       String table,
       String startingTimestamp) {
     ShareEntity granted = shares.requireGranted(recipient, share);
-    Instant timestamp = parseTimestamp(startingTimestamp);
+    Instant timestamp = parseTimestamp("startingTimestamp", startingTimestamp);
     return kernel.getVersion(resolve(granted, schema, table), timestamp, owner(granted));
   }
 
+  /**
+   * The protocol and metadata of one table in the share, as of the latest version, {@code
+   * version}, or an ISO-8601 {@code timestamp}, in the response format {@code capabilities} asks
+   * for.
+   */
+  public DeltaTableMetadataReader.Result metadata(
+      RecipientPrincipal recipient,
+      String share,
+      String schema,
+      String table,
+      Long version,
+      String timestamp,
+      String capabilities) {
+    ShareEntity granted = shares.requireGranted(recipient, share);
+    Instant asOf = parseTimestamp("timestamp", timestamp);
+    return metadataReader.read(
+        resolve(granted, schema, table), version, asOf, owner(granted), capabilities);
+  }
+
   private List<SharedDataObjectEntity> inSchema(ShareEntity share, String schema) {
     List<SharedDataObjectEntity> included =
         objects.listInSchema(share, ObjectNames.normalize(schema));
@@ -132,14 +155,14 @@ private ResolvedAsset resolve(ShareEntity share, String schema, String table) {
     return find(share, schema, table).resolved().get();
   }
 
-  private static Instant parseTimestamp(String value) {
+  private static Instant parseTimestamp(String name, String value) {
     if (value == null || value.isBlank()) {
       return null;
     }
     try {
       return Instant.parse(value);
     } catch (DateTimeParseException invalid) {
-      throw ApiException.invalidParameter("startingTimestamp must be an ISO-8601 UTC timestamp");
+      throw ApiException.invalidParameter(name + " must be an ISO-8601 UTC timestamp");
     }
   }
 

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

@@ -0,0 +1,155 @@
+package io.opensharing.asset.table;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import io.delta.kernel.defaults.internal.json.JsonUtils;
+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.util.VectorUtils;
+import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat;
+import io.opensharing.catalog.ResolvedAsset;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+/** Protocol and metadata NDJSON actions shared by Query Table Metadata and Query Table. */
+public final class TableActions {
+
+  private static final ObjectMapper JSON =
+      new ObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_NULL);
+
+  private TableActions() {}
+
+  public static String protocol(Protocol protocol, Long version, ResponseFormat format) {
+    return line(
+        new ProtocolLine(
+            format == ResponseFormat.DELTA
+                ? new DeltaProtocol(version, protocol)
+                : new ParquetProtocol(1)));
+  }
+
+  public static String metadata(
+      Metadata metadata,
+      ResolvedAsset table,
+      Long version,
+      Long size,
+      Long numFiles,
+      ResponseFormat format) {
+    if (format == ResponseFormat.DELTA) {
+      return line(
+          new MetadataLine(
+              new DeltaMetadata(
+                  version,
+                  size,
+                  numFiles,
+                  location(table),
+                  auxiliaryLocations(table),
+                  TableAccessModes.forTable(table),
+                  metadata)));
+    }
+    Format tableFormat = metadata.getFormat();
+    return line(
+        new MetadataLine(
+            new ParquetMetadata(
+                metadata.getId(),
+                metadata.getName().orElse(null),
+                metadata.getDescription().orElse(null),
+                location(table),
+                auxiliaryLocations(table),
+                TableAccessModes.forTable(table),
+                new FormatBody(tableFormat.getProvider()),
+                metadata.getSchemaString(),
+                VectorUtils.toJavaList(metadata.getPartitionColumns()),
+                emptyMapToNull(metadata.getConfiguration()),
+                version,
+                size,
+                numFiles)));
+  }
+
+  private static String line(Object value) {
+    try {
+      return JSON.writeValueAsString(value) + "\n";
+    } catch (JsonProcessingException e) {
+      throw new IllegalStateException("failed to encode table action", e);
+    }
+  }
+
+  private static String location(ResolvedAsset table) {
+    return table.location() == null ? null : table.location().storageLocation();
+  }
+
+  private static List<String> auxiliaryLocations(ResolvedAsset table) {
+    return table.location() == null || table.location().auxiliaryLocations().isEmpty()
+        ? null
+        : table.location().auxiliaryLocations();
+  }
+
+  private static Map<String, String> emptyMapToNull(Map<String, String> values) {
+    return values == null || values.isEmpty() ? null : values;
+  }
+
+  /** Writes Kernel protocol/metadata rows as Delta log JSON. */
+  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()));
+    }
+  }
+
+  public record ProtocolLine(ProtocolBody protocol) {}
+
+  public sealed interface ProtocolBody permits ParquetProtocol, DeltaProtocol {}
+
+  public record ParquetProtocol(int minReaderVersion) implements ProtocolBody {}
+
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  public record DeltaProtocol(
+      Long version,
+      @JsonSerialize(using = KernelActionSerializer.class) Protocol deltaProtocol)
+      implements ProtocolBody {}
+
+  public record MetadataLine(MetadataBody metaData) {}
+
+  public sealed interface MetadataBody permits ParquetMetadata, DeltaMetadata {}
+
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  public record ParquetMetadata(
+      String id,
+      String name,
+      String description,
+      String location,
+      List<String> auxiliaryLocations,
+      List<String> accessModes,
+      FormatBody format,
+      String schemaString,
+      List<String> partitionColumns,
+      Map<String, String> configuration,
+      Long version,
+      Long size,
+      Long numFiles)
+      implements MetadataBody {}
+
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  public record DeltaMetadata(
+      Long version,
+      Long size,
+      Long numFiles,
+      String location,
+      List<String> auxiliaryLocations,
+      List<String> accessModes,
+      @JsonSerialize(using = KernelActionSerializer.class) Metadata deltaMetadata)
+      implements MetadataBody {}
+
+  public record FormatBody(String provider) {}
+}

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

@@ -52,7 +52,8 @@ static long versionAtOrAfter(TableImpl table, Engine engine, Instant startingTim
     }
   }
 
-  private Session open(ResolvedAsset table, AuthContext auth) {
+  /** Opens the table's root with read credentials the catalog vends to {@code auth}. */
+  Session open(ResolvedAsset table, AuthContext auth) {
     if (!(table.additionalProperties() instanceof TableProperties properties)
         || properties.dataSourceFormat() != DataSourceFormat.DELTA) {
       throw ApiException.invalidParameter("table '" + table.fullName() + "' is not a Delta table");
@@ -82,7 +83,7 @@ private Session open(ResolvedAsset table, AuthContext auth) {
     Engine engine =
         DefaultEngine.create(HadoopStorageConfiguration.from(rootCredentials, location));
     String path = HadoopStorageConfiguration.kernelPath(location);
-    return new Session(engine, Table.forPath(engine, path));
+    return new Session(engine, Table.forPath(engine, path), location);
   }
 
   private static boolean covers(String prefix, String location) {
@@ -93,5 +94,5 @@ private static boolean covers(String prefix, String location) {
     return location.equals(prefix) || location.startsWith(normalized);
   }
 
-  private record Session(Engine engine, Table table) {}
+  record Session(Engine engine, Table table, String location) {}
 }

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

@@ -0,0 +1,29 @@
+package io.opensharing.asset.table.delta;
+
+import io.delta.kernel.Snapshot;
+import io.delta.kernel.exceptions.KernelException;
+import io.delta.kernel.exceptions.TableNotFoundException;
+import io.opensharing.http.ApiException;
+import java.time.Instant;
+
+/** Opens a Kernel snapshot for a version, timestamp, or the latest log. */
+final class DeltaSnapshots {
+
+  private DeltaSnapshots() {}
+
+  static Snapshot open(DeltaKernel.Session session, Long version, Instant timestamp) {
+    try {
+      if (version != null) {
+        return session.table().getSnapshotAsOfVersion(session.engine(), version);
+      }
+      if (timestamp != null) {
+        return session.table().getSnapshotAsOfTimestamp(session.engine(), timestamp.toEpochMilli());
+      }
+      return session.table().getLatestSnapshot(session.engine());
+    } catch (TableNotFoundException missing) {
+      throw ApiException.invalidParameter("table '" + session.location() + "' has no Delta log");
+    } catch (KernelException | IllegalArgumentException invalid) {
+      throw ApiException.invalidParameter(invalid.getMessage());
+    }
+  }
+}