From 35a0d52154cb20072e8fc4c7d2c0b928880abdaa Mon Sep 17 00:00:00 2001 From: Sun Cao Date: Tue, 13 Oct 2026 18:31:27 -0700 Subject: [PATCH] Serve protocol table metadata as NDJSON through Delta Kernel, on Armeria. --- .../asset/table/DeltaSharingCapabilities.java | 100 +++++++++++ .../asset/table/ProtocolTables.java | 29 +++- .../opensharing/asset/table/TableActions.java | 155 ++++++++++++++++++ .../asset/table/delta/DeltaKernel.java | 7 +- .../asset/table/delta/DeltaSnapshots.java | 29 ++++ .../table/delta/DeltaTableMetadataReader.java | 61 +++++++ .../protocol/DeltaSharingProtocolService.java | 39 +++++ .../io/opensharing/runtime/OpenSharing.java | 5 +- .../table/DeltaSharingCapabilitiesTest.java | 50 ++++++ .../asset/table/TableActionsTest.java | 117 +++++++++++++ .../test/java/io/opensharing/TestServer.java | 21 ++- .../protocol/LocalDeltaTableSupport.java | 7 +- .../ProtocolTableMetadataIntegrationTest.java | 145 ++++++++++++++++ .../protocol/ProtocolTableMetadataTest.java | 112 +++++++++++++ .../test/resources/local-catalog-cloud.yml | 9 +- 15 files changed, 872 insertions(+), 14 deletions(-) create mode 100644 server/opensharing-server-core/src/main/java/io/opensharing/asset/table/DeltaSharingCapabilities.java create mode 100644 server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableActions.java create mode 100644 server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaSnapshots.java create mode 100644 server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaTableMetadataReader.java create mode 100644 server/opensharing-server-core/src/test/java/io/opensharing/asset/table/DeltaSharingCapabilitiesTest.java create mode 100644 server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TableActionsTest.java create mode 100644 server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableMetadataIntegrationTest.java create mode 100644 server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableMetadataTest.java diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/DeltaSharingCapabilities.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/DeltaSharingCapabilities.java new file mode 100644 index 000000000..bba7bcd85 --- /dev/null +++ b/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 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 parse(String header) { + if (header == null || header.isBlank()) { + return Map.of(); + } + Map 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 responseFormats(Map capabilities) { + EnumSet 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; + } +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java index f569095fb..171e49340 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java +++ b/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 inSchema(ShareEntity share, String schema) { List 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"); } } diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableActions.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableActions.java new file mode 100644 index 000000000..dc92a103d --- /dev/null +++ b/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 auxiliaryLocations(ResolvedAsset table) { + return table.location() == null || table.location().auxiliaryLocations().isEmpty() + ? null + : table.location().auxiliaryLocations(); + } + + private static Map emptyMapToNull(Map values) { + return values == null || values.isEmpty() ? null : values; + } + + /** Writes Kernel protocol/metadata rows as Delta log JSON. */ + public static final class KernelActionSerializer extends JsonSerializer { + @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 auxiliaryLocations, + List accessModes, + FormatBody format, + String schemaString, + List partitionColumns, + Map 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 auxiliaryLocations, + List accessModes, + @JsonSerialize(using = KernelActionSerializer.class) Metadata deltaMetadata) + implements MetadataBody {} + + public record FormatBody(String provider) {} +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java index d93d7d41d..123525dd9 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java +++ b/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) {} } diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaSnapshots.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaSnapshots.java new file mode 100644 index 000000000..74323f2d0 --- /dev/null +++ b/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()); + } + } +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaTableMetadataReader.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaTableMetadataReader.java new file mode 100644 index 000000000..1ab6b278f --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaTableMetadataReader.java @@ -0,0 +1,61 @@ +package io.opensharing.asset.table.delta; + +import io.delta.kernel.Snapshot; +import io.delta.kernel.internal.SnapshotImpl; +import io.delta.kernel.internal.checksum.CRCInfo; +import io.opensharing.asset.table.DeltaSharingCapabilities; +import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat; +import io.opensharing.asset.table.TableActions; +import io.opensharing.auth.AuthContext; +import io.opensharing.catalog.ResolvedAsset; +import io.opensharing.http.ApiException; +import java.time.Instant; + +/** + * Reads Query Table Metadata from a Delta log. The parquet or delta response is two NDJSON lines: + * protocol, then metaData. + */ +public class DeltaTableMetadataReader { + + private final DeltaKernel kernel; + + public DeltaTableMetadataReader(DeltaKernel kernel) { + this.kernel = kernel; + } + + public Result read( + ResolvedAsset table, + Long version, + Instant timestamp, + AuthContext auth, + String capabilities) { + if (version != null && timestamp != null) { + throw ApiException.invalidParameter("version and timestamp are mutually exclusive"); + } + DeltaKernel.Session session = kernel.open(table, auth); + Snapshot snapshot = DeltaSnapshots.open(session, version, timestamp); + SnapshotImpl impl = (SnapshotImpl) snapshot; + boolean historical = version != null || timestamp != null; + ResponseFormat format = DeltaSharingCapabilities.choose(capabilities); + CRCInfo crc = crcFor(impl, snapshot.getVersion()); + return new Result( + snapshot.getVersion(), + TableActions.protocol(impl.getProtocol(), null, format) + + TableActions.metadata( + impl.getMetadata(), + table, + historical ? snapshot.getVersion() : null, + crc == null ? null : crc.getTableSizeBytes(), + crc == null ? null : crc.getNumFiles(), + format)); + } + + private static CRCInfo crcFor(SnapshotImpl snapshot, long version) { + return snapshot + .getCurrentCrcInfo() + .filter(crc -> crc.getVersion() == version) + .orElse(null); + } + + public record Result(long version, String ndjson) {} +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java b/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java index e39570a8f..d3492ea0a 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java +++ b/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java @@ -1,15 +1,21 @@ package io.opensharing.protocol; +import com.linecorp.armeria.common.HttpData; import com.linecorp.armeria.common.HttpResponse; import com.linecorp.armeria.common.HttpStatus; +import com.linecorp.armeria.common.MediaType; import com.linecorp.armeria.common.ResponseHeaders; import com.linecorp.armeria.common.annotation.Nullable; import com.linecorp.armeria.server.ServiceRequestContext; import com.linecorp.armeria.server.annotation.Blocking; import com.linecorp.armeria.server.annotation.Get; +import com.linecorp.armeria.server.annotation.Header; import com.linecorp.armeria.server.annotation.Param; +import com.linecorp.armeria.server.annotation.Produces; import com.linecorp.armeria.server.annotation.ProducesJson; +import io.opensharing.asset.table.DeltaSharingCapabilities; import io.opensharing.asset.table.ProtocolTables; +import io.opensharing.asset.table.delta.DeltaTableMetadataReader; /** * The Delta Sharing protocol endpoints for one shared table, which recipients call with their @@ -20,6 +26,7 @@ public final class DeltaSharingProtocolService { private static final String DELTA_TABLE_VERSION = "Delta-Table-Version"; + private static final MediaType NDJSON = MediaType.parse("application/x-ndjson; charset=utf-8"); private final ProtocolTables tables; @@ -44,4 +51,36 @@ public HttpResponse getTableVersion( return HttpResponse.of( ResponseHeaders.of(HttpStatus.OK, DELTA_TABLE_VERSION, Long.toString(version))); } + + /** + * {@code GET /shares/{share}/schemas/{schema}/tables/{table}/metadata}: the table's protocol and + * metadata as NDJSON, with its version in the {@code Delta-Table-Version} header. + */ + @Get("/shares/{share}/schemas/{schema}/tables/{table}/metadata") + @Produces("application/x-ndjson") + public HttpResponse getTableMetadata( + ServiceRequestContext ctx, + @Param String share, + @Param String schema, + @Param String table, + @Param @Nullable Long version, + @Param @Nullable String timestamp, + @Header(DeltaSharingCapabilities.HEADER) @Nullable String capabilities) { + DeltaTableMetadataReader.Result result = + tables.metadata( + RecipientTokenDecorator.recipient(ctx), + share, + schema, + table, + version, + timestamp, + capabilities); + ResponseHeaders headers = + ResponseHeaders.builder(HttpStatus.OK) + .contentType(NDJSON) + .add(DELTA_TABLE_VERSION, Long.toString(result.version())) + .add(DeltaSharingCapabilities.HEADER, DeltaSharingCapabilities.responded(capabilities)) + .build(); + return HttpResponse.of(headers, HttpData.ofUtf8(result.ndjson())); + } } diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/runtime/OpenSharing.java b/server/opensharing-server-core/src/main/java/io/opensharing/runtime/OpenSharing.java index 1aba934f9..83d4adbd4 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/runtime/OpenSharing.java +++ b/server/opensharing-server-core/src/main/java/io/opensharing/runtime/OpenSharing.java @@ -7,6 +7,7 @@ import io.opensharing.asset.share.ProtocolShares; import io.opensharing.asset.table.ProtocolTables; import io.opensharing.asset.table.delta.DeltaKernel; +import io.opensharing.asset.table.delta.DeltaTableMetadataReader; import io.opensharing.catalog.CatalogConnector; import io.opensharing.recipient.RecipientEntity; import io.opensharing.recipient.RecipientService; @@ -44,8 +45,10 @@ private OpenSharing(CatalogConnector catalog, Transactions tx) { this.recipients = new RecipientService(catalog, recipientStore); this.protocolShares = new ProtocolShares(shareStore, permissionStore); this.protocolSchemas = new ProtocolSchemas(protocolShares, objectStore); + DeltaKernel kernel = new DeltaKernel(catalog); this.protocolTables = - new ProtocolTables(catalog, new DeltaKernel(catalog), protocolShares, objectStore); + new ProtocolTables( + catalog, kernel, new DeltaTableMetadataReader(kernel), protocolShares, objectStore); } public static Builder builder() { diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/DeltaSharingCapabilitiesTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/DeltaSharingCapabilitiesTest.java new file mode 100644 index 000000000..80d62f02d --- /dev/null +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/DeltaSharingCapabilitiesTest.java @@ -0,0 +1,50 @@ +package io.opensharing.asset.table; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat; +import io.opensharing.http.ApiException; +import org.junit.jupiter.api.Test; + +class DeltaSharingCapabilitiesTest { + + @Test + void defaultsToParquetWithoutAHeader() { + assertEquals(ResponseFormat.PARQUET, DeltaSharingCapabilities.choose(null)); + assertEquals(ResponseFormat.PARQUET, DeltaSharingCapabilities.choose("")); + assertEquals("responseformat=parquet", DeltaSharingCapabilities.responded(null)); + assertEquals( + "responseformat=delta", DeltaSharingCapabilities.responded("responseformat=delta")); + } + + @Test + void respectsASingleRequestedFormat() { + assertEquals(ResponseFormat.DELTA, DeltaSharingCapabilities.choose("responseformat=delta")); + assertEquals(ResponseFormat.PARQUET, DeltaSharingCapabilities.choose("responseformat=parquet")); + } + + @Test + void prefersDeltaWhenBothAreListed() { + assertEquals( + ResponseFormat.DELTA, DeltaSharingCapabilities.choose("responseFormat=delta,parquet")); + assertEquals( + ResponseFormat.DELTA, + DeltaSharingCapabilities.choose( + "RESPONSEFORMAT=PARQUET,DELTA;readerfeatures=deletionvectors")); + } + + @Test + void rejectsUnknownFormatsKeysAndMalformedHeader() { + assertThrows(ApiException.class, () -> DeltaSharingCapabilities.choose("responseformat=json")); + assertThrows(ApiException.class, () -> DeltaSharingCapabilities.choose("responseformat=")); + assertThrows(ApiException.class, () -> DeltaSharingCapabilities.choose("not-a-capability")); + assertThrows(ApiException.class, () -> DeltaSharingCapabilities.choose("unknown=true")); + assertEquals(ResponseFormat.PARQUET, DeltaSharingCapabilities.choose("asyncquery=true")); + assertEquals( + ResponseFormat.DELTA, + DeltaSharingCapabilities.choose("responseformat=delta;asyncquery=true")); + assertEquals( + ResponseFormat.PARQUET, DeltaSharingCapabilities.choose("readerfeatures=notAFeature")); + } +} diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TableActionsTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TableActionsTest.java new file mode 100644 index 000000000..8c33bd7d5 --- /dev/null +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TableActionsTest.java @@ -0,0 +1,117 @@ +package io.opensharing.asset.table; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +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.delta.kernel.types.StringType; +import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat; +import io.opensharing.catalog.AssetLocation; +import io.opensharing.catalog.AssetType; +import io.opensharing.catalog.DataSourceFormat; +import io.opensharing.catalog.ResolvedAsset; +import io.opensharing.catalog.TableProperties; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import org.junit.jupiter.api.Test; + +class TableActionsTest { + + private static final String SCHEMA = + "{\"type\":\"struct\",\"fields\":[" + + "{\"name\":\"eventTime\",\"type\":\"timestamp\",\"nullable\":true,\"metadata\":{}}," + + "{\"name\":\"date\",\"type\":\"date\",\"nullable\":true,\"metadata\":{}}]}"; + + private static final String LOCATIONS = + "\"location\":\"s3://delta-share-demo/tables/table1\"," + + "\"auxiliaryLocations\":[\"s3://delta-share-demo/tables/table1-aux1\"]," + + "\"accessModes\":[\"url\",\"dir\"]"; + + private static final ResolvedAsset TABLE = + ResolvedAsset.builder(AssetType.TABLE, "main.sales.table1") + .location( + new AssetLocation( + "s3://delta-share-demo/tables/table1", + List.of("s3://delta-share-demo/tables/table1-aux1"))) + .additionalProperties( + new TableProperties(DataSourceFormat.DELTA, Map.of("subtype", "MANAGED"))) + .build(); + + @Test + void encodesParquetProtocolAndMetadata() { + assertEquals(parquetBody(""), encode(ResponseFormat.PARQUET, null, null, null)); + assertEquals(parquetBody(",\"version\":20"), encode(ResponseFormat.PARQUET, 20L, null, null)); + } + + @Test + void encodesDeltaProtocolAndMetadataFromKernel() { + Protocol protocol = protocol(); + Metadata metadata = metadata(); + String deltaProtocol = JsonUtils.rowToJson(protocol.toRow()); + String deltaMetadata = JsonUtils.rowToJson(metadata.toRow()); + assertEquals( + deltaBody(deltaProtocol, "", deltaMetadata), + encode(ResponseFormat.DELTA, null, null, null)); + assertEquals( + deltaBody(deltaProtocol, "\"size\":123456,\"numFiles\":5,", deltaMetadata), + encode(ResponseFormat.DELTA, null, 123456L, 5L)); + assertEquals( + """ + {"protocol":{"version":5,"deltaProtocol":%s}} + """ + .formatted(deltaProtocol), + TableActions.protocol(protocol, 5L, ResponseFormat.DELTA)); + } + + private static String encode(ResponseFormat format, Long version, Long size, Long numFiles) { + return TableActions.protocol(protocol(), null, format) + + TableActions.metadata(metadata(), TABLE, version, size, numFiles, format); + } + + private static String parquetBody(String versionField) { + String schemaString = SCHEMA.replace("\"", "\\\""); + return "{\"protocol\":{\"minReaderVersion\":1}}\n" + + "{\"metaData\":{\"id\":\"f8d5c169-3d01-4ca3-ad9e-7dc3355aedb2\"," + + LOCATIONS + + ",\"format\":{\"provider\":\"parquet\"},\"schemaString\":\"" + + schemaString + + "\",\"partitionColumns\":[\"date\"]," + + "\"configuration\":{\"enableChangeDataFeed\":\"true\"}" + + versionField + + "}}\n"; + } + + private static String deltaBody(String deltaProtocol, String stats, String deltaMetadata) { + return "{\"protocol\":{\"deltaProtocol\":" + + deltaProtocol + + "}}\n" + + "{\"metaData\":{" + + stats + + LOCATIONS + + ",\"deltaMetadata\":" + + deltaMetadata + + "}}\n"; + } + + private static Protocol protocol() { + return new Protocol(3, 7, Set.of("columnMapping"), Set.of("columnMapping", "identityColumns")); + } + + private static Metadata metadata() { + return new Metadata( + "f8d5c169-3d01-4ca3-ad9e-7dc3355aedb2", + Optional.empty(), + Optional.empty(), + new Format(), + SCHEMA, + null, + VectorUtils.buildArrayValue(List.of("date"), StringType.STRING), + Optional.empty(), + VectorUtils.stringStringMapValue(Map.of("enableChangeDataFeed", "true"))); + } +} diff --git a/server/opensharing-server/src/test/java/io/opensharing/TestServer.java b/server/opensharing-server/src/test/java/io/opensharing/TestServer.java index 193932094..f063a00ed 100644 --- a/server/opensharing-server/src/test/java/io/opensharing/TestServer.java +++ b/server/opensharing-server/src/test/java/io/opensharing/TestServer.java @@ -74,12 +74,19 @@ public Response delete(String path, String token) { return call(HttpMethod.DELETE, path, token, null); } - /** Calls {@code path} with {@code token} as the bearer token, unless it is null. */ - public Response call(HttpMethod method, String path, String token, String json) { + /** + * Calls {@code path} with {@code token} as the bearer token, unless it is null, and with {@code + * extraHeaders} as name, value pairs. + */ + public Response call( + HttpMethod method, String path, String token, String json, String... extraHeaders) { RequestHeadersBuilder headers = RequestHeaders.builder(method, path); if (token != null) { headers.add(HttpHeaderNames.AUTHORIZATION, "Bearer " + token); } + for (int i = 0; i < extraHeaders.length; i += 2) { + headers.add(extraHeaders[i], extraHeaders[i + 1]); + } HttpRequest request = json == null ? HttpRequest.of(headers.build()) @@ -93,13 +100,17 @@ public void close() { server.close(); } - /** A response with its body parsed as JSON, or a missing node when it has none. */ - public record Response(int status, ResponseHeaders headers, JsonNode body) { + /** + * A response with its raw {@code text}, and its body parsed as JSON, or a missing node when it + * has none. + */ + public record Response(int status, ResponseHeaders headers, String text, JsonNode body) { static Response of(AggregatedHttpResponse response) { + String text = response.contentUtf8(); try { return new Response( - response.status().code(), response.headers(), JSON.readTree(response.contentUtf8())); + response.status().code(), response.headers(), text, JSON.readTree(text)); } catch (IOException e) { throw new UncheckedIOException(e); } diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/LocalDeltaTableSupport.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/LocalDeltaTableSupport.java index 9f41048f4..85e07a37e 100644 --- a/server/opensharing-server/src/test/java/io/opensharing/protocol/LocalDeltaTableSupport.java +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/LocalDeltaTableSupport.java @@ -74,6 +74,11 @@ static void startLocalDeltaServer() { restartServer(properties); } + /** The storage location of table {@code name}. */ + protected static String location(String name) { + return tables.resolve(name).toUri().toString(); + } + /** Writes a Delta log of {@code commits} commits and returns the table's location. */ private static String deltaTable(String name, int commits) { for (int version = 0; version < commits; version++) { @@ -89,7 +94,7 @@ private static String deltaTable(String name, int commits) { throw new UncheckedIOException(e); } } - return tables.resolve(name).toUri().toString(); + return location(name); } private static Path write(String relativePath, String content) { diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableMetadataIntegrationTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableMetadataIntegrationTest.java new file mode 100644 index 000000000..eb888381e --- /dev/null +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableMetadataIntegrationTest.java @@ -0,0 +1,145 @@ +package io.opensharing.protocol; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.linecorp.armeria.common.HttpMethod; +import com.linecorp.armeria.common.MediaType; +import io.opensharing.TestServer.Response; +import io.opensharing.http.ErrorCodes; +import java.util.Properties; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; +import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable; + +/** + * Recipient Query Table Metadata through the local catalog, Kernel, and real cloud Delta logs. + * Individual cloud cases are skipped unless their repository credentials are present. + */ +@Timeout(60) +class ProtocolTableMetadataIntegrationTest extends ProtocolApiSupport { + + @BeforeAll + static void startCloudServer() { + Properties properties = new Properties(); + properties.setProperty("catalog.local.file", "classpath:local-catalog-cloud.yml"); + restartServer(properties); + } + + @Test + @EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+") + @EnabledIfEnvironmentVariable(named = "AWS_SECRET_ACCESS_KEY", matches = ".+") + void queriesLatestAndHistoricalMetadataForTable1() { + String bearer = shareTable("table1-metadata", "main.sales.table1", "sales.table1"); + String endpoint = PROTOCOL + "/shares/table1-metadata/schemas/sales/tables/table1/metadata"; + + // Latest metadata defaults to parquet NDJSON at table1 version 2. + Response latest = metadata(endpoint, bearer, null); + assertVersion("2", latest); + assertTrue(latest.headers().contentType().is(MediaType.parse("application/x-ndjson"))); + assertParquetMetadata(latest.text()); + assertTrue(latest.text().contains("eventTime") || latest.text().contains("date")); + + // Explicit parquet keeps protocol and schemaString, not delta wrappers. + Response parquet = metadata(endpoint, bearer, "responseformat=parquet"); + assertVersion("2", parquet); + assertParquetMetadata(parquet.text()); + // Dual responseformat prefers delta over parquet. + Response both = metadata(endpoint, bearer, "responseformat=delta,parquet"); + assertVersion("2", both); + assertDeltaMetadata(both.text()); + // Delta-only wraps Kernel protocol and metadata. + Response delta = metadata(endpoint, bearer, "responseformat=delta"); + assertVersion("2", delta); + assertDeltaMetadata(delta.text()); + + // Historical version 0, and as of the first commit time, return that snapshot. + assertVersion("0", metadata(endpoint + "?version=0", bearer, null)); + assertVersion("0", metadata(endpoint + "?timestamp=2021-05-04T17:24:06Z", bearer, null)); + // As of before version 0 is INVALID_PARAMETER_VALUE. + assertInvalid(metadata(endpoint + "?timestamp=1970-01-01T00:00:00Z", bearer, null)); + // version and timestamp together are mutually exclusive. + assertInvalid(metadata(endpoint + "?version=1×tamp=2021-05-04T17:24:06Z", bearer, null)); + } + + @Test + @EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+") + @EnabledIfEnvironmentVariable(named = "AWS_SECRET_ACCESS_KEY", matches = ".+") + void queriesDeltaMetadataForDeletionVectorsTable() { + String bearer = + shareTable("dv-metadata", "main.sales.deletionvectors", "sales.deletionvectors"); + String endpoint = + PROTOCOL + "/shares/dv-metadata/schemas/sales/tables/deletionvectors/metadata"; + + // Deletion-vector table metadata in delta format includes the deletionVectors feature. + Response delta = metadata(endpoint, bearer, "responseformat=delta"); + assertEquals(200, delta.status()); + assertTrue(delta.headers().contains("Delta-Table-Version")); + assertDeltaMetadata(delta.text()); + assertTrue(delta.text().contains("deletionVectors"), delta.text()); + + // Dual responseformat still prefers delta for a reader-v3 table. + Response both = metadata(endpoint, bearer, "responseformat=delta,parquet"); + assertEquals(200, both.status()); + assertDeltaMetadata(both.text()); + } + + @Test + @EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+") + @EnabledIfEnvironmentVariable(named = "AWS_SECRET_ACCESS_KEY", matches = ".+") + void queriesMetadataThroughASchemaGrant() { + createShare("schema-metadata"); + addObject("schema-metadata", "SCHEMA", "main.sales", "sales"); + String bearer = createAndActivateRecipient("schema-metadata-partner"); + grant("schema-metadata", "schema-metadata-partner"); + + // A SCHEMA grant can Query Table Metadata for catalog children (table1). + Response metadata = + metadata( + PROTOCOL + "/shares/schema-metadata/schemas/SALES/tables/TABLE1/metadata", + bearer, + null); + assertVersion("2", metadata); + assertParquetMetadata(metadata.text()); + } + + private static void assertParquetMetadata(String ndjson) { + assertTrue(ndjson.contains("\"protocol\""), ndjson); + assertTrue(ndjson.contains("\"metaData\""), ndjson); + assertTrue(ndjson.contains("\"schemaString\""), ndjson); + assertFalse(ndjson.contains("\"deltaProtocol\""), ndjson); + assertFalse(ndjson.contains("\"deltaMetadata\""), ndjson); + } + + private static void assertDeltaMetadata(String ndjson) { + assertTrue(ndjson.contains("\"deltaProtocol\""), ndjson); + assertTrue(ndjson.contains("\"deltaMetadata\""), ndjson); + } + + private static String shareTable(String share, String catalogName, String sharedAs) { + createShare(share); + addObject(share, "TABLE", catalogName, sharedAs); + String bearer = createAndActivateRecipient(share + "-partner"); + grant(share, share + "-partner"); + return bearer; + } + + private static Response metadata(String path, String bearer, String capabilities) { + return capabilities == null + ? server.get(path, bearer) + : server.call( + HttpMethod.GET, path, bearer, null, "delta-sharing-capabilities", capabilities); + } + + private static void assertVersion(String expected, Response response) { + assertEquals(200, response.status()); + assertEquals(expected, response.headers().get("Delta-Table-Version")); + } + + private static void assertInvalid(Response response) { + assertEquals(400, response.status()); + assertEquals(ErrorCodes.INVALID_PARAMETER_VALUE, response.errorCode()); + } +} diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableMetadataTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableMetadataTest.java new file mode 100644 index 000000000..ed00631ce --- /dev/null +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableMetadataTest.java @@ -0,0 +1,112 @@ +package io.opensharing.protocol; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.linecorp.armeria.common.HttpMethod; +import com.linecorp.armeria.common.MediaType; +import io.opensharing.TestServer.Response; +import io.opensharing.http.ErrorCodes; +import java.util.ArrayList; +import java.util.List; +import org.junit.jupiter.api.Test; + +/** Query Table Metadata through the local catalog and Kernel, on local Delta logs. */ +class ProtocolTableMetadataTest extends LocalDeltaTableSupport { + + private static final ObjectMapper JSON = new ObjectMapper(); + + @Test + void queriesGrantedTableMetadata() { + createShare("table-metadata"); + addObject("table-metadata", "TABLE", "main.sales.orders", "sales.orders"); + addObject("table-metadata", "SCHEMA", "main.hr", "hr"); + createShare("hidden-metadata"); + addObject("hidden-metadata", "TABLE", "main.sales.customers", "sales.customers"); + String bearer = createAndActivateRecipient("table-metadata-partner"); + grant("table-metadata", "table-metadata-partner"); + + // Latest metadata defaults to parquet NDJSON: protocol, then metaData. + String orders = PROTOCOL + "/shares/TABLE-METADATA/schemas/SALES/tables/ORDERS/metadata"; + Response latest = metadata(orders, bearer, null); + assertEquals(200, latest.status()); + assertTrue(latest.headers().contentType().is(MediaType.parse("application/x-ndjson"))); + assertEquals("2", latest.headers().get("Delta-Table-Version")); + assertEquals("responseformat=parquet", latest.headers().get("delta-sharing-capabilities")); + List lines = lines(latest); + assertEquals(2, lines.size()); + assertEquals(1, lines.get(0).path("protocol").path("minReaderVersion").asInt()); + JsonNode metaData = lines.get(1).path("metaData"); + assertEquals("orders", metaData.path("id").asText()); + assertEquals("parquet", metaData.path("format").path("provider").asText()); + assertTrue(metaData.path("schemaString").asText().contains("\"name\":\"id\"")); + assertEquals(location("orders"), metaData.path("location").asText()); + assertEquals("url", metaData.path("accessModes").path(0).asText()); + assertTrue(metaData.path("version").isMissingNode()); + assertTrue(metaData.path("deltaMetadata").isMissingNode()); + + // responseformat=delta wraps Kernel protocol and metadata as deltaProtocol/deltaMetadata. + Response delta = metadata(orders, bearer, "responseformat=delta"); + assertEquals(200, delta.status()); + assertEquals("responseformat=delta", delta.headers().get("delta-sharing-capabilities")); + lines = lines(delta); + JsonNode deltaProtocol = lines.get(0).path("protocol").path("deltaProtocol"); + assertEquals(1, deltaProtocol.path("minReaderVersion").asInt()); + assertEquals(2, deltaProtocol.path("minWriterVersion").asInt()); + assertEquals("orders", lines.get(1).path("metaData").path("deltaMetadata").path("id").asText()); + + // Unknown responseformat and capability keys are INVALID_PARAMETER_VALUE. + assertInvalid(metadata(orders, bearer, "responseformat=json")); + assertInvalid(metadata(orders, bearer, "unknown=true")); + + // A historical version or timestamp is echoed in metaData.version. + Response atVersion = metadata(orders + "?version=1", bearer, null); + assertEquals(200, atVersion.status()); + assertEquals("1", atVersion.headers().get("Delta-Table-Version")); + assertEquals(1, lines(atVersion).get(1).path("metaData").path("version").asLong()); + String salaries = PROTOCOL + "/shares/table-metadata/schemas/hr/tables/SALARIES/metadata"; + Response atTimestamp = metadata(salaries + "?timestamp=2024-01-01T01:30:00Z", bearer, null); + assertEquals(200, atTimestamp.status()); + assertEquals("1", atTimestamp.headers().get("Delta-Table-Version")); + + assertInvalid(metadata(orders + "?version=9", bearer, null)); + assertInvalid(metadata(orders + "?version=1×tamp=2024-01-01T01:30:00Z", bearer, null)); + assertInvalid(metadata(orders + "?timestamp=not-a-timestamp", bearer, null)); + + // A table in an ungranted share is 404, not 403. + Response hidden = + metadata( + PROTOCOL + "/shares/hidden-metadata/schemas/sales/tables/customers/metadata", + bearer, + null); + assertEquals(404, hidden.status()); + assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, hidden.errorCode()); + } + + private static Response metadata(String path, String bearer, String capabilities) { + return capabilities == null + ? server.get(path, bearer) + : server.call( + HttpMethod.GET, path, bearer, null, "delta-sharing-capabilities", capabilities); + } + + private static List lines(Response response) { + List lines = new ArrayList<>(); + for (String line : response.text().split("\n")) { + try { + lines.add(JSON.readTree(line)); + } catch (JsonProcessingException e) { + throw new AssertionError("not an NDJSON line: " + line, e); + } + } + return lines; + } + + private static void assertInvalid(Response response) { + assertEquals(400, response.status()); + assertEquals(ErrorCodes.INVALID_PARAMETER_VALUE, response.errorCode()); + } +} diff --git a/server/opensharing-server/src/test/resources/local-catalog-cloud.yml b/server/opensharing-server/src/test/resources/local-catalog-cloud.yml index 3187833e4..e93b22057 100644 --- a/server/opensharing-server/src/test/resources/local-catalog-cloud.yml +++ b/server/opensharing-server/src/test/resources/local-catalog-cloud.yml @@ -1,4 +1,4 @@ -# Local catalog used by cloud-backed table-version integration tests. +# Local catalog used by cloud-backed protocol integration tests. # ENV mode chooses AWS, Azure, or GCP credentials from each asset's URI scheme. credentials: mode: ENV @@ -28,6 +28,13 @@ assets: storageLocation: s3://delta-exchange-test/delta-exchange-test/table_startingtimestamp_test/ format: delta + - identifier: main.sales.deletionvectors + type: TABLE + attributes: + subtype: MANAGED + storageLocation: s3://delta-exchange-test/delta-exchange-test/deletion_vectors_with_dvs_dv_property_on + format: delta + - identifier: main.sales.azure type: TABLE attributes: