protocol: list tables in granted schemas

server/opensharing-server-core/src/main/java/io/opensharing/asset/SharedDataObjectStore.java

@@ -87,6 +87,20 @@ public List<SharedDataObjectEntity> list(ShareEntity share) {
                 .getResultList());
   }
 
+  /** The objects {@code share} shares under {@code schema}, ordered by table. */
+  public List<SharedDataObjectEntity> 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<String> listSchemas(ShareEntity share, int offset, int limit) {
     return tx.inTransaction(

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.
+ *
+ * <p>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<TableResponse> list(
+      RecipientPrincipal recipient,
+      String share,
+      String schema,
+      Integer maxResults,
+      String pageToken) {
+    ShareEntity granted = shares.requireGranted(recipient, share);
+    List<SharedDataObjectEntity> 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<TableResponse> 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<ResolvedAsset> resolved) {}
+
+  private List<Listed> listed(ShareEntity share, List<SharedDataObjectEntity> included) {
+    List<Listed> 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<TableResponse> page(
+      ShareEntity share, List<Listed> 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<ResolvedAsset> tablesIn(ShareEntity share, SharedDataObjectEntity schema) {
+    Asset parent = new Asset(AssetType.SCHEMA, schema.getName());
+    List<ResolvedAsset> 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);
+  }
+}

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<String> 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;
+  }
+}

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:
+ *
+ * <pre>{@code
+ * {"name": "<table name>", "schema": "<schema name>", "share": "<share name>",
+ *  "shareId": "<share id>", "id": "<catalog table id>", "location": "<storage location>",
+ *  "auxiliaryLocations": ["<storage location>", ...], "accessModes": ["url", "dir"]}
+ * }</pre>
+ */
+public record TableResponse(
+    String name,
+    String schema,
+    String share,
+    String shareId,
+    String id,
+    String location,
+    List<String> auxiliaryLocations,
+    List<String> accessModes) {}

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<SchemaResponse> 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<TableResponse> 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<TableResponse> listAllTables(
+      ServiceRequestContext ctx,
+      @Param String share,
+      @Param @Nullable Integer maxResults,
+      @Param @Nullable String pageToken) {
+    return tables.listAll(RecipientTokenDecorator.recipient(ctx), share, maxResults, pageToken);
+  }
 }

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)

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;

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: