diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/SharedDataObjectStore.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/SharedDataObjectStore.java index 434b1d46..d5551199 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/asset/SharedDataObjectStore.java +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/SharedDataObjectStore.java @@ -87,6 +87,20 @@ public List list(ShareEntity share) { .getResultList()); } + /** The objects {@code share} shares under {@code schema}, ordered by table. */ + public List listInSchema(ShareEntity share, String schema) { + return tx.inTransaction( + true, + em -> + em.createQuery( + "select o from SharedDataObjectEntity o where o.share.id = :shareId" + + " and o.sharedAsSchema = :schema order by o.sharedAsTable", + SharedDataObjectEntity.class) + .setParameter("shareId", share.getId()) + .setParameter("schema", schema) + .getResultList()); + } + /** One page of the schemas recipients see in {@code share}, ordered by name. */ public List listSchemas(ShareEntity share, int offset, int limit) { return tx.inTransaction( 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 00000000..2afe27f1 --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java @@ -0,0 +1,150 @@ +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.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.http.ApiException; +import io.opensharing.http.ListResponse; +import io.opensharing.http.Pagination; +import io.opensharing.recipient.RecipientPrincipal; +import io.opensharing.share.ShareEntity; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.function.Supplier; + +/** + * The tables a recipient sees 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 ProtocolShares shares; + private final SharedDataObjectStore objects; + + public ProtocolTables( + CatalogConnector catalog, ProtocolShares shares, SharedDataObjectStore objects) { + this.catalog = catalog; + 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); + List included = + objects.listInSchema(granted, ObjectNames.normalize(schema)); + if (included.isEmpty()) { + throw ApiException.notFound( + "schema '" + schema + "' does not exist in share '" + granted.getName() + "'"); + } + return page(granted, listed(granted, included), 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); + } + + /** + * 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 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 00000000..1742baae --- /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 00000000..0b9b0434 --- /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/protocol/OpenSharingProtocolService.java b/server/opensharing-server-core/src/main/java/io/opensharing/protocol/OpenSharingProtocolService.java
index 37c9be71..cc62b6ec 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
@@ -10,6 +10,8 @@
 import io.opensharing.asset.schema.SchemaResponse;
 import io.opensharing.asset.share.GetShareResponse;
 import io.opensharing.asset.share.ProtocolShares;
+import io.opensharing.asset.table.ProtocolTables;
+import io.opensharing.asset.table.TableResponse;
 import io.opensharing.http.ListResponse;
 import io.opensharing.share.ShareResponse;
 
@@ -23,10 +25,13 @@ public final class OpenSharingProtocolService {
 
   private final ProtocolShares shares;
   private final ProtocolSchemas schemas;
+  private final ProtocolTables tables;
 
-  public OpenSharingProtocolService(ProtocolShares shares, ProtocolSchemas schemas) {
+  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. */
@@ -53,4 +58,26 @@ public ListResponse listSchemas(
       @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);
+  }
 }
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 28244db7..3a9ea008 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
@@ -41,7 +41,9 @@ public static void mount(
         .decorator(errors)
         .build(
             new OpenSharingProtocolService(
-                openSharing.protocolShares(), openSharing.protocolSchemas()));
+                openSharing.protocolShares(),
+                openSharing.protocolSchemas(),
+                openSharing.protocolTables()));
     builder
         .annotatedService()
         .pathPrefix(activationBasePath)
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 9cf8a5a1..058a8674 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
@@ -5,6 +5,7 @@
 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.catalog.CatalogConnector;
 import io.opensharing.recipient.RecipientEntity;
 import io.opensharing.recipient.RecipientService;
@@ -29,6 +30,7 @@ public final class OpenSharing {
   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;
@@ -41,6 +43,7 @@ private OpenSharing(CatalogConnector catalog, Transactions tx) {
     this.recipients = new RecipientService(catalog, recipientStore);
     this.protocolShares = new ProtocolShares(shareStore, permissionStore);
     this.protocolSchemas = new ProtocolSchemas(protocolShares, objectStore);
+    this.protocolTables = new ProtocolTables(catalog, protocolShares, objectStore);
   }
 
   public static Builder builder() {
@@ -79,6 +82,11 @@ 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/resources/local-catalog.yml b/server/opensharing-server/src/main/resources/local-catalog.yml
index 4708decc..2274d6de 100644
--- a/server/opensharing-server/src/main/resources/local-catalog.yml
+++ b/server/opensharing-server/src/main/resources/local-catalog.yml
@@ -47,9 +47,30 @@ assets:
     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:
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 00000000..62bf7502
--- /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());
+  }
+}