"}
+ * }
+ */
+public record GetTableResponse(
+ String name, String schema, String share, String shareId, String id, String format) {}
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
new file mode 100644
index 000000000..f569095fb
--- /dev/null
+++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java
@@ -0,0 +1,231 @@
+package io.opensharing.asset.table;
+
+import io.opensharing.ObjectNames;
+import io.opensharing.asset.SharedDataObjectEntity;
+import io.opensharing.asset.SharedDataObjectStore;
+import io.opensharing.asset.share.ProtocolShares;
+import io.opensharing.asset.table.delta.DeltaKernel;
+import io.opensharing.auth.AuthContext;
+import io.opensharing.auth.UserContext;
+import io.opensharing.catalog.Asset;
+import io.opensharing.catalog.AssetLocation;
+import io.opensharing.catalog.AssetPage;
+import io.opensharing.catalog.AssetType;
+import io.opensharing.catalog.CatalogConnector;
+import io.opensharing.catalog.ResolvedAsset;
+import io.opensharing.catalog.TableProperties;
+import io.opensharing.http.ApiException;
+import io.opensharing.http.ListResponse;
+import io.opensharing.http.Pagination;
+import io.opensharing.recipient.RecipientPrincipal;
+import io.opensharing.share.ShareEntity;
+import java.time.Instant;
+import java.time.format.DateTimeParseException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+import java.util.function.Supplier;
+
+/**
+ * The tables a recipient reads through the sharing protocol in a share it holds SELECT on: the
+ * tables the share includes one by one, plus the tables the catalog currently lists under each
+ * schema the share includes as a whole.
+ *
+ * Every catalog call is made as the share's owner, so the catalog decides whether the owner may
+ * still read each table.
+ */
+public class ProtocolTables {
+
+ private static final int CATALOG_PAGE_SIZE = 1000;
+
+ private final CatalogConnector catalog;
+ private final DeltaKernel kernel;
+ private final ProtocolShares shares;
+ private final SharedDataObjectStore objects;
+
+ public ProtocolTables(
+ CatalogConnector catalog,
+ DeltaKernel kernel,
+ ProtocolShares shares,
+ SharedDataObjectStore objects) {
+ this.catalog = catalog;
+ this.kernel = kernel;
+ this.shares = shares;
+ this.objects = objects;
+ }
+
+ /** Lists one page of the tables in one schema of the share, by name. */
+ public ListResponse list(
+ RecipientPrincipal recipient,
+ String share,
+ String schema,
+ Integer maxResults,
+ String pageToken) {
+ ShareEntity granted = shares.requireGranted(recipient, share);
+ return page(granted, listed(granted, inSchema(granted, schema)), maxResults, pageToken);
+ }
+
+ /** Lists one page of all the tables in the share, by schema and then name. */
+ public ListResponse listAll(
+ RecipientPrincipal recipient, String share, Integer maxResults, String pageToken) {
+ ShareEntity granted = shares.requireGranted(recipient, share);
+ return page(granted, listed(granted, objects.list(granted)), maxResults, pageToken);
+ }
+
+ /** Gets one table in one schema of the share. */
+ public GetTableResponse get(
+ RecipientPrincipal recipient, String share, String schema, String table) {
+ ShareEntity granted = shares.requireGranted(recipient, share);
+ Listed found = find(granted, schema, table);
+ return new GetTableResponse(
+ found.name(),
+ found.schema(),
+ granted.getName(),
+ granted.getId(),
+ found.id(),
+ format(found.resolved().get()));
+ }
+
+ /**
+ * The Delta version of one table in the share: the latest, or with an ISO-8601 {@code
+ * startingTimestamp} the earliest version committed at or after it.
+ */
+ public long version(
+ RecipientPrincipal recipient,
+ String share,
+ String schema,
+ String table,
+ String startingTimestamp) {
+ ShareEntity granted = shares.requireGranted(recipient, share);
+ Instant timestamp = parseTimestamp(startingTimestamp);
+ return kernel.getVersion(resolve(granted, schema, table), timestamp, owner(granted));
+ }
+
+ private List inSchema(ShareEntity share, String schema) {
+ List included =
+ objects.listInSchema(share, ObjectNames.normalize(schema));
+ if (included.isEmpty()) {
+ throw ApiException.notFound(
+ "schema '" + schema + "' does not exist in share '" + share.getName() + "'");
+ }
+ return included;
+ }
+
+ private Listed find(ShareEntity share, String schema, String table) {
+ String name = ObjectNames.normalize(table);
+ return listed(share, inSchema(share, schema)).stream()
+ .filter(listed -> name.equals(ObjectNames.normalize(listed.name())))
+ .findFirst()
+ .orElseThrow(
+ () ->
+ ApiException.notFound(
+ "table '"
+ + schema
+ + "."
+ + table
+ + "' does not exist in share '"
+ + share.getName()
+ + "'"));
+ }
+
+ private ResolvedAsset resolve(ShareEntity share, String schema, String table) {
+ return find(share, schema, table).resolved().get();
+ }
+
+ private static Instant parseTimestamp(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");
+ }
+ }
+
+ /**
+ * A table on the listing. Included tables are resolved only when they land on the requested
+ * page; catalog children arrive resolved.
+ */
+ private record Listed(String schema, String name, String id, Supplier resolved) {}
+
+ private List listed(ShareEntity share, List included) {
+ List tables = new ArrayList<>();
+ for (SharedDataObjectEntity object : included) {
+ String schema = object.getSharedAsSchema();
+ if (object.getType() == AssetType.SCHEMA) {
+ for (ResolvedAsset child : tablesIn(share, object)) {
+ String name = lastSegment(child.fullName());
+ tables.add(new Listed(schema, name, child.catalogAssetId(), () -> child));
+ }
+ } else if (object.getType() == AssetType.TABLE) {
+ Asset asset = new Asset(AssetType.TABLE, object.getName());
+ tables.add(
+ new Listed(
+ schema,
+ object.getSharedAsTable(),
+ object.getSourceAssetId(),
+ () -> catalog.resolveAsset(asset, owner(share))));
+ }
+ }
+ tables.sort(
+ Comparator.comparing(Listed::schema)
+ .thenComparing(table -> ObjectNames.normalize(table.name())));
+ return tables;
+ }
+
+ private ListResponse page(
+ ShareEntity share, List tables, Integer maxResults, String pageToken) {
+ return Pagination.protocolPage(
+ maxResults,
+ pageToken,
+ (offset, limit) -> {
+ int from = Math.min(offset, tables.size());
+ return tables.subList(from, from + Math.min(limit, tables.size() - from));
+ },
+ table -> response(share, table));
+ }
+
+ private List tablesIn(ShareEntity share, SharedDataObjectEntity schema) {
+ Asset parent = new Asset(AssetType.SCHEMA, schema.getName());
+ List tables = new ArrayList<>();
+ String pageToken = null;
+ do {
+ AssetPage page = catalog.listChildren(parent, CATALOG_PAGE_SIZE, pageToken, owner(share));
+ page.assets().stream().filter(asset -> asset.type() == AssetType.TABLE).forEach(tables::add);
+ pageToken = page.nextPageToken();
+ } while (pageToken != null);
+ return tables;
+ }
+
+ private static TableResponse response(ShareEntity share, Listed table) {
+ ResolvedAsset resolved = table.resolved().get();
+ AssetLocation location = resolved.location();
+ return new TableResponse(
+ table.name(),
+ table.schema(),
+ share.getName(),
+ share.getId(),
+ table.id(),
+ location == null ? null : location.storageLocation(),
+ location == null || location.auxiliaryLocations().isEmpty()
+ ? null
+ : location.auxiliaryLocations(),
+ TableAccessModes.forTable(resolved));
+ }
+
+ private static String format(ResolvedAsset table) {
+ return table.additionalProperties() instanceof TableProperties properties
+ && properties.dataSourceFormat() != null
+ ? properties.dataSourceFormat().wireName()
+ : null;
+ }
+
+ private static AuthContext owner(ShareEntity share) {
+ return AuthContext.of(UserContext.fromUserIdAndName(share.getOwnerId(), null));
+ }
+
+ private static String lastSegment(String fullName) {
+ return fullName.substring(fullName.lastIndexOf('.') + 1);
+ }
+}
diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableAccessModes.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableAccessModes.java
new file mode 100644
index 000000000..1742baae6
--- /dev/null
+++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableAccessModes.java
@@ -0,0 +1,25 @@
+package io.opensharing.asset.table;
+
+import io.opensharing.catalog.DataSourceFormat;
+import io.opensharing.catalog.ResolvedAsset;
+import io.opensharing.catalog.TableProperties;
+import java.util.List;
+
+/** Access modes supported by a resolved table. */
+final class TableAccessModes {
+
+ private TableAccessModes() {}
+
+ /** Managed and external Delta tables support QueryTable and temporary credentials. */
+ static List forTable(ResolvedAsset table) {
+ if (!(table.additionalProperties() instanceof TableProperties properties)
+ || properties.dataSourceFormat() != DataSourceFormat.DELTA) {
+ return null;
+ }
+ String subtype = properties.attributes().get("subtype");
+ if ("MANAGED".equalsIgnoreCase(subtype) || "EXTERNAL".equalsIgnoreCase(subtype)) {
+ return List.of("url", "dir");
+ }
+ return null;
+ }
+}
diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableResponse.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableResponse.java
new file mode 100644
index 000000000..0b9b04346
--- /dev/null
+++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableResponse.java
@@ -0,0 +1,22 @@
+package io.opensharing.asset.table;
+
+import java.util.List;
+
+/**
+ * Protocol table listed under a share schema:
+ *
+ * {@code
+ * {"name": "", "schema": "", "share": "",
+ * "shareId": "", "id": "", "location": "",
+ * "auxiliaryLocations": ["", ...], "accessModes": ["url", "dir"]}
+ * }
+ */
+public record TableResponse(
+ String name,
+ String schema,
+ String share,
+ String shareId,
+ String id,
+ String location,
+ List auxiliaryLocations,
+ List accessModes) {}
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
new file mode 100644
index 000000000..d93d7d41d
--- /dev/null
+++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java
@@ -0,0 +1,97 @@
+package io.opensharing.asset.table.delta;
+
+import io.delta.kernel.Table;
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.exceptions.KernelException;
+import io.delta.kernel.internal.TableImpl;
+import io.opensharing.auth.AuthContext;
+import io.opensharing.catalog.AssetType;
+import io.opensharing.catalog.CatalogConnector;
+import io.opensharing.catalog.CredentialRequest;
+import io.opensharing.catalog.DataSourceFormat;
+import io.opensharing.catalog.ResolvedAsset;
+import io.opensharing.catalog.StorageCredentials;
+import io.opensharing.catalog.StorageOperation;
+import io.opensharing.catalog.TableProperties;
+import io.opensharing.exception.CatalogException;
+import io.opensharing.http.ApiException;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+
+/** Opens a Kernel session against catalog-vended credentials for a Delta table root. */
+public class DeltaKernel {
+
+ private static final Duration CREDENTIAL_TTL = Duration.ofMinutes(5);
+
+ private final CatalogConnector catalog;
+
+ public DeltaKernel(CatalogConnector catalog) {
+ this.catalog = catalog;
+ }
+
+ /**
+ * The table's latest version, or with {@code startingTimestamp} the earliest version committed
+ * at or after it. Reads the table as {@code auth}.
+ */
+ public long getVersion(ResolvedAsset table, Instant startingTimestamp, AuthContext auth) {
+ Session session = open(table, auth);
+ return versionAtOrAfter((TableImpl) session.table(), session.engine(), startingTimestamp);
+ }
+
+ static long versionAtOrAfter(TableImpl table, Engine engine, Instant startingTimestamp) {
+ if (startingTimestamp == null) {
+ return table.getLatestSnapshot(engine).getVersion();
+ }
+ try {
+ return table.getVersionAtOrAfterTimestamp(engine, startingTimestamp.toEpochMilli());
+ } catch (IllegalArgumentException | KernelException invalid) {
+ throw ApiException.invalidParameter(
+ "startingTimestamp is after the latest table version timestamp");
+ }
+ }
+
+ private 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");
+ }
+ String location = table.location() == null ? null : table.location().storageLocation();
+ if (location == null || location.isBlank()) {
+ throw new CatalogException("table '" + table.fullName() + "' has no storage location");
+ }
+ List credentials =
+ catalog.getStorageCredentials(
+ new CredentialRequest(
+ AssetType.TABLE,
+ table.fullName(),
+ table.catalogAssetId(),
+ location,
+ StorageOperation.READ,
+ CREDENTIAL_TTL),
+ auth);
+ StorageCredentials rootCredentials =
+ credentials.stream()
+ .filter(candidate -> covers(candidate.prefix(), location))
+ .findFirst()
+ .orElseThrow(
+ () ->
+ new CatalogException(
+ "catalog returned no credentials for table root '" + location + "'"));
+ Engine engine =
+ DefaultEngine.create(HadoopStorageConfiguration.from(rootCredentials, location));
+ String path = HadoopStorageConfiguration.kernelPath(location);
+ return new Session(engine, Table.forPath(engine, path));
+ }
+
+ private static boolean covers(String prefix, String location) {
+ if (prefix == null || prefix.isBlank()) {
+ return false;
+ }
+ String normalized = prefix.endsWith("/") ? prefix : prefix + "/";
+ return location.equals(prefix) || location.startsWith(normalized);
+ }
+
+ private record Session(Engine engine, Table table) {}
+}
diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/GcsAccessTokenProvider.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/GcsAccessTokenProvider.java
new file mode 100644
index 000000000..42b70d171
--- /dev/null
+++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/GcsAccessTokenProvider.java
@@ -0,0 +1,34 @@
+package io.opensharing.asset.table.delta;
+
+import com.google.cloud.hadoop.util.AccessTokenProvider;
+import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+
+/** Supplies one catalog-vended OAuth token to the Hadoop GCS connector. */
+public final class GcsAccessTokenProvider implements AccessTokenProvider {
+
+ static final String TOKEN = "opensharing.gcs.oauth-token";
+ static final String EXPIRATION = "opensharing.gcs.oauth-token-expiration";
+
+ private Configuration configuration;
+
+ @Override
+ public AccessToken getAccessToken() {
+ return new AccessToken(configuration.get(TOKEN), configuration.getLong(EXPIRATION, 0));
+ }
+
+ @Override
+ public void refresh() throws IOException {
+ // The sharing request is shorter than the credential TTL; a new request obtains a fresh token.
+ }
+
+ @Override
+ public void setConf(Configuration configuration) {
+ this.configuration = configuration;
+ }
+
+ @Override
+ public Configuration getConf() {
+ return configuration;
+ }
+}
diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/HadoopStorageConfiguration.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/HadoopStorageConfiguration.java
new file mode 100644
index 000000000..0bd49f1e8
--- /dev/null
+++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/HadoopStorageConfiguration.java
@@ -0,0 +1,89 @@
+package io.opensharing.asset.table.delta;
+
+import io.opensharing.catalog.StorageCredentials;
+import io.opensharing.exception.CatalogException;
+import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
+import org.apache.hadoop.conf.Configuration;
+
+/** Builds a Hadoop {@link Configuration} from catalog-vended storage credentials. */
+final class HadoopStorageConfiguration {
+
+ private static final Duration CREDENTIAL_TTL = Duration.ofMinutes(5);
+
+ private HadoopStorageConfiguration() {}
+
+ static Configuration from(StorageCredentials credentials, String location) {
+ Configuration configuration = new Configuration();
+ switch (credentials.provider()) {
+ case AWS, R2 -> configureS3(configuration, credentials);
+ case AZURE -> configureAzure(configuration, credentials, location);
+ case GCP -> configureGcs(configuration, credentials);
+ }
+ return configuration;
+ }
+
+ /** Hadoop's S3 connector is s3a://; catalogs typically return s3://. */
+ static String kernelPath(String location) {
+ return location.startsWith("s3://") ? "s3a://" + location.substring(5) : location;
+ }
+
+ private static void configureS3(Configuration configuration, StorageCredentials credentials) {
+ configuration.set("fs.s3a.access.key", credentials.require(StorageCredentials.ACCESS_KEY_ID));
+ configuration.set(
+ "fs.s3a.secret.key", credentials.require(StorageCredentials.SECRET_ACCESS_KEY));
+ String token = credentials.credentials().get(StorageCredentials.SESSION_TOKEN);
+ if (token != null && !token.isBlank()) {
+ configuration.set("fs.s3a.session.token", token);
+ configuration.set(
+ "fs.s3a.aws.credentials.provider",
+ "org.apache.hadoop.fs.s3a.TemporaryAWSCredentialsProvider");
+ }
+ String region = credentials.credentials().get(StorageCredentials.REGION);
+ if (region != null && !region.isBlank()) {
+ configuration.set("fs.s3a.endpoint.region", region);
+ }
+ }
+
+ private static void configureAzure(
+ Configuration configuration, StorageCredentials credentials, String location) {
+ String host = URI.create(location).getHost();
+ if (host == null || host.isBlank()) {
+ throw new CatalogException("Azure table location has no storage account host");
+ }
+ String accountKey = credentials.credentials().get(StorageCredentials.AZURE_ACCOUNT_KEY);
+ if (accountKey != null && !accountKey.isBlank()) {
+ configuration.set("fs.azure.account.auth.type." + host, "SharedKey");
+ configuration.set("fs.azure.account.key." + host, accountKey);
+ return;
+ }
+ configuration.set("fs.azure.account.auth.type." + host, "SAS");
+ configuration.set(
+ "fs.azure.sas.token.provider.type." + host,
+ "org.apache.hadoop.fs.azurebfs.services.FixedSASTokenProvider");
+ configuration.set(
+ "fs.azure.sas.fixed.token." + host, credentials.require(StorageCredentials.SAS_TOKEN));
+ }
+
+ private static void configureGcs(Configuration configuration, StorageCredentials credentials) {
+ String keyFile =
+ credentials.credentials().get(StorageCredentials.GOOGLE_SERVICE_ACCOUNT_KEY_FILE);
+ if (keyFile != null && !keyFile.isBlank()) {
+ configuration.set("fs.gs.auth.type", "SERVICE_ACCOUNT_JSON_KEYFILE");
+ configuration.set("fs.gs.auth.service.account.json.keyfile", keyFile);
+ return;
+ }
+ configuration.set("fs.gs.auth.type", "ACCESS_TOKEN_PROVIDER");
+ configuration.set(
+ "fs.gs.auth.access.token.provider.impl", GcsAccessTokenProvider.class.getName());
+ configuration.set(
+ GcsAccessTokenProvider.TOKEN, credentials.require(StorageCredentials.OAUTH_TOKEN));
+ Instant expiration =
+ credentials.expiration() != null
+ ? credentials.expiration()
+ : Instant.now().plus(CREDENTIAL_TTL);
+ configuration.set(
+ GcsAccessTokenProvider.EXPIRATION, Long.toString(expiration.toEpochMilli()));
+ }
+}
diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/catalog/StorageCredentials.java b/server/opensharing-server-core/src/main/java/io/opensharing/catalog/StorageCredentials.java
index 04247b063..5623a5844 100644
--- a/server/opensharing-server-core/src/main/java/io/opensharing/catalog/StorageCredentials.java
+++ b/server/opensharing-server-core/src/main/java/io/opensharing/catalog/StorageCredentials.java
@@ -26,10 +26,19 @@ public record StorageCredentials(
/** Shared Access Signature token scoped to the storage prefix. */
public static final String SAS_TOKEN = "sasToken";
+ /** Storage account key. Local integration tests only; production catalogs vend a SAS token. */
+ public static final String AZURE_ACCOUNT_KEY = "azureAccountKey";
+
// GCP
/** OAuth 2.0 access token for Google Cloud Storage. */
public static final String OAUTH_TOKEN = "oauthToken";
+ /**
+ * Path to a service account JSON key file. Local integration tests only; production catalogs
+ * vend an OAuth token.
+ */
+ public static final String GOOGLE_SERVICE_ACCOUNT_KEY_FILE = "googleServiceAccountKeyFile";
+
public StorageCredentials {
credentials = credentials == null ? Map.of() : Map.copyOf(credentials);
}
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
new file mode 100644
index 000000000..e39570a8f
--- /dev/null
+++ b/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java
@@ -0,0 +1,47 @@
+package io.opensharing.protocol;
+
+import com.linecorp.armeria.common.HttpResponse;
+import com.linecorp.armeria.common.HttpStatus;
+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.Param;
+import com.linecorp.armeria.server.annotation.ProducesJson;
+import io.opensharing.asset.table.ProtocolTables;
+
+/**
+ * The Delta Sharing protocol endpoints for one shared table, which recipients call with their
+ * bearer token, served behind {@link RecipientTokenDecorator}.
+ */
+@Blocking
+@ProducesJson
+public final class DeltaSharingProtocolService {
+
+ private static final String DELTA_TABLE_VERSION = "Delta-Table-Version";
+
+ private final ProtocolTables tables;
+
+ public DeltaSharingProtocolService(ProtocolTables tables) {
+ this.tables = tables;
+ }
+
+ /**
+ * {@code GET /shares/{share}/schemas/{schema}/tables/{table}/version}: the table's Delta version
+ * in the {@code Delta-Table-Version} header, with no body.
+ */
+ @Get("/shares/{share}/schemas/{schema}/tables/{table}/version")
+ public HttpResponse getTableVersion(
+ ServiceRequestContext ctx,
+ @Param String share,
+ @Param String schema,
+ @Param String table,
+ @Param @Nullable String startingTimestamp) {
+ long version =
+ tables.version(
+ RecipientTokenDecorator.recipient(ctx), share, schema, table, startingTimestamp);
+ return HttpResponse.of(
+ ResponseHeaders.of(HttpStatus.OK, DELTA_TABLE_VERSION, Long.toString(version)));
+ }
+}
diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/protocol/OpenSharingProtocolService.java b/server/opensharing-server-core/src/main/java/io/opensharing/protocol/OpenSharingProtocolService.java
index 208e52d70..bfc381b71 100644
--- a/server/opensharing-server-core/src/main/java/io/opensharing/protocol/OpenSharingProtocolService.java
+++ b/server/opensharing-server-core/src/main/java/io/opensharing/protocol/OpenSharingProtocolService.java
@@ -6,23 +6,33 @@
import com.linecorp.armeria.server.annotation.Get;
import com.linecorp.armeria.server.annotation.Param;
import com.linecorp.armeria.server.annotation.ProducesJson;
+import io.opensharing.asset.schema.ProtocolSchemas;
+import io.opensharing.asset.schema.SchemaResponse;
import io.opensharing.asset.share.GetShareResponse;
import io.opensharing.asset.share.ProtocolShares;
+import io.opensharing.asset.table.GetTableResponse;
+import io.opensharing.asset.table.ProtocolTables;
+import io.opensharing.asset.table.TableResponse;
import io.opensharing.http.ListResponse;
import io.opensharing.share.ShareResponse;
/**
- * The Delta Sharing protocol API: every endpoint recipients call with their bearer token, served
- * behind {@link RecipientTokenDecorator}.
+ * The OpenSharing protocol endpoints that list and get shares, schemas and tables, which recipients
+ * call with their bearer token, served behind {@link RecipientTokenDecorator}.
*/
@Blocking
@ProducesJson
public final class OpenSharingProtocolService {
private final ProtocolShares shares;
+ private final ProtocolSchemas schemas;
+ private final ProtocolTables tables;
- public OpenSharingProtocolService(ProtocolShares shares) {
+ public OpenSharingProtocolService(
+ ProtocolShares shares, ProtocolSchemas schemas, ProtocolTables tables) {
this.shares = shares;
+ this.schemas = schemas;
+ this.tables = tables;
}
/** {@code GET /shares}: lists one page of the shares the recipient holds SELECT on. */
@@ -39,4 +49,43 @@ public ListResponse listShares(
public GetShareResponse getShare(ServiceRequestContext ctx, @Param String share) {
return shares.get(RecipientTokenDecorator.recipient(ctx), share);
}
+
+ /** {@code GET /shares/{share}/schemas}: lists one page of the schemas in a granted share. */
+ @Get("/shares/{share}/schemas")
+ public ListResponse listSchemas(
+ ServiceRequestContext ctx,
+ @Param String share,
+ @Param @Nullable Integer maxResults,
+ @Param @Nullable String pageToken) {
+ return schemas.list(RecipientTokenDecorator.recipient(ctx), share, maxResults, pageToken);
+ }
+
+ /** {@code GET /shares/{share}/schemas/{schema}/tables}: lists one page of a schema's tables. */
+ @Get("/shares/{share}/schemas/{schema}/tables")
+ public ListResponse listTables(
+ ServiceRequestContext ctx,
+ @Param String share,
+ @Param String schema,
+ @Param @Nullable Integer maxResults,
+ @Param @Nullable String pageToken) {
+ return tables.list(
+ RecipientTokenDecorator.recipient(ctx), share, schema, maxResults, pageToken);
+ }
+
+ /** {@code GET /shares/{share}/all-tables}: lists one page of all the tables in a share. */
+ @Get("/shares/{share}/all-tables")
+ public ListResponse listAllTables(
+ ServiceRequestContext ctx,
+ @Param String share,
+ @Param @Nullable Integer maxResults,
+ @Param @Nullable String pageToken) {
+ return tables.listAll(RecipientTokenDecorator.recipient(ctx), share, maxResults, pageToken);
+ }
+
+ /** {@code GET /shares/{share}/schemas/{schema}/tables/{table}}: gets a table in a schema. */
+ @Get("/shares/{share}/schemas/{schema}/tables/{table}")
+ public GetTableResponse getTable(
+ ServiceRequestContext ctx, @Param String share, @Param String schema, @Param String table) {
+ return tables.get(RecipientTokenDecorator.recipient(ctx), share, schema, table);
+ }
}
diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/protocol/OpenSharingRecipientApi.java b/server/opensharing-server-core/src/main/java/io/opensharing/protocol/OpenSharingRecipientApi.java
index 4c0bf1de8..410e9d40d 100644
--- a/server/opensharing-server-core/src/main/java/io/opensharing/protocol/OpenSharingRecipientApi.java
+++ b/server/opensharing-server-core/src/main/java/io/opensharing/protocol/OpenSharingRecipientApi.java
@@ -1,5 +1,6 @@
package io.opensharing.protocol;
+import com.linecorp.armeria.server.AnnotatedServiceBindingBuilder;
import com.linecorp.armeria.server.ServerBuilder;
import com.linecorp.armeria.server.annotation.JacksonRequestConverterFunction;
import com.linecorp.armeria.server.annotation.JacksonResponseConverterFunction;
@@ -30,16 +31,15 @@ public static void mount(
String activationBasePath,
OpenSharing openSharing) {
ApiExceptionHandler errors = new ApiExceptionHandler();
- // The token check must sit inside the error decorator, so that its 401s get the protocol body.
- builder
- .annotatedService()
- .pathPrefix(protocolBasePath)
- .requestConverters(new JacksonRequestConverterFunction(ApiJson.MAPPER))
- .responseConverters(new JacksonResponseConverterFunction(ApiJson.MAPPER))
- .exceptionHandlers(errors)
- .decorator(new RecipientTokenDecorator(openSharing.recipients()))
- .decorator(errors)
- .build(new OpenSharingProtocolService(openSharing.protocolShares()));
+ RecipientTokenDecorator tokens = new RecipientTokenDecorator(openSharing.recipients());
+ protocol(builder, protocolBasePath, errors, tokens)
+ .build(
+ new OpenSharingProtocolService(
+ openSharing.protocolShares(),
+ openSharing.protocolSchemas(),
+ openSharing.protocolTables()));
+ protocol(builder, protocolBasePath, errors, tokens)
+ .build(new DeltaSharingProtocolService(openSharing.protocolTables()));
builder
.annotatedService()
.pathPrefix(activationBasePath)
@@ -47,4 +47,20 @@ public static void mount(
.decorator(errors)
.build(new ActivationService(openSharing.recipients(), publicUrl + protocolBasePath));
}
+
+ private static AnnotatedServiceBindingBuilder protocol(
+ ServerBuilder builder,
+ String basePath,
+ ApiExceptionHandler errors,
+ RecipientTokenDecorator tokens) {
+ // The token check must sit inside the error decorator, so that its 401s get the protocol body.
+ return builder
+ .annotatedService()
+ .pathPrefix(basePath)
+ .requestConverters(new JacksonRequestConverterFunction(ApiJson.MAPPER))
+ .responseConverters(new JacksonResponseConverterFunction(ApiJson.MAPPER))
+ .exceptionHandlers(errors)
+ .decorator(tokens)
+ .decorator(errors);
+ }
}
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 d576f1d53..1aba934f9 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
@@ -3,7 +3,10 @@
import io.opensharing.Transactions;
import io.opensharing.asset.SharedDataObjectEntity;
import io.opensharing.asset.SharedDataObjectStore;
+import io.opensharing.asset.schema.ProtocolSchemas;
import io.opensharing.asset.share.ProtocolShares;
+import io.opensharing.asset.table.ProtocolTables;
+import io.opensharing.asset.table.delta.DeltaKernel;
import io.opensharing.catalog.CatalogConnector;
import io.opensharing.recipient.RecipientEntity;
import io.opensharing.recipient.RecipientService;
@@ -27,22 +30,22 @@ public final class OpenSharing {
private final ShareService shares;
private final RecipientService recipients;
private final ProtocolShares protocolShares;
+ private final ProtocolSchemas protocolSchemas;
+ private final ProtocolTables protocolTables;
private OpenSharing(CatalogConnector catalog, Transactions tx) {
this.catalog = catalog;
ShareStore shareStore = new ShareStore(tx);
+ SharedDataObjectStore objectStore = new SharedDataObjectStore(tx);
SharePermissionStore permissionStore = new SharePermissionStore(tx);
RecipientStore recipientStore = new RecipientStore(tx);
this.shares =
- new ShareService(
- catalog,
- tx,
- shareStore,
- new SharedDataObjectStore(tx),
- permissionStore,
- recipientStore);
+ new ShareService(catalog, tx, shareStore, objectStore, permissionStore, recipientStore);
this.recipients = new RecipientService(catalog, recipientStore);
this.protocolShares = new ProtocolShares(shareStore, permissionStore);
+ this.protocolSchemas = new ProtocolSchemas(protocolShares, objectStore);
+ this.protocolTables =
+ new ProtocolTables(catalog, new DeltaKernel(catalog), protocolShares, objectStore);
}
public static Builder builder() {
@@ -76,6 +79,16 @@ public ProtocolShares protocolShares() {
return protocolShares;
}
+ /** The schemas recipients see in their shares through the sharing protocol. */
+ public ProtocolSchemas protocolSchemas() {
+ return protocolSchemas;
+ }
+
+ /** The tables recipients see in their shares through the sharing protocol. */
+ public ProtocolTables protocolTables() {
+ return protocolTables;
+ }
+
public static final class Builder {
private CatalogConnector catalog;
diff --git a/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogConnector.java b/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogConnector.java
index 018a72d21..5a7237d68 100644
--- a/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogConnector.java
+++ b/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogConnector.java
@@ -20,6 +20,7 @@
import io.opensharing.catalog.TableProperties;
import io.opensharing.catalog.DataSourceFormat;
import io.opensharing.exception.UnsupportedAssetTypeException;
+import java.net.URI;
import java.security.SecureRandom;
import java.time.Duration;
import java.time.Instant;
@@ -33,7 +34,10 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/** File-backed catalog for local testing: YAML lookup and placeholder or static credentials. */
+/**
+ * File-backed catalog for local testing: YAML lookup and placeholder, static, or environment
+ * credentials.
+ */
public final class LocalCatalogConnector implements CatalogConnector {
public static final String NAME = "local";
@@ -188,11 +192,16 @@ public List getStorageCredentials(
}
Duration ttl = request.ttl() != null ? request.ttl() : configuredTtl();
Instant expiration = Instant.now().plus(ttl);
- CloudProvider provider = credentials.provider();
+ CloudProvider provider =
+ credentials.mode() == LocalCatalogFile.CredentialMode.ENV
+ ? providerFor(location)
+ : credentials.provider();
Map values =
- credentials.mode() == LocalCatalogFile.CredentialMode.STATIC
- ? staticValues(provider)
- : fakeValues(provider, expiration);
+ switch (credentials.mode()) {
+ case STATIC -> staticValues(provider);
+ case ENV -> envValues(provider);
+ case FAKE -> fakeValues(provider, expiration);
+ };
return List.of(new StorageCredentials(location, provider, values, expiration));
}
@@ -230,6 +239,65 @@ private Duration configuredTtl() {
return seconds == null ? DEFAULT_TTL : Duration.ofSeconds(seconds);
}
+ private Map envValues(CloudProvider provider) {
+ Map values = new LinkedHashMap<>();
+ switch (provider) {
+ case AWS, R2 -> {
+ values.put(StorageCredentials.ACCESS_KEY_ID, requireEnv("AWS_ACCESS_KEY_ID"));
+ values.put(StorageCredentials.SECRET_ACCESS_KEY, requireEnv("AWS_SECRET_ACCESS_KEY"));
+ String region = firstEnv("AWS_REGION", "AWS_DEFAULT_REGION");
+ if (region != null) {
+ values.put(StorageCredentials.REGION, region);
+ }
+ String session = firstEnv("AWS_SESSION_TOKEN");
+ if (session != null) {
+ values.put(StorageCredentials.SESSION_TOKEN, session);
+ }
+ }
+ case AZURE ->
+ values.put(StorageCredentials.AZURE_ACCOUNT_KEY, requireEnv("AZURE_TEST_ACCOUNT_KEY"));
+ case GCP ->
+ values.put(
+ StorageCredentials.GOOGLE_SERVICE_ACCOUNT_KEY_FILE,
+ requireEnv("GOOGLE_APPLICATION_CREDENTIALS"));
+ }
+ return values;
+ }
+
+ private static CloudProvider providerFor(String location) {
+ String scheme = URI.create(location).getScheme();
+ if (scheme == null) {
+ throw new CatalogException("storage location '" + location + "' has no URI scheme");
+ }
+ return switch (scheme.toLowerCase(Locale.ROOT)) {
+ case "s3", "s3a" -> CloudProvider.AWS;
+ case "abfs", "abfss", "wasb", "wasbs" -> CloudProvider.AZURE;
+ case "gs" -> CloudProvider.GCP;
+ default ->
+ throw new CatalogException(
+ "cannot select environment credentials for storage scheme '" + scheme + "'");
+ };
+ }
+
+ private static String requireEnv(String name) {
+ String value = firstEnv(name);
+ if (value == null) {
+ throw new CatalogException(
+ "local catalog credentials.mode is ENV but environment variable '" + name + "' is unset");
+ }
+ return value;
+ }
+
+ private static String firstEnv(String... names) {
+ for (String name : names) {
+ String value = System.getenv(name);
+ if (value != null && !value.isBlank()) {
+ return value;
+ }
+ }
+ return null;
+ }
+
private Map staticValues(CloudProvider provider) {
Map values = new LinkedHashMap<>();
for (String key : requiredKeys(provider)) {
diff --git a/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogFile.java b/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogFile.java
index 8cdda32ce..f557f0f7b 100644
--- a/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogFile.java
+++ b/server/opensharing-server/src/main/java/io/opensharing/catalog/local/LocalCatalogFile.java
@@ -23,7 +23,8 @@ public LocalCatalogFile(Credentials credentials, List assets) {
public enum CredentialMode {
FAKE,
- STATIC
+ STATIC,
+ ENV
}
/** Provider identity the local catalog accepts for admin APIs, holding only its privileges. */
diff --git a/server/opensharing-server/src/main/resources/local-catalog.yml b/server/opensharing-server/src/main/resources/local-catalog.yml
index 4c01c06fd..2274d6dea 100644
--- a/server/opensharing-server/src/main/resources/local-catalog.yml
+++ b/server/opensharing-server/src/main/resources/local-catalog.yml
@@ -46,3 +46,34 @@ assets:
subtype: MANAGED
storageLocation: s3://test/main.sales.orders/
format: delta
+
+ - identifier: main.sales.customers
+ type: TABLE
+ attributes:
+ subtype: MANAGED
+ storageLocation: s3://test/main.sales.customers/
+ format: delta
+
+ - identifier: main.hr
+ type: SCHEMA
+
+ - identifier: main.hr.employees
+ type: TABLE
+ attributes:
+ subtype: MANAGED
+ storageLocation: s3://test/main.hr.employees/
+ format: delta
+
+ - identifier: main.hr.salaries
+ type: TABLE
+ attributes:
+ subtype: MANAGED
+ storageLocation: s3://test/main.hr.salaries/
+ format: delta
+
+ - identifier: main.ops.people
+ type: TABLE
+ attributes:
+ subtype: MANAGED
+ storageLocation: s3://test/main.ops.people/
+ format: delta
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 bc085aaec..193932094 100644
--- a/server/opensharing-server/src/test/java/io/opensharing/TestServer.java
+++ b/server/opensharing-server/src/test/java/io/opensharing/TestServer.java
@@ -33,8 +33,14 @@ private TestServer(OpenSharingServer server) {
/** Starts on a free port that is also the public URL, so links in responses point back to it. */
public static TestServer start() {
+ return start(new Properties());
+ }
+
+ /** Starts with {@code overrides} on top of {@code server.properties}, like {@link #start()}. */
+ public static TestServer start(Properties overrides) {
int port = freePort();
Properties props = new Properties();
+ props.putAll(overrides);
props.setProperty("server.port", Integer.toString(port));
props.setProperty("server.public-url", "http://127.0.0.1:" + port);
return new TestServer(OpenSharingServer.start(new ServerProperties(props)));
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
new file mode 100644
index 000000000..9f41048f4
--- /dev/null
+++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/LocalDeltaTableSupport.java
@@ -0,0 +1,104 @@
+package io.opensharing.protocol;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.attribute.FileTime;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Properties;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Protocol tests that read Delta tables through Kernel. The server's local catalog points at Delta
+ * logs written to a temporary directory: {@code main.sales.orders} with 3 commits, {@code
+ * main.sales.customers} with 1, and {@code main.hr.salaries} with 4. Commit {@code n} of each table
+ * is timestamped 2024-01-01T00:00:00Z plus {@code n} hours, and table {@code t} has metadata id
+ * {@code t} and one {@code long} column, {@code id}.
+ */
+abstract class LocalDeltaTableSupport extends ProtocolApiSupport {
+
+ private static final Instant COMMIT_TIME = Instant.parse("2024-01-01T00:00:00Z");
+
+ private static final String PROTOCOL_ACTION =
+ "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}\n";
+ private static final String METADATA_ACTION =
+ "{\"metaData\":{\"id\":\"%s\",\"format\":{\"provider\":\"parquet\",\"options\":{}},"
+ + "\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[{\\\"name\\\":"
+ + "\\\"id\\\",\\\"type\\\":\\\"long\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}]}\","
+ + "\"partitionColumns\":[],\"configuration\":{},\"createdTime\":0}}\n";
+ private static final String ADD_ACTION =
+ "{\"add\":{\"path\":\"part-%d.parquet\",\"partitionValues\":{},\"size\":1,"
+ + "\"modificationTime\":0,\"dataChange\":true}}\n";
+
+ @TempDir static Path tables;
+
+ @BeforeAll
+ static void startLocalDeltaServer() {
+ String orders = deltaTable("orders", 3);
+ String customers = deltaTable("customers", 1);
+ String salaries = deltaTable("salaries", 4);
+ String catalog =
+ """
+ principals:
+ - bearerToken: alice-token
+ userId: catalog-alice-id
+ userName: alice
+ privileges: [CREATE_SHARE, CREATE_RECIPIENT, USE_SHARE, USE_RECIPIENT]
+ assets:
+ - identifier: main.sales
+ type: SCHEMA
+ - identifier: main.sales.orders
+ attributes:
+ subtype: MANAGED
+ storageLocation: %s
+ format: delta
+ - identifier: main.sales.customers
+ attributes:
+ subtype: MANAGED
+ storageLocation: %s
+ format: delta
+ - identifier: main.hr
+ type: SCHEMA
+ - identifier: main.hr.salaries
+ attributes:
+ subtype: MANAGED
+ storageLocation: %s
+ format: delta
+ """
+ .formatted(orders, customers, salaries);
+ Properties properties = new Properties();
+ properties.setProperty("catalog.local.file", write("local-catalog.yml", catalog).toString());
+ restartServer(properties);
+ }
+
+ /** 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++) {
+ String actions = version == 0 ? PROTOCOL_ACTION + METADATA_ACTION.formatted(name) : "";
+ Path commit =
+ write(
+ name + "/_delta_log/%020d.json".formatted(version),
+ actions + ADD_ACTION.formatted(version));
+ try {
+ Files.setLastModifiedTime(
+ commit, FileTime.from(COMMIT_TIME.plus(Duration.ofHours(version))));
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+ return tables.resolve(name).toUri().toString();
+ }
+
+ private static Path write(String relativePath, String content) {
+ Path file = tables.resolve(relativePath);
+ try {
+ Files.createDirectories(file.getParent());
+ return Files.writeString(file, content);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+}
diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolApiSupport.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolApiSupport.java
index 053be8fed..e25628d4a 100644
--- a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolApiSupport.java
+++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolApiSupport.java
@@ -7,6 +7,7 @@
import java.net.URI;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
+import java.util.Properties;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
@@ -33,11 +34,32 @@ static void stopServer() {
server.close();
}
+ /** Replaces the default server with one started with {@code overrides}. */
+ protected static void restartServer(Properties overrides) {
+ server.close();
+ server = TestServer.start(overrides);
+ }
+
+ protected static void createShare(String name) {
+ String body = "{\"name\":\"%s\"}".formatted(name);
+ assertEquals(201, server.post(PROVIDER + "/shares", ALICE, body).status());
+ }
+
protected static void createShare(String name, String displayName) {
String body = "{\"name\":\"%s\",\"displayName\":\"%s\"}".formatted(name, displayName);
assertEquals(201, server.post(PROVIDER + "/shares", ALICE, body).status());
}
+ /** Adds the catalog asset {@code name} of {@code type} to the share as {@code sharedAs}. */
+ protected static void addObject(String share, String type, String name, String sharedAs) {
+ String body =
+ "{\"updates\":[{\"action\":\"ADD\",\"dataObject\":"
+ + "{\"name\":\"%s\",\"type\":\"%s\",\"sharedAs\":\"%s\"}}]}";
+ Response added =
+ server.patch(PROVIDER + "/shares/" + share, ALICE, body.formatted(name, type, sharedAs));
+ assertEquals(200, added.status());
+ }
+
/** Creates a TOKEN recipient, redeems its activation URL, and returns the bearer token. */
protected static String createAndActivateRecipient(String name) {
Response created =
diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolGetTableTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolGetTableTest.java
new file mode 100644
index 000000000..29d572698
--- /dev/null
+++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolGetTableTest.java
@@ -0,0 +1,70 @@
+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.fasterxml.jackson.databind.JsonNode;
+import io.opensharing.TestServer.Response;
+import io.opensharing.http.ErrorCodes;
+import org.junit.jupiter.api.Test;
+
+class ProtocolGetTableTest extends ProtocolApiSupport {
+
+ @Test
+ void getsAnIncludedTable() {
+ createShare("get-table");
+ addObject("get-table", "TABLE", "main.sales.orders", "sales.orders");
+ String bearer = createAndActivateRecipient("get-table-partner");
+ grant("get-table", "get-table-partner");
+
+ // Share, schema and table lookups are case-insensitive.
+ Response got = server.get(PROTOCOL + "/shares/GET-TABLE/schemas/SALES/tables/ORDERS", bearer);
+ assertEquals(200, got.status());
+ JsonNode table = got.body();
+ assertEquals("orders", table.path("name").asText());
+ assertEquals("sales", table.path("schema").asText());
+ assertEquals("get-table", table.path("share").asText());
+ assertFalse(table.path("shareId").asText().isEmpty());
+ assertEquals("main.sales.orders", table.path("id").asText());
+ assertEquals("delta", table.path("format").asText());
+ assertTrue(table.path("location").isMissingNode());
+ }
+
+ @Test
+ void getsACatalogChildOfASharedSchema() {
+ createShare("get-schema-table");
+ addObject("get-schema-table", "SCHEMA", "main.hr", "hr");
+ String bearer = createAndActivateRecipient("get-schema-table-partner");
+ grant("get-schema-table", "get-schema-table-partner");
+
+ Response got =
+ server.get(PROTOCOL + "/shares/get-schema-table/schemas/hr/tables/salaries", bearer);
+ assertEquals(200, got.status());
+ assertEquals("salaries", got.body().path("name").asText());
+ assertEquals("hr", got.body().path("schema").asText());
+ assertEquals("main.hr.salaries", got.body().path("id").asText());
+ assertEquals("delta", got.body().path("format").asText());
+ }
+
+ @Test
+ void reportsMissingTablesSchemasAndShares() {
+ createShare("get-missing");
+ addObject("get-missing", "TABLE", "main.sales.orders", "sales.orders");
+ createShare("get-hidden");
+ addObject("get-hidden", "TABLE", "main.sales.orders", "sales.orders");
+ String bearer = createAndActivateRecipient("get-missing-partner");
+ grant("get-missing", "get-missing-partner");
+
+ String tables = PROTOCOL + "/shares/get-missing/schemas/";
+ assertNotFound(server.get(tables + "sales/tables/customers", bearer));
+ assertNotFound(server.get(tables + "missing/tables/orders", bearer));
+ // An ungranted share is reported as missing.
+ assertNotFound(server.get(PROTOCOL + "/shares/get-hidden/schemas/sales/tables/orders", bearer));
+ }
+
+ private static void assertNotFound(Response response) {
+ assertEquals(404, response.status());
+ assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, response.errorCode());
+ }
+}
diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolSchemasTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolSchemasTest.java
new file mode 100644
index 000000000..f2f263d71
--- /dev/null
+++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolSchemasTest.java
@@ -0,0 +1,85 @@
+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.fasterxml.jackson.databind.JsonNode;
+import com.linecorp.armeria.common.MediaType;
+import io.opensharing.TestServer.Response;
+import io.opensharing.http.ErrorCodes;
+import java.nio.charset.StandardCharsets;
+import org.junit.jupiter.api.Test;
+
+class ProtocolSchemasTest extends ProtocolApiSupport {
+
+ @Test
+ void listsDistinctSchemasInAGrantedShare() {
+ createShare("schema-share");
+ addObject("schema-share", "TABLE", "main.sales.table1", "alpha.t1");
+ addObject("schema-share", "TABLE", "main.sales.orders", "alpha.t2");
+ addObject("schema-share", "SCHEMA", "main.hr", "beta");
+ addObject("schema-share", "TABLE", "main.ops.people", "gamma.t3");
+ createShare("hidden-share");
+ addObject("hidden-share", "TABLE", "main.sales.table1", "hidden.t");
+ String bearer = createAndActivateRecipient("schema-partner");
+ grant("schema-share", "schema-partner");
+
+ // Distinct shared-as schemas, ordered by name; the share lookup is case-insensitive.
+ Response list = server.get(PROTOCOL + "/shares/SCHEMA-SHARE/schemas", bearer);
+ assertEquals(200, list.status());
+ MediaType contentType = list.headers().contentType();
+ assertTrue(contentType.is(MediaType.JSON), contentType.toString());
+ assertEquals(StandardCharsets.UTF_8, contentType.charset());
+ JsonNode items = list.body().path("items");
+ assertEquals(3, items.size());
+ assertEquals("alpha", items.path(0).path("name").asText());
+ assertEquals("schema-share", items.path(0).path("share").asText());
+ assertEquals("beta", items.path(1).path("name").asText());
+ assertEquals("gamma", items.path(2).path("name").asText());
+ assertTrue(list.body().path("nextPageToken").isMissingNode());
+
+ // Ungranted and missing shares are both reported as missing.
+ Response hidden = server.get(PROTOCOL + "/shares/hidden-share/schemas", bearer);
+ assertEquals(404, hidden.status());
+ assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, hidden.errorCode());
+ Response missing = server.get(PROTOCOL + "/shares/missing/schemas", bearer);
+ assertEquals(404, missing.status());
+ assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, missing.errorCode());
+ }
+
+ @Test
+ void paginatesSchemasAndAllowsAnEmptyGrantedShare() {
+ createShare("paged-schemas");
+ addObject("paged-schemas", "TABLE", "main.sales.table1", "alpha.t");
+ addObject("paged-schemas", "TABLE", "main.sales.orders", "beta.t");
+ createShare("empty-share");
+ String bearer = createAndActivateRecipient("schema-paged");
+ grant("paged-schemas", "schema-paged");
+ grant("empty-share", "schema-paged");
+
+ String schemas = PROTOCOL + "/shares/paged-schemas/schemas";
+ Response first = server.get(page(schemas, 1, null), bearer);
+ assertEquals(200, first.status());
+ assertEquals(1, first.body().path("items").size());
+ assertEquals("alpha", first.body().path("items").path(0).path("name").asText());
+ String pageToken = first.body().path("nextPageToken").asText();
+ assertFalse(pageToken.isEmpty());
+ Response second = server.get(page(schemas, 1, pageToken), bearer);
+ assertEquals(200, second.status());
+ assertEquals(1, second.body().path("items").size());
+ assertEquals("beta", second.body().path("items").path(0).path("name").asText());
+ assertTrue(second.body().path("nextPageToken").isMissingNode());
+
+ // A page size of 0 returns no schemas, but a token to resume from.
+ Response zero = server.get(page(schemas, 0, null), bearer);
+ assertEquals(0, zero.body().path("items").size());
+ assertFalse(zero.body().path("nextPageToken").isMissingNode());
+
+ // A granted share with no objects lists no schemas.
+ Response empty = server.get(PROTOCOL + "/shares/empty-share/schemas", bearer);
+ assertEquals(200, empty.status());
+ assertEquals(0, empty.body().path("items").size());
+ assertTrue(empty.body().path("nextPageToken").isMissingNode());
+ }
+}
diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableVersionIntegrationTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableVersionIntegrationTest.java
new file mode 100644
index 000000000..f5a7242aa
--- /dev/null
+++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableVersionIntegrationTest.java
@@ -0,0 +1,130 @@
+package io.opensharing.protocol;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.opensharing.TestServer.Response;
+import io.opensharing.http.ErrorCodes;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+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 Version through the local catalog, Kernel, and real cloud Delta logs.
+ * Individual cloud cases are skipped unless their repository credentials are present.
+ */
+@Timeout(60)
+class ProtocolTableVersionIntegrationTest 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 queriesLatestAndTimestampVersionsForTable1() {
+ String bearer = shareTable("table1-version", "main.sales.table1", "sales.table1");
+ String endpoint = PROTOCOL + "/shares/table1-version/schemas/sales/tables/table1/version";
+
+ Response latest = version(endpoint, bearer, null);
+ assertVersion("2", latest);
+ assertTrue(latest.body().isMissingNode());
+ assertVersion("0", version(endpoint, bearer, "1970-01-01T00:00:00Z"));
+ assertInvalid(version(endpoint, bearer, "2099-01-01T00:00:00Z"));
+ assertInvalid(version(endpoint, bearer, "not-a-timestamp"));
+ }
+
+ @Test
+ @EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+")
+ @EnabledIfEnvironmentVariable(named = "AWS_SECRET_ACCESS_KEY", matches = ".+")
+ void queriesAtOrAfterTimestampsOnTheStartingTimestampTable() {
+ String bearer =
+ shareTable("ts-version", "main.sales.startingtimestamp", "sales.startingtimestamp");
+ String endpoint =
+ PROTOCOL + "/shares/ts-version/schemas/sales/tables/startingtimestamp/version";
+
+ assertVersion("5", version(endpoint, bearer, null));
+ assertVersion("0", version(endpoint, bearer, "2023-05-09T08:00:00Z"));
+ assertVersion("3", version(endpoint, bearer, "2023-05-09T08:32:55.517Z"));
+ assertVersion("4", version(endpoint, bearer, "2023-05-09T08:45:00Z"));
+ assertVersion("5", version(endpoint, bearer, "2023-05-09T09:09:58.088Z"));
+ assertInvalid(version(endpoint, bearer, "2023-05-09T10:00:00Z"));
+ }
+
+ @Test
+ @EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+")
+ @EnabledIfEnvironmentVariable(named = "AWS_SECRET_ACCESS_KEY", matches = ".+")
+ void queriesVersionThroughASchemaGrant() {
+ createShare("schema-version");
+ addObject("schema-version", "SCHEMA", "main.sales", "sales");
+ String bearer = createAndActivateRecipient("schema-version-partner");
+ grant("schema-version", "schema-version-partner");
+
+ String tables = PROTOCOL + "/shares/schema-version/schemas/";
+ assertVersion("2", version(tables + "SALES/tables/TABLE1/version", bearer, null));
+ assertVersion(
+ "2",
+ version(
+ tables + "sales/tables/startingtimestamp/version", bearer, "2023-05-09T08:32:00Z"));
+ }
+
+ @Test
+ @EnabledIfEnvironmentVariable(named = "AZURE_TEST_ACCOUNT_KEY", matches = ".+")
+ void queriesAzureTableVersions() {
+ queryCloudTableVersions("azure-version", "main.sales.azure", "sales.azure");
+ }
+
+ @Test
+ @EnabledIfEnvironmentVariable(named = "GOOGLE_APPLICATION_CREDENTIALS", matches = ".+")
+ void queriesGoogleTableVersions() {
+ queryCloudTableVersions("gcs-version", "main.sales.gcs", "sales.gcs");
+ }
+
+ 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 void queryCloudTableVersions(String share, String catalogName, String sharedAs) {
+ String bearer = shareTable(share, catalogName, sharedAs);
+ String[] alias = sharedAs.split("\\.");
+ String endpoint =
+ PROTOCOL + "/shares/" + share + "/schemas/" + alias[0] + "/tables/" + alias[1] + "/version";
+
+ Response latest = version(endpoint, bearer, null);
+ assertEquals(200, latest.status());
+ String header = latest.headers().get("Delta-Table-Version");
+ assertTrue(header != null && Long.parseLong(header) >= 0, "latest version was " + header);
+ assertVersion("0", version(endpoint, bearer, "1970-01-01T00:00:00Z"));
+ assertInvalid(version(endpoint, bearer, "2099-01-01T00:00:00Z"));
+ }
+
+ private static Response version(String path, String bearer, String startingTimestamp) {
+ String query =
+ startingTimestamp == null
+ ? ""
+ : "?startingTimestamp=" + URLEncoder.encode(startingTimestamp, StandardCharsets.UTF_8);
+ return server.get(path + query, bearer);
+ }
+
+ 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/ProtocolTableVersionTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableVersionTest.java
new file mode 100644
index 000000000..8b27df2ef
--- /dev/null
+++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableVersionTest.java
@@ -0,0 +1,72 @@
+package io.opensharing.protocol;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import io.opensharing.TestServer.Response;
+import io.opensharing.http.ErrorCodes;
+import java.net.URLEncoder;
+import java.nio.charset.StandardCharsets;
+import org.junit.jupiter.api.Test;
+
+/** Query Table Version through the local catalog and Kernel, on local Delta logs. */
+class ProtocolTableVersionTest extends LocalDeltaTableSupport {
+
+ @Test
+ void queriesAGrantedTableVersion() {
+ createShare("table-version");
+ addObject("table-version", "TABLE", "main.sales.orders", "sales.orders");
+ addObject("table-version", "SCHEMA", "main.hr", "hr");
+ createShare("hidden-version");
+ addObject("hidden-version", "TABLE", "main.sales.customers", "sales.customers");
+ String bearer = createAndActivateRecipient("table-version-partner");
+ grant("table-version", "table-version-partner");
+
+ // The latest version, as a header with no body; lookups are case-insensitive.
+ String orders = PROTOCOL + "/shares/TABLE-VERSION/schemas/SALES/tables/ORDERS/version";
+ Response latest = version(orders, bearer, null);
+ assertEquals(200, latest.status());
+ assertEquals("2", latest.headers().get("Delta-Table-Version"));
+ assertTrue(latest.body().isMissingNode());
+
+ // A table under a shared schema: the earliest version committed at or after the timestamp.
+ String salaries = PROTOCOL + "/shares/table-version/schemas/hr/tables/SALARIES/version";
+ assertVersion("0", version(salaries, bearer, "2023-12-31T00:00:00Z"));
+ assertVersion("1", version(salaries, bearer, "2024-01-01T01:00:00Z"));
+ assertVersion("2", version(salaries, bearer, "2024-01-01T01:30:00Z"));
+ assertVersion("3", version(salaries, bearer, null));
+
+ Response afterLatest = version(salaries, bearer, "2024-01-01T03:00:01Z");
+ assertEquals(400, afterLatest.status());
+ assertEquals(ErrorCodes.INVALID_PARAMETER_VALUE, afterLatest.errorCode());
+ Response invalid = version(orders, bearer, "not-a-timestamp");
+ assertEquals(400, invalid.status());
+ assertEquals(ErrorCodes.INVALID_PARAMETER_VALUE, invalid.errorCode());
+
+ // Missing tables and ungranted shares are both reported as missing.
+ Response missing =
+ version(PROTOCOL + "/shares/table-version/schemas/hr/tables/missing/version", bearer, null);
+ assertEquals(404, missing.status());
+ assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, missing.errorCode());
+ Response hidden =
+ version(
+ PROTOCOL + "/shares/hidden-version/schemas/sales/tables/customers/version",
+ bearer,
+ null);
+ assertEquals(404, hidden.status());
+ assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, hidden.errorCode());
+ }
+
+ private static Response version(String path, String bearer, String startingTimestamp) {
+ String query =
+ startingTimestamp == null
+ ? ""
+ : "?startingTimestamp=" + URLEncoder.encode(startingTimestamp, StandardCharsets.UTF_8);
+ return server.get(path + query, bearer);
+ }
+
+ private static void assertVersion(String expected, Response response) {
+ assertEquals(200, response.status());
+ assertEquals(expected, response.headers().get("Delta-Table-Version"));
+ }
+}
diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTablesTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTablesTest.java
new file mode 100644
index 000000000..62bf75029
--- /dev/null
+++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTablesTest.java
@@ -0,0 +1,157 @@
+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.fasterxml.jackson.databind.JsonNode;
+import com.linecorp.armeria.common.MediaType;
+import io.opensharing.TestServer.Response;
+import io.opensharing.http.ErrorCodes;
+import java.nio.charset.StandardCharsets;
+import org.junit.jupiter.api.Test;
+
+class ProtocolTablesTest extends ProtocolApiSupport {
+
+ @Test
+ void listsGrantedTablesInASchema() {
+ createShare("table-share");
+ addObject("table-share", "TABLE", "main.sales.orders", "sales.orders");
+ addObject("table-share", "TABLE", "main.sales.customers", "sales.customers");
+ addObject("table-share", "TABLE", "main.ops.people", "ops.people");
+ String bearer = createAndActivateRecipient("table-partner");
+ grant("table-share", "table-partner");
+
+ // Share and schema lookups are case-insensitive.
+ Response list = server.get(PROTOCOL + "/shares/TABLE-SHARE/schemas/SALES/tables", bearer);
+ assertEquals(200, list.status());
+ assertJsonUtf8(list);
+ JsonNode items = list.body().path("items");
+ assertEquals(2, items.size());
+ JsonNode customers = items.path(0);
+ assertEquals("customers", customers.path("name").asText());
+ assertEquals("sales", customers.path("schema").asText());
+ assertEquals("table-share", customers.path("share").asText());
+ assertTrue(customers.has("shareId"));
+ assertEquals("main.sales.customers", customers.path("id").asText());
+ assertEquals("s3://test/main.sales.customers/", customers.path("location").asText());
+ assertEquals("url", customers.path("accessModes").path(0).asText());
+ assertEquals("dir", customers.path("accessModes").path(1).asText());
+ assertEquals("orders", items.path(1).path("name").asText());
+ assertTrue(list.body().path("nextPageToken").isMissingNode());
+
+ Response missing = server.get(PROTOCOL + "/shares/table-share/schemas/missing/tables", bearer);
+ assertEquals(404, missing.status());
+ assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, missing.errorCode());
+ }
+
+ @Test
+ void listsCatalogChildrenOfASharedSchema() {
+ createShare("schema-tables");
+ addObject("schema-tables", "SCHEMA", "main.hr", "hr");
+ createShare("hidden-tables");
+ addObject("hidden-tables", "TABLE", "main.sales.orders", "sales.orders");
+ String bearer = createAndActivateRecipient("schema-tables-partner");
+ grant("schema-tables", "schema-tables-partner");
+
+ Response list = server.get(PROTOCOL + "/shares/schema-tables/schemas/hr/tables", bearer);
+ assertEquals(200, list.status());
+ JsonNode items = list.body().path("items");
+ assertEquals(2, items.size());
+ assertEquals("employees", items.path(0).path("name").asText());
+ assertEquals("s3://test/main.hr.employees/", items.path(0).path("location").asText());
+ assertEquals("main.hr.employees", items.path(0).path("id").asText());
+ JsonNode salaries = items.path(1);
+ assertEquals("salaries", salaries.path("name").asText());
+ assertEquals("s3://test/main.hr.salaries/", salaries.path("location").asText());
+ assertEquals("main.hr.salaries", salaries.path("id").asText());
+ assertEquals("url", salaries.path("accessModes").path(0).asText());
+ assertEquals("dir", salaries.path("accessModes").path(1).asText());
+
+ // An ungranted share is reported as missing.
+ Response hidden = server.get(PROTOCOL + "/shares/hidden-tables/schemas/sales/tables", bearer);
+ assertEquals(404, hidden.status());
+ assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, hidden.errorCode());
+ }
+
+ @Test
+ void paginatesTablesInASchema() {
+ createShare("paged-tables");
+ addObject("paged-tables", "TABLE", "main.sales.table1", "sales.a");
+ addObject("paged-tables", "TABLE", "main.sales.orders", "sales.b");
+ String bearer = createAndActivateRecipient("table-paged");
+ grant("paged-tables", "table-paged");
+
+ String tables = PROTOCOL + "/shares/paged-tables/schemas/sales/tables";
+ Response first = server.get(page(tables, 1, null), bearer);
+ assertEquals(200, first.status());
+ assertEquals(1, first.body().path("items").size());
+ assertEquals("a", first.body().path("items").path(0).path("name").asText());
+ String pageToken = first.body().path("nextPageToken").asText();
+ assertFalse(pageToken.isEmpty());
+ Response second = server.get(page(tables, 1, pageToken), bearer);
+ assertEquals(200, second.status());
+ assertEquals(1, second.body().path("items").size());
+ assertEquals("b", second.body().path("items").path(0).path("name").asText());
+ assertTrue(second.body().path("nextPageToken").isMissingNode());
+
+ // A page size of 0 returns no tables, but a token to resume from.
+ Response zero = server.get(page(tables, 0, null), bearer);
+ assertEquals(0, zero.body().path("items").size());
+ assertFalse(zero.body().path("nextPageToken").isMissingNode());
+ }
+
+ @Test
+ void listsAllTablesInAGrantedShare() {
+ createShare("all-tables");
+ addObject("all-tables", "TABLE", "main.sales.orders", "sales.orders");
+ addObject("all-tables", "SCHEMA", "main.hr", "hr");
+ createShare("empty-tables");
+ String bearer = createAndActivateRecipient("all-tables-partner");
+ grant("all-tables", "all-tables-partner");
+ grant("empty-tables", "all-tables-partner");
+
+ // Catalog children of a shared schema and included tables, by schema and then name.
+ String all = PROTOCOL + "/shares/all-tables/all-tables";
+ Response list = server.get(PROTOCOL + "/shares/ALL-TABLES/all-tables", bearer);
+ assertEquals(200, list.status());
+ assertJsonUtf8(list);
+ JsonNode items = list.body().path("items");
+ assertEquals(3, items.size());
+ assertEquals("hr", items.path(0).path("schema").asText());
+ assertEquals("employees", items.path(0).path("name").asText());
+ assertEquals("hr", items.path(1).path("schema").asText());
+ assertEquals("salaries", items.path(1).path("name").asText());
+ assertEquals("sales", items.path(2).path("schema").asText());
+ assertEquals("orders", items.path(2).path("name").asText());
+ assertTrue(list.body().path("nextPageToken").isMissingNode());
+
+ Response first = server.get(page(all, 1, null), bearer);
+ assertEquals(200, first.status());
+ assertEquals(1, first.body().path("items").size());
+ assertEquals("employees", first.body().path("items").path(0).path("name").asText());
+ String pageToken = first.body().path("nextPageToken").asText();
+ assertFalse(pageToken.isEmpty());
+ Response second = server.get(page(all, 2, pageToken), bearer);
+ assertEquals(200, second.status());
+ JsonNode rest = second.body().path("items");
+ assertEquals(2, rest.size());
+ assertEquals("salaries", rest.path(0).path("name").asText());
+ assertEquals("orders", rest.path(1).path("name").asText());
+ assertTrue(second.body().path("nextPageToken").isMissingNode());
+
+ Response empty = server.get(PROTOCOL + "/shares/empty-tables/all-tables", bearer);
+ assertEquals(200, empty.status());
+ assertEquals(0, empty.body().path("items").size());
+ assertTrue(empty.body().path("nextPageToken").isMissingNode());
+ Response missing = server.get(PROTOCOL + "/shares/missing/all-tables", bearer);
+ assertEquals(404, missing.status());
+ assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, missing.errorCode());
+ }
+
+ private static void assertJsonUtf8(Response response) {
+ MediaType contentType = response.headers().contentType();
+ assertTrue(contentType.is(MediaType.JSON), contentType.toString());
+ assertEquals(StandardCharsets.UTF_8, contentType.charset());
+ }
+}
diff --git a/server/opensharing-server/src/test/resources/local-catalog-cloud.yml b/server/opensharing-server/src/test/resources/local-catalog-cloud.yml
new file mode 100644
index 000000000..3187833e4
--- /dev/null
+++ b/server/opensharing-server/src/test/resources/local-catalog-cloud.yml
@@ -0,0 +1,43 @@
+# Local catalog used by cloud-backed table-version integration tests.
+# ENV mode chooses AWS, Azure, or GCP credentials from each asset's URI scheme.
+credentials:
+ mode: ENV
+ ttlSeconds: 900
+
+principals:
+ - bearerToken: alice-token
+ userId: catalog-alice-id
+ userName: alice
+ privileges: [CREATE_SHARE, CREATE_RECIPIENT, USE_SHARE, USE_RECIPIENT]
+
+assets:
+ - identifier: main.sales
+ type: SCHEMA
+
+ - identifier: main.sales.table1
+ type: TABLE
+ attributes:
+ subtype: MANAGED
+ storageLocation: s3://delta-exchange-test/delta-exchange-test/table1/
+ format: delta
+
+ - identifier: main.sales.startingtimestamp
+ type: TABLE
+ attributes:
+ subtype: MANAGED
+ storageLocation: s3://delta-exchange-test/delta-exchange-test/table_startingtimestamp_test/
+ format: delta
+
+ - identifier: main.sales.azure
+ type: TABLE
+ attributes:
+ subtype: MANAGED
+ storageLocation: abfss://delta-sharing-test-container@deltasharingtest.dfs.core.windows.net/delta-sharing-test/table1
+ format: delta
+
+ - identifier: main.sales.gcs
+ type: TABLE
+ attributes:
+ subtype: MANAGED
+ storageLocation: gs://delta-sharing-dev/delta-sharing-test/table1
+ format: delta
diff --git a/server/pom.xml b/server/pom.xml
index b6f779251..6021f879a 100644
--- a/server/pom.xml
+++ b/server/pom.xml
@@ -78,6 +78,7 @@
3.4.3
2.35.4
hadoop3-2.2.30
+ 3.17.0