Delta Sharing protocol: query a shared Delta table version

.github/workflows/ci.yml

@@ -44,15 +44,22 @@ jobs:
           cache-dependency-path: server/**/pom.xml
 
       # Repo secrets are available on same-repo PRs and pushes. Fork PRs get empty values, and
-      # the AWS integration test is skipped unless both keys are non-empty.
+      # each cloud integration test is skipped unless its credentials are non-empty.
       - name: Run tests
         if: steps.server.outputs.present == 'true'
         working-directory: server
         env:
           AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
           AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
           AWS_REGION: us-west-2
-        run: mvn -B -ntp verify
+          AZURE_TEST_ACCOUNT_KEY: ${{ secrets.AZURE_TEST_ACCOUNT_KEY }}
+          GOOGLE_SERVICE_ACCOUNT_KEY: ${{ secrets.GOOGLE_SERVICE_ACCOUNT_KEY }}
+        run: |
+          if [ -n "$GOOGLE_SERVICE_ACCOUNT_KEY" ]; then
+            export GOOGLE_APPLICATION_CREDENTIALS="$RUNNER_TEMP/google-service-account-key.json"
+            printf '%s' "$GOOGLE_SERVICE_ACCOUNT_KEY" > "$GOOGLE_APPLICATION_CREDENTIALS"
+          fi
+          mvn -B -ntp verify
 
       - name: Skip Maven tests
         if: steps.server.outputs.present != 'true'

server/opensharing-server-core/pom.xml

@@ -65,6 +65,13 @@
         </exclusion>
       </exclusions>
     </dependency>
+    <!-- hadoop-aws needs unshaded commons-lang3; hadoop-client-runtime only has a shaded copy. -->
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+      <version>${commons-lang3.version}</version>
+      <scope>runtime</scope>
+    </dependency>
     <dependency>
       <groupId>software.amazon.awssdk</groupId>
       <artifactId>s3</artifactId>

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

@@ -87,6 +87,35 @@ 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(
+        true,
+        em ->
+            em.createQuery(
+                    "select distinct o.sharedAsSchema from SharedDataObjectEntity o"
+                        + " where o.share.id = :shareId order by o.sharedAsSchema",
+                    String.class)
+                .setParameter("shareId", share.getId())
+                .setFirstResult(offset)
+                .setMaxResults(limit)
+                .getResultList());
+  }
+
   /** Removes the object shared under the alias. Fails unless it exists with the given type. */
   public void removeByAlias(
       EntityManager em,

server/opensharing-server-core/src/main/java/io/opensharing/asset/schema/ProtocolSchemas.java

@@ -0,0 +1,34 @@
+package io.opensharing.asset.schema;
+
+import io.opensharing.asset.SharedDataObjectStore;
+import io.opensharing.asset.share.ProtocolShares;
+import io.opensharing.http.ListResponse;
+import io.opensharing.http.Pagination;
+import io.opensharing.recipient.RecipientPrincipal;
+import io.opensharing.share.ShareEntity;
+
+/**
+ * The schemas a recipient sees through the sharing protocol: the distinct schema aliases of the
+ * objects in a share it holds SELECT on.
+ */
+public class ProtocolSchemas {
+
+  private final ProtocolShares shares;
+  private final SharedDataObjectStore objects;
+
+  public ProtocolSchemas(ProtocolShares shares, SharedDataObjectStore objects) {
+    this.shares = shares;
+    this.objects = objects;
+  }
+
+  /** Lists one page of the share's schemas by name. */
+  public ListResponse<SchemaResponse> list(
+      RecipientPrincipal recipient, String share, Integer maxResults, String pageToken) {
+    ShareEntity granted = shares.requireGranted(recipient, share);
+    return Pagination.protocolPage(
+        maxResults,
+        pageToken,
+        (offset, limit) -> objects.listSchemas(granted, offset, limit),
+        name -> new SchemaResponse(name, granted.getName()));
+  }
+}

server/opensharing-server-core/src/main/java/io/opensharing/asset/schema/SchemaResponse.java

@@ -0,0 +1,10 @@
+package io.opensharing.asset.schema;
+
+/**
+ * Protocol schema listed under a share:
+ *
+ * <pre>{@code
+ * {"name": "<schema name>", "share": "<share name>"}
+ * }</pre>
+ */
+public record SchemaResponse(String name, String share) {}

server/opensharing-server-core/src/main/java/io/opensharing/asset/share/ProtocolShares.java

@@ -4,6 +4,7 @@
 import io.opensharing.http.ListResponse;
 import io.opensharing.http.Pagination;
 import io.opensharing.recipient.RecipientPrincipal;
+import io.opensharing.share.ShareEntity;
 import io.opensharing.share.ShareResponse;
 import io.opensharing.share.SharePermissionStore;
 import io.opensharing.share.ShareStore;
@@ -34,10 +35,16 @@ public ListResponse<ShareResponse> list(
 
   /** Gets one of the recipient's shares by name in any case. */
   public GetShareResponse get(RecipientPrincipal recipient, String share) {
+    return new GetShareResponse(ShareResponse.from(requireGranted(recipient, share)));
+  }
+
+  /**
+   * Looks up a share by name in any case. Fails with not-found unless the recipient holds SELECT.
+   */
+  public ShareEntity requireGranted(RecipientPrincipal recipient, String share) {
     return shares
         .find(share)
         .filter(candidate -> permissions.hasSelect(candidate, recipient.recipientId()))
-        .map(granted -> new GetShareResponse(ShareResponse.from(granted)))
         .orElseThrow(() -> ApiException.notFound("share '" + share + "' does not exist"));
   }
 }

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

@@ -0,0 +1,12 @@
+package io.opensharing.asset.table;
+
+/**
+ * Protocol {@code GET /shares/{share}/schemas/{schema}/tables/{table}}:
+ *
+ * <pre>{@code
+ * {"name": "<table name>", "schema": "<schema name>", "share": "<share name>",
+ *  "shareId": "<share id>", "id": "<catalog table id>", "format": "<table format>"}
+ * }</pre>
+ */
+public record GetTableResponse(
+    String name, String schema, String share, String shareId, String id, String format) {}

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.
+ *
+ * <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 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<TableResponse> 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<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);
+  }
+
+  /** 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<SharedDataObjectEntity> inSchema(ShareEntity share, String schema) {
+    List<SharedDataObjectEntity> 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<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 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);
+  }
+}