diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/DeltaSharingCapabilities.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/DeltaSharingCapabilities.java index bba7bcd8..282007d5 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/DeltaSharingCapabilities.java +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/DeltaSharingCapabilities.java @@ -1,11 +1,16 @@ package io.opensharing.asset.table; +import io.delta.kernel.internal.actions.Protocol; +import io.delta.kernel.internal.tablefeatures.TableFeature; +import io.delta.kernel.internal.tablefeatures.TableFeatures; import io.opensharing.http.ApiException; import java.util.EnumSet; import java.util.LinkedHashMap; +import java.util.LinkedHashSet; import java.util.Locale; import java.util.Map; import java.util.Set; +import java.util.stream.Collectors; /** Parses {@code delta-sharing-capabilities} and chooses parquet vs delta response actions. */ public final class DeltaSharingCapabilities { @@ -20,6 +25,12 @@ public enum ResponseFormat { private static final Set KEYS = Set.of("responseformat", "readerfeatures", "includeendstreamaction"); + private static final Set READER_FEATURES = + TableFeatures.TABLE_FEATURES.stream() + .filter(TableFeature::isReaderWriterFeature) + .map(feature -> feature.featureName().toLowerCase(Locale.ROOT)) + .collect(Collectors.toUnmodifiableSet()); + private DeltaSharingCapabilities() {} /** @@ -39,6 +50,32 @@ public static String responded(String requestHeader) { : "responseformat=parquet"; } + /** + * Query Table checks reader features only for delta responses on snapshots that declare them. + * Parquet never carries {@code readerfeatures}; a parquet query of an advanced table is rejected. + * Metadata does not enforce this so clients can inspect protocol before querying files. + */ + public static void requireReaderFeatures( + String header, ResponseFormat format, Protocol protocol) { + Set required = protocol.getReaderFeatures(); + if (required == null || required.isEmpty()) { + return; + } + if (format != ResponseFormat.DELTA) { + throw ApiException.invalidParameter( + "The table requires reader features that are only available in delta format"); + } + Set advertised = advertisedReaderFeatures(header); + for (String feature : required) { + if (!advertised.contains(feature.toLowerCase(Locale.ROOT))) { + throw ApiException.invalidParameter( + "The table requires reader feature '" + + feature + + "' which was not specified in delta-sharing-capabilities"); + } + } + } + static Map parse(String header) { if (header == null || header.isBlank()) { return Map.of(); @@ -97,4 +134,23 @@ static Set responseFormats(Map capabilities) { } return requested; } + + private static Set advertisedReaderFeatures(String header) { + String value = parse(header).get("readerfeatures"); + if (value == null) { + return Set.of(); + } + if (value.isBlank()) { + throw ApiException.invalidParameter("Unsupported readerfeatures: ''"); + } + Set advertised = new LinkedHashSet<>(); + for (String token : value.split(",")) { + String feature = token.trim(); + if (feature.isEmpty() || !READER_FEATURES.contains(feature)) { + throw ApiException.invalidParameter("Unsupported readerfeatures: '" + feature + "'"); + } + advertised.add(feature); + } + return advertised; + } } diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/FileIdHash.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/FileIdHash.java new file mode 100644 index 00000000..7293bee1 --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/FileIdHash.java @@ -0,0 +1,53 @@ +package io.opensharing.asset.table; + +import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat; +import io.opensharing.http.ApiException; +import java.nio.charset.StandardCharsets; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.HexFormat; +import java.util.Locale; +import java.util.Set; + +/** + * Parses {@code fileidhash} and hashes file paths the way the OSS server does: {@code parquet} is + * MD5, {@code delta} is SHA-256. When the header is omitted, the hash follows the responded format. + */ +public final class FileIdHash { + + public static final String HEADER = "fileidhash"; + + private static final Set VALID = Set.of("parquet", "delta"); + + private FileIdHash() {} + + /** Lowercase {@code parquet} or {@code delta}, or {@code null} when the header is absent. */ + public static String parse(String header) { + if (header == null || header.isBlank()) { + return null; + } + String normalized = header.trim().toLowerCase(Locale.ROOT); + if (!VALID.contains(normalized)) { + throw ApiException.invalidParameter( + "Unsupported fileidhash: '" + header + "'. Supported: parquet, delta"); + } + return normalized; + } + + public static String hash(String path, String fileIdHash, ResponseFormat format) { + boolean sha256 = + "delta".equals(fileIdHash) + || (fileIdHash == null && format == ResponseFormat.DELTA); + return digest(sha256 ? "SHA-256" : "MD5", path); + } + + private static String digest(String algorithm, String path) { + try { + return HexFormat.of() + .formatHex( + MessageDigest.getInstance(algorithm).digest(path.getBytes(StandardCharsets.UTF_8))); + } catch (NoSuchAlgorithmException e) { + throw new IllegalStateException(e); + } + } +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java index 7f736c3e..565e6db0 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/ProtocolTables.java @@ -6,6 +6,7 @@ import io.opensharing.asset.share.ProtocolShares; import io.opensharing.asset.table.delta.DeltaKernel; import io.opensharing.asset.table.delta.DeltaTableMetadataReader; +import io.opensharing.asset.table.delta.DeltaTableQueryReader; import io.opensharing.auth.AuthContext; import io.opensharing.auth.UserContext; import io.opensharing.catalog.Asset; @@ -46,6 +47,7 @@ public class ProtocolTables { private final CatalogConnector catalog; private final DeltaKernel kernel; private final DeltaTableMetadataReader metadataReader; + private final DeltaTableQueryReader queryReader; private final ProtocolShares shares; private final SharedDataObjectStore objects; @@ -53,11 +55,13 @@ public ProtocolTables( CatalogConnector catalog, DeltaKernel kernel, DeltaTableMetadataReader metadataReader, + DeltaTableQueryReader queryReader, ProtocolShares shares, SharedDataObjectStore objects) { this.catalog = catalog; this.kernel = kernel; this.metadataReader = metadataReader; + this.queryReader = queryReader; this.shares = shares; this.objects = objects; } @@ -165,6 +169,46 @@ public TemporaryTableCredentialsResponse temporaryCredentials( return new TemporaryTableCredentialsResponse(TemporaryCredentials.from(minted)); } + /** + * One snapshot of a table in the share that supports url access, as of the latest version, + * {@code version}, or an ISO-8601 {@code timestamp} in {@code request}, with presigned file + * URLs. Pushdown hints and startingVersion queries are not implemented. + */ + public DeltaTableQueryReader.Result query( + RecipientPrincipal recipient, + String share, + String schema, + String table, + QueryTableRequest request, + DeltaTableQueryReader.ResponseOptions options) { + ShareEntity granted = shares.requireGranted(recipient, share); + ResolvedAsset resolved = resolve(granted, schema, table); + List accessModes = TableAccessModes.forTable(resolved); + if (accessModes == null || !accessModes.contains("url")) { + throw ApiException.invalidParameter( + "table '" + schema + "." + table + "' does not support url access"); + } + QueryTableRequest body = request == null ? QueryTableRequest.EMPTY : request; + if (body.startingVersion() != null || body.endingVersion() != null) { + throw ApiException.notImplemented("startingVersion queries are not supported"); + } + if (hasPushdownHint(body)) { + throw ApiException.notImplemented("predicate and limit pushdown is not supported"); + } + return queryReader.read( + resolved, + owner(granted), + options, + body.version(), + parseTimestamp("timestamp", body.timestamp())); + } + + private static boolean hasPushdownHint(QueryTableRequest body) { + return (body.predicateHints() != null && !body.predicateHints().isEmpty()) + || (body.jsonPredicateHints() != null && !body.jsonPredicateHints().isBlank()) + || body.limitHint() != null; + } + private List inSchema(ShareEntity share, String schema) { List included = objects.listInSchema(share, ObjectNames.normalize(schema)); diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/QueryTableRequest.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/QueryTableRequest.java new file mode 100644 index 00000000..b3eaa92c --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/QueryTableRequest.java @@ -0,0 +1,22 @@ +package io.opensharing.asset.table; + +import java.util.List; + +/** + * Optional body for Query Table. Snapshot queries use version or timestamp; streaming is not + * implemented. + */ +public record QueryTableRequest( + List predicateHints, + String jsonPredicateHints, + Integer limitHint, + Long version, + String timestamp, + Long startingVersion, + Long endingVersion, + Boolean includeHistoricalProtocol, + String idempotencyKey) { + + static final QueryTableRequest EMPTY = + new QueryTableRequest(null, null, null, null, null, null, null, null, null); +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableActions.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableActions.java index dc92a103..9c168708 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableActions.java +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableActions.java @@ -7,18 +7,23 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializerProvider; import com.fasterxml.jackson.databind.annotation.JsonSerialize; +import io.delta.kernel.data.Row; import io.delta.kernel.defaults.internal.json.JsonUtils; +import io.delta.kernel.internal.actions.AddFile; +import io.delta.kernel.internal.actions.DeletionVectorDescriptor; import io.delta.kernel.internal.actions.Format; import io.delta.kernel.internal.actions.Metadata; import io.delta.kernel.internal.actions.Protocol; +import io.delta.kernel.internal.data.DelegateRow; import io.delta.kernel.internal.util.VectorUtils; import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat; import io.opensharing.catalog.ResolvedAsset; import java.io.IOException; +import java.util.HashMap; import java.util.List; import java.util.Map; -/** Protocol and metadata NDJSON actions shared by Query Table Metadata and Query Table. */ +/** Protocol, metadata, and file NDJSON actions shared by Query Table Metadata and Query Table. */ public final class TableActions { private static final ObjectMapper JSON = @@ -72,6 +77,70 @@ public static String metadata( numFiles))); } + public static String parquetFile( + String url, + String id, + Map partitionValues, + long size, + String stats, + Long version, + Long timestamp, + Long expirationTimestamp) { + return line( + new FileLine( + new ParquetFile( + url, + id, + partitionValues == null ? Map.of() : partitionValues, + size, + stats, + version, + timestamp, + expirationTimestamp))); + } + + public static String deltaFile( + String id, + String deletionVectorFileId, + Long expirationTimestamp, + Long version, + Long timestamp, + AddFile add) { + return line( + new FileLine( + new DeltaFile( + id, + deletionVectorFileId, + version, + timestamp, + expirationTimestamp, + new DeltaSingleAction(add)))); + } + + /** + * Replace the add path with a signed file URL. When {@code deletionVectorUrl} is set, rewrite the + * on-disk deletion vector as a path-type descriptor so clients fetch the signed object. + */ + public static AddFile withPath(AddFile add, String path, String deletionVectorUrl) { + Row row = add.toRow(); + Map overrides = new HashMap<>(); + overrides.put(row.getSchema().indexOf("path"), path); + if (deletionVectorUrl != null) { + int dvOrdinal = row.getSchema().indexOf("deletionVector"); + Row dv = row.getStruct(dvOrdinal); + overrides.put( + dvOrdinal, + new DelegateRow( + dv, + Map.of( + dv.getSchema().indexOf("storageType"), + (Object) DeletionVectorDescriptor.PATH_DV_MARKER, + dv.getSchema().indexOf("pathOrInlineDv"), + deletionVectorUrl))); + } + return new AddFile(new DelegateRow(row, overrides)); + } + private static String line(Object value) { try { return JSON.writeValueAsString(value) + "\n"; @@ -99,11 +168,16 @@ public static final class KernelActionSerializer extends JsonSerializer @Override public void serialize(Object value, JsonGenerator gen, SerializerProvider serializers) throws IOException { - gen.writeRawValue( - JsonUtils.rowToJson( - value instanceof Protocol protocol - ? protocol.toRow() - : ((Metadata) value).toRow())); + Row row = + switch (value) { + case Protocol protocol -> protocol.toRow(); + case Metadata metadata -> metadata.toRow(); + case AddFile add -> add.toRow(); + default -> + throw new IllegalArgumentException( + "unsupported Kernel action: " + value.getClass().getName()); + }; + gen.writeRawValue(JsonUtils.rowToJson(row)); } } @@ -152,4 +226,33 @@ public record DeltaMetadata( implements MetadataBody {} public record FormatBody(String provider) {} + + public record FileLine(FileBody file) {} + + public sealed interface FileBody permits ParquetFile, DeltaFile {} + + @JsonInclude(JsonInclude.Include.NON_NULL) + public record ParquetFile( + String url, + String id, + Map partitionValues, + long size, + String stats, + Long version, + Long timestamp, + Long expirationTimestamp) + implements FileBody {} + + @JsonInclude(JsonInclude.Include.NON_NULL) + public record DeltaFile( + String id, + String deletionVectorFileId, + Long version, + Long timestamp, + Long expirationTimestamp, + DeltaSingleAction deltaSingleAction) + implements FileBody {} + + public record DeltaSingleAction( + @JsonSerialize(using = KernelActionSerializer.class) AddFile add) {} } diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java index 06df9fa5..5e762ea6 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaKernel.java @@ -54,6 +54,11 @@ static long versionAtOrAfter(TableImpl table, Engine engine, Instant startingTim /** Opens the table's root with read credentials the catalog vends to {@code auth}. */ public Session open(ResolvedAsset table, AuthContext auth) { + return open(table, auth, CREDENTIAL_TTL); + } + + /** Like {@link #open(ResolvedAsset, AuthContext)}, with credentials that last {@code ttl}. */ + public Session open(ResolvedAsset table, AuthContext auth, Duration ttl) { if (!(table.additionalProperties() instanceof TableProperties properties) || properties.dataSourceFormat() != DataSourceFormat.DELTA) { throw ApiException.invalidParameter("table '" + table.fullName() + "' is not a Delta table"); @@ -70,7 +75,7 @@ public Session open(ResolvedAsset table, AuthContext auth) { table.catalogAssetId(), location, StorageOperation.READ, - CREDENTIAL_TTL), + ttl), auth); StorageCredentials rootCredentials = credentials.stream() @@ -83,8 +88,9 @@ public Session open(ResolvedAsset table, AuthContext auth) { Engine engine = DefaultEngine.create(HadoopStorageConfiguration.from(rootCredentials, location)); String path = HadoopStorageConfiguration.kernelPath(location); - return new Session(engine, Table.forPath(engine, path), location); + return new Session(engine, Table.forPath(engine, path), location, rootCredentials); } - public record Session(Engine engine, Table table, String location) {} + public record Session( + Engine engine, Table table, String location, StorageCredentials credentials) {} } diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaTableQueryReader.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaTableQueryReader.java new file mode 100644 index 00000000..872a5c96 --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/DeltaTableQueryReader.java @@ -0,0 +1,238 @@ +package io.opensharing.asset.table.delta; + +import io.delta.kernel.Scan; +import io.delta.kernel.Snapshot; +import io.delta.kernel.data.FilteredColumnarBatch; +import io.delta.kernel.data.Row; +import io.delta.kernel.internal.InternalScanFileUtils; +import io.delta.kernel.internal.SnapshotImpl; +import io.delta.kernel.internal.actions.AddFile; +import io.delta.kernel.internal.actions.DeletionVectorDescriptor; +import io.delta.kernel.internal.actions.Metadata; +import io.delta.kernel.internal.actions.Protocol; +import io.delta.kernel.internal.checksum.CRCInfo; +import io.delta.kernel.utils.CloseableIterator; +import io.delta.kernel.utils.FileStatus; +import io.opensharing.asset.table.DeltaSharingCapabilities; +import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat; +import io.opensharing.asset.table.FileIdHash; +import io.opensharing.asset.table.TableActions; +import io.opensharing.asset.table.signer.SignedUrl; +import io.opensharing.asset.table.signer.UrlSigners; +import io.opensharing.auth.AuthContext; +import io.opensharing.catalog.ResolvedAsset; +import io.opensharing.http.ApiException; +import java.io.IOException; +import java.io.UncheckedIOException; +import java.time.Duration; +import java.time.Instant; +import java.util.Map; +import java.util.Optional; + +/** + * Reads Query Table snapshot files from a Delta log. The parquet or delta response is NDJSON: + * protocol, metaData, then one file action per data file. + * + *

Every file URL is presigned for an hour, or until the catalog's credentials expire if + * sooner, so the client reads the data files directly from storage without the provider's + * credentials. + */ +public class DeltaTableQueryReader { + + private static final Duration URL_TTL = Duration.ofHours(1); + + private final DeltaKernel kernel; + private final UrlSigners signers; + + public DeltaTableQueryReader(DeltaKernel kernel, UrlSigners signers) { + this.kernel = kernel; + this.signers = signers; + } + + /** + * Reads one snapshot of {@code table} as Query Table NDJSON. + * + * @param auth the share owner the catalog vends storage credentials for + * @param version the table version to read; null with a null {@code timestamp} reads the latest + * @param timestamp read the latest version committed at or before this time + */ + public Result read( + ResolvedAsset table, + AuthContext auth, + ResponseOptions options, + Long version, + Instant timestamp) { + if (version != null && timestamp != null) { + throw ApiException.invalidParameter("version and timestamp are mutually exclusive"); + } + // A time-travel query reports the snapshot version in metaData and on every file. + boolean historical = version != null || timestamp != null; + + DeltaKernel.Session session = kernel.open(table, auth, URL_TTL); + Snapshot snapshot = DeltaSnapshots.open(session, version, timestamp); + SnapshotImpl impl = (SnapshotImpl) snapshot; + ResponseWriter out = new ResponseWriter(session, table, options); + // Fails before any file is read when the client cannot read the table in this format. + out.requireReaderFeatures(impl.getProtocol()); + Long fileVersion = historical ? snapshot.getVersion() : null; + Long fileTimestamp = historical ? snapshot.getTimestamp(session.engine()) : null; + out.protocol(impl.getProtocol(), null); + out.metadata(impl.getMetadata(), fileVersion, crcFor(impl, snapshot.getVersion())); + + // Scan files are the add actions that are live in the snapshot after log replay. + Scan scan = snapshot.getScanBuilder().build(); + try (CloseableIterator batches = scan.getScanFiles(session.engine())) { + while (batches.hasNext()) { + // getRows skips rows the selection vector drops, such as files removed by later commits. + try (CloseableIterator rows = batches.next().getRows()) { + while (rows.hasNext()) { + out.snapshotFile(rows.next(), fileVersion, fileTimestamp); + } + } + } + } catch (IOException e) { + throw new UncheckedIOException(e); + } + + return new Result(snapshot.getVersion(), out.ndjson()); + } + + /** + * The snapshot's checksum file, if it was written for exactly {@code version}. Its size and + * file count go into metaData; without it they are omitted rather than computed by a scan. + */ + private static CRCInfo crcFor(SnapshotImpl snapshot, long version) { + return snapshot + .getCurrentCrcInfo() + .filter(crc -> crc.getVersion() == version) + .orElse(null); + } + + /** {@code version} is the snapshot read, returned as the {@code Delta-Table-Version} header. */ + public record Result(long version, String ndjson) {} + + /** + * How one response is written. + * + * @param capabilities the request's {@code delta-sharing-capabilities} header; picks the format + * @param fileIdHash the parsed {@code fileidhash} header, or null to follow the format + */ + public record ResponseOptions(String capabilities, String fileIdHash) {} + + /** + * Builds one NDJSON response. File URLs are presigned with the table's storage credentials. + */ + private final class ResponseWriter { + + final ResponseFormat format; + private final DeltaKernel.Session session; + private final ResolvedAsset table; + private final String capabilities; + private final String fileIdHash; + private final StringBuilder ndjson = new StringBuilder(); + + ResponseWriter(DeltaKernel.Session session, ResolvedAsset table, ResponseOptions options) { + this.format = DeltaSharingCapabilities.choose(options.capabilities()); + this.session = session; + this.table = table; + this.capabilities = options.capabilities(); + this.fileIdHash = options.fileIdHash(); + } + + void requireReaderFeatures(Protocol protocol) { + DeltaSharingCapabilities.requireReaderFeatures(capabilities, format, protocol); + } + + /** {@code version} is null for the leading protocol and set for a later one. */ + void protocol(Protocol protocol, Long version) { + ndjson.append(TableActions.protocol(protocol, version, format)); + } + + /** {@code crc} supplies the table size and file count; null omits them. */ + void metadata(Metadata metadata, Long version, CRCInfo crc) { + ndjson.append( + TableActions.metadata( + metadata, + table, + version, + crc == null ? null : crc.getTableSizeBytes(), + crc == null ? null : crc.getNumFiles(), + format)); + } + + /** Appends the file action for a snapshot scan file. */ + void snapshotFile(Row scanFile, Long version, Long timestamp) { + AddFile add = new AddFile(scanFile.getStruct(InternalScanFileUtils.ADD_FILE_ORDINAL)); + FileStatus status = InternalScanFileUtils.getAddFileStatus(scanFile); + SignedFile signed = sign(status.getPath(), add.getPath(), add.getDeletionVector()); + String stats = add.getStatsJson().orElse(null); + if (format == ResponseFormat.DELTA) { + // The delta format nests the add action, with its paths replaced by presigned URLs. + ndjson.append( + TableActions.deltaFile( + signed.id(), + signed.deletionVectorFileId(), + signed.expirationTimestamp(), + version, + timestamp, + TableActions.withPath(add, signed.url(), signed.deletionVectorUrl()))); + } else { + // Parquet file actions always carry partitionValues, empty for an unpartitioned table. + Map partitionValues = InternalScanFileUtils.getPartitionValues(scanFile); + ndjson.append( + TableActions.parquetFile( + signed.url(), + signed.id(), + partitionValues == null ? Map.of() : partitionValues, + status.getSize(), + stats, + version, + timestamp, + signed.expirationTimestamp())); + } + } + + String ndjson() { + return ndjson.toString(); + } + + /** + * Presigns the data file and, for delta responses, its on-disk deletion vector. Inline + * deletion vectors have no object to sign. Ids hash the log path, not the URL, so they stay + * stable across queries and re-signing. + */ + private SignedFile sign( + String absolutePath, String logPath, Optional deletionVector) { + SignedUrl file = signers.sign(absolutePath, session.credentials(), URL_TTL); + long expirationTimestamp = file.expiration().toEpochMilli(); + String deletionVectorUrl = null; + String deletionVectorFileId = null; + DeletionVectorDescriptor dv = + format == ResponseFormat.DELTA + ? deletionVector.filter(DeletionVectorDescriptor::isOnDisk).orElse(null) + : null; + if (dv != null) { + String dvPath = + dv.getAbsolutePath(HadoopStorageConfiguration.kernelPath(session.location())); + SignedUrl signedDv = signers.sign(dvPath, session.credentials(), URL_TTL); + deletionVectorUrl = signedDv.url(); + deletionVectorFileId = FileIdHash.hash(dvPath, fileIdHash, format); + expirationTimestamp = Math.min(expirationTimestamp, signedDv.expiration().toEpochMilli()); + } + return new SignedFile( + file.url(), + FileIdHash.hash(logPath, fileIdHash, format), + deletionVectorUrl, + deletionVectorFileId, + expirationTimestamp); + } + } + + /** {@code expirationTimestamp} is the earlier of the file and deletion vector URL expiries. */ + private record SignedFile( + String url, + String id, + String deletionVectorUrl, + String deletionVectorFileId, + long expirationTimestamp) {} +} diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java b/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java index c739194c..138924dc 100644 --- a/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java +++ b/server/opensharing-server-core/src/main/java/io/opensharing/protocol/DeltaSharingProtocolService.java @@ -7,6 +7,7 @@ import com.linecorp.armeria.common.HttpStatus; import com.linecorp.armeria.common.MediaType; import com.linecorp.armeria.common.ResponseHeaders; +import com.linecorp.armeria.common.ResponseHeadersBuilder; import com.linecorp.armeria.common.annotation.Nullable; import com.linecorp.armeria.server.ServiceRequestContext; import com.linecorp.armeria.server.annotation.Blocking; @@ -17,10 +18,13 @@ import com.linecorp.armeria.server.annotation.Produces; import com.linecorp.armeria.server.annotation.ProducesJson; import io.opensharing.asset.table.DeltaSharingCapabilities; +import io.opensharing.asset.table.FileIdHash; import io.opensharing.asset.table.ProtocolTables; +import io.opensharing.asset.table.QueryTableRequest; import io.opensharing.asset.table.TemporaryTableCredentialsRequest; import io.opensharing.asset.table.TemporaryTableCredentialsResponse; import io.opensharing.asset.table.delta.DeltaTableMetadataReader; +import io.opensharing.asset.table.delta.DeltaTableQueryReader; import io.opensharing.http.ApiJson; /** @@ -110,4 +114,38 @@ public TemporaryTableCredentialsResponse generateTemporaryTableCredentials( return tables.temporaryCredentials( RecipientTokenDecorator.recipient(ctx), share, schema, table, options); } + + /** + * {@code POST /shares/{share}/schemas/{schema}/tables/{table}/query}: one snapshot of the table + * as NDJSON with presigned file URLs, and its version in the {@code Delta-Table-Version} header. + * The request body is optional. + */ + @Post("/shares/{share}/schemas/{schema}/tables/{table}/query") + @Produces("application/x-ndjson") + public HttpResponse queryTable( + ServiceRequestContext ctx, + @Param String share, + @Param String schema, + @Param String table, + @Header(DeltaSharingCapabilities.HEADER) @Nullable String capabilities, + @Header(FileIdHash.HEADER) @Nullable String fileIdHash, + AggregatedHttpRequest request) + throws JsonProcessingException { + String body = request.contentUtf8(); + QueryTableRequest query = + body.isBlank() ? null : ApiJson.MAPPER.readValue(body, QueryTableRequest.class); + DeltaTableQueryReader.ResponseOptions options = + new DeltaTableQueryReader.ResponseOptions(capabilities, FileIdHash.parse(fileIdHash)); + DeltaTableQueryReader.Result result = + tables.query(RecipientTokenDecorator.recipient(ctx), share, schema, table, query, options); + ResponseHeadersBuilder headers = + ResponseHeaders.builder(HttpStatus.OK) + .contentType(NDJSON) + .add(DELTA_TABLE_VERSION, Long.toString(result.version())) + .add(DeltaSharingCapabilities.HEADER, DeltaSharingCapabilities.responded(capabilities)); + if (options.fileIdHash() != null) { + headers.add(FileIdHash.HEADER, options.fileIdHash()); + } + return HttpResponse.of(headers.build(), HttpData.ofUtf8(result.ndjson())); + } } diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/runtime/OpenSharing.java b/server/opensharing-server-core/src/main/java/io/opensharing/runtime/OpenSharing.java index 83d4adbd..5298e7cd 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 @@ -8,6 +8,8 @@ import io.opensharing.asset.table.ProtocolTables; import io.opensharing.asset.table.delta.DeltaKernel; import io.opensharing.asset.table.delta.DeltaTableMetadataReader; +import io.opensharing.asset.table.delta.DeltaTableQueryReader; +import io.opensharing.asset.table.signer.UrlSigners; import io.opensharing.catalog.CatalogConnector; import io.opensharing.recipient.RecipientEntity; import io.opensharing.recipient.RecipientService; @@ -48,7 +50,12 @@ private OpenSharing(CatalogConnector catalog, Transactions tx) { DeltaKernel kernel = new DeltaKernel(catalog); this.protocolTables = new ProtocolTables( - catalog, kernel, new DeltaTableMetadataReader(kernel), protocolShares, objectStore); + catalog, + kernel, + new DeltaTableMetadataReader(kernel), + new DeltaTableQueryReader(kernel, new UrlSigners()), + protocolShares, + objectStore); } public static Builder builder() { diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/DeltaSharingCapabilitiesTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/DeltaSharingCapabilitiesTest.java index 80d62f02..813f6e68 100644 --- a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/DeltaSharingCapabilitiesTest.java +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/DeltaSharingCapabilitiesTest.java @@ -1,10 +1,13 @@ package io.opensharing.asset.table; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; +import io.delta.kernel.internal.actions.Protocol; import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat; import io.opensharing.http.ApiException; +import java.util.Set; import org.junit.jupiter.api.Test; class DeltaSharingCapabilitiesTest { @@ -47,4 +50,49 @@ void rejectsUnknownFormatsKeysAndMalformedHeader() { assertEquals( ResponseFormat.PARQUET, DeltaSharingCapabilities.choose("readerfeatures=notAFeature")); } + + @Test + void queryRequiresReaderFeaturesOnlyForDeltaAdvancedTables() { + Protocol basic = new Protocol(1, 2); + Protocol advanced = + new Protocol(3, 7, Set.of("columnMapping", "deletionVectors"), Set.of("columnMapping")); + assertDoesNotThrow( + () -> + DeltaSharingCapabilities.requireReaderFeatures( + null, ResponseFormat.PARQUET, basic)); + assertDoesNotThrow( + () -> + DeltaSharingCapabilities.requireReaderFeatures( + "responseformat=delta", ResponseFormat.DELTA, basic)); + assertDoesNotThrow( + () -> + DeltaSharingCapabilities.requireReaderFeatures( + "responseformat=delta;readerfeatures=columnmapping,deletionvectors", + ResponseFormat.DELTA, + advanced)); + assertThrows( + ApiException.class, + () -> + DeltaSharingCapabilities.requireReaderFeatures( + "responseformat=parquet", ResponseFormat.PARQUET, advanced)); + assertThrows( + ApiException.class, + () -> + DeltaSharingCapabilities.requireReaderFeatures( + "responseformat=delta", ResponseFormat.DELTA, advanced)); + assertThrows( + ApiException.class, + () -> + DeltaSharingCapabilities.requireReaderFeatures( + "responseformat=delta;readerfeatures=columnmapping", + ResponseFormat.DELTA, + advanced)); + assertThrows( + ApiException.class, + () -> + DeltaSharingCapabilities.requireReaderFeatures( + "responseformat=delta;readerfeatures=notAFeature", + ResponseFormat.DELTA, + advanced)); + } } diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/FileIdHashTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/FileIdHashTest.java new file mode 100644 index 00000000..30908267 --- /dev/null +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/FileIdHashTest.java @@ -0,0 +1,35 @@ +package io.opensharing.asset.table; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import io.opensharing.asset.table.DeltaSharingCapabilities.ResponseFormat; +import io.opensharing.http.ApiException; +import org.junit.jupiter.api.Test; + +class FileIdHashTest { + + private static final String PATH = "date=2021-04-28/part.parquet"; + + @Test + void parsesParquetAndDeltaAndRejectsUnknown() { + assertNull(FileIdHash.parse(null)); + assertNull(FileIdHash.parse("")); + assertEquals("parquet", FileIdHash.parse("PARQUET")); + assertEquals("delta", FileIdHash.parse(" delta ")); + assertThrows(ApiException.class, () -> FileIdHash.parse("md5")); + } + + @Test + void defaultsFollowResponseFormatAndExplicitHeaderWins() { + String md5 = FileIdHash.hash(PATH, "parquet", ResponseFormat.PARQUET); + String sha = FileIdHash.hash(PATH, "delta", ResponseFormat.DELTA); + assertEquals(32, md5.length()); + assertEquals(64, sha.length()); + assertEquals(md5, FileIdHash.hash(PATH, null, ResponseFormat.PARQUET)); + assertEquals(sha, FileIdHash.hash(PATH, null, ResponseFormat.DELTA)); + assertEquals(md5, FileIdHash.hash(PATH, "parquet", ResponseFormat.DELTA)); + assertEquals(sha, FileIdHash.hash(PATH, "delta", ResponseFormat.PARQUET)); + } +} diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TableActionsTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TableActionsTest.java index 8c33bd7d..c8f6473c 100644 --- a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TableActionsTest.java +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/TableActionsTest.java @@ -1,8 +1,10 @@ package io.opensharing.asset.table; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import io.delta.kernel.defaults.internal.json.JsonUtils; +import io.delta.kernel.internal.actions.AddFile; import io.delta.kernel.internal.actions.Format; import io.delta.kernel.internal.actions.Metadata; import io.delta.kernel.internal.actions.Protocol; @@ -48,6 +50,75 @@ void encodesParquetProtocolAndMetadata() { assertEquals(parquetBody(",\"version\":20"), encode(ResponseFormat.PARQUET, 20L, null, null)); } + @Test + void encodesParquetAndDeltaFiles() { + assertEquals( + """ + {"file":{"url":"https://example.invalid/part.parquet","id":"file-1",\ + "partitionValues":{"date":"2021-04-28"},"size":573,"stats":"{\\"numRecords\\":1}",\ + "expirationTimestamp":1652140800000}} + """, + TableActions.parquetFile( + "https://example.invalid/part.parquet", + "file-1", + Map.of("date", "2021-04-28"), + 573, + "{\"numRecords\":1}", + null, + null, + 1652140800000L)); + assertEquals( + """ + {"file":{"url":"https://example.invalid/part.parquet","id":"file-1",\ + "partitionValues":{},"size":1,"expirationTimestamp":1}} + """, + TableActions.parquetFile( + "https://example.invalid/part.parquet", "file-1", Map.of(), 1, null, null, null, 1L)); + AddFile add = + new AddFile( + AddFile.createAddFileRow( + AddFile.FULL_SCHEMA, + "https://example.invalid/part.parquet", + VectorUtils.stringStringMapValue(Map.of("date", "2021-04-28")), + 573, + 1, + true, + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty(), + Optional.empty())); + String addJson = JsonUtils.rowToJson(add.toRow()); + assertEquals( + """ + {"file":{"id":"file-1","expirationTimestamp":1652140800000,"deltaSingleAction":{"add":%s}}} + """ + .formatted(addJson), + TableActions.deltaFile("file-1", null, 1652140800000L, null, null, add)); + AddFile addWithDv = + new AddFile( + JsonUtils.rowFromJson( + """ + {"path":"part.parquet","partitionValues":{},"size":1,"modificationTime":1,\ + "dataChange":true,"deletionVector":{"storageType":"p","pathOrInlineDv":"dv.bin",\ + "sizeInBytes":36,"cardinality":2}} + """, + AddFile.FULL_SCHEMA)); + AddFile signed = + TableActions.withPath( + addWithDv, "https://example.invalid/part.parquet", "https://example.invalid/dv.bin"); + String signedJson = JsonUtils.rowToJson(signed.toRow()); + assertTrue(signedJson.contains("\"storageType\":\"p\"")); + assertTrue(signedJson.contains("https://example.invalid/dv.bin")); + assertEquals( + """ + {"file":{"id":"file-1","deletionVectorFileId":"dv-1","expirationTimestamp":1,\ + "deltaSingleAction":{"add":%s}}} + """ + .formatted(signedJson), + TableActions.deltaFile("file-1", "dv-1", 1L, null, null, signed)); + } + @Test void encodesDeltaProtocolAndMetadataFromKernel() { Protocol protocol = protocol(); diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/LocalDeltaTableSupport.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/LocalDeltaTableSupport.java index 85e07a37..544de127 100644 --- a/server/opensharing-server/src/test/java/io/opensharing/protocol/LocalDeltaTableSupport.java +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/LocalDeltaTableSupport.java @@ -14,9 +14,10 @@ /** * 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}. + * main.sales.customers} with 1, and {@code main.hr.salaries} with 4, each adding one data file, and + * {@code main.ops.events} with 2 commits and no data files. Commit {@code n} of each table is + * timestamped 2024-01-01T00:00:00Z plus {@code n} hours. Table {@code t} has metadata id {@code t} + * and one {@code long} column, {@code id}. */ abstract class LocalDeltaTableSupport extends ProtocolApiSupport { @@ -32,6 +33,7 @@ abstract class LocalDeltaTableSupport extends ProtocolApiSupport { private static final String ADD_ACTION = "{\"add\":{\"path\":\"part-%d.parquet\",\"partitionValues\":{},\"size\":1," + "\"modificationTime\":0,\"dataChange\":true}}\n"; + private static final String COMMIT_INFO_ACTION = "{\"commitInfo\":{}}\n"; @TempDir static Path tables; @@ -40,6 +42,7 @@ static void startLocalDeltaServer() { String orders = deltaTable("orders", 3); String customers = deltaTable("customers", 1); String salaries = deltaTable("salaries", 4); + String events = deltaTable("events", 2, false); String catalog = """ principals: @@ -67,8 +70,15 @@ static void startLocalDeltaServer() { subtype: MANAGED storageLocation: %s format: delta + - identifier: main.ops + type: SCHEMA + - identifier: main.ops.events + attributes: + subtype: MANAGED + storageLocation: %s + format: delta """ - .formatted(orders, customers, salaries); + .formatted(orders, customers, salaries, events); Properties properties = new Properties(); properties.setProperty("catalog.local.file", write("local-catalog.yml", catalog).toString()); restartServer(properties); @@ -79,14 +89,19 @@ protected static String location(String name) { return tables.resolve(name).toUri().toString(); } - /** Writes a Delta log of {@code commits} commits and returns the table's location. */ private static String deltaTable(String name, int commits) { + return deltaTable(name, commits, true); + } + + /** + * Writes a Delta log of {@code commits} commits, each adding a data file if {@code files}, and + * returns the table's location. + */ + private static String deltaTable(String name, int commits, boolean files) { 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)); + String change = files ? ADD_ACTION.formatted(version) : COMMIT_INFO_ACTION; + Path commit = write(name + "/_delta_log/%020d.json".formatted(version), actions + change); try { Files.setLastModifiedTime( commit, FileTime.from(COMMIT_TIME.plus(Duration.ofHours(version)))); diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableQueryIntegrationTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableQueryIntegrationTest.java new file mode 100644 index 00000000..382c015a --- /dev/null +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableQueryIntegrationTest.java @@ -0,0 +1,197 @@ +package io.opensharing.protocol; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.linecorp.armeria.common.HttpMethod; +import com.linecorp.armeria.common.MediaType; +import io.opensharing.TestServer.Response; +import io.opensharing.http.ErrorCodes; +import java.util.Arrays; +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 through the local catalog, Kernel, real cloud Delta logs, and presigned + * file URLs. Individual cloud cases are skipped unless their repository credentials are present. + */ +@Timeout(60) +class ProtocolTableQueryIntegrationTest 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 queriesLatestAndHistoricalFilesForTable1() { + String bearer = shareTable("table1-query", "main.sales.table1", "sales.table1"); + String endpoint = PROTOCOL + "/shares/table1-query/schemas/sales/tables/table1/query"; + + // Latest snapshot defaults to parquet NDJSON at table1 version 2. + Response latest = query(endpoint, bearer, "{}", null); + assertVersion("2", latest); + assertTrue(latest.headers().contentType().is(MediaType.parse("application/x-ndjson"))); + assertParquetQuery(latest.text()); + assertTrue(latest.text().contains("https://"), latest.text()); + assertTrue(latest.text().contains("X-Amz-Signature="), latest.text()); + + // Explicit parquet keeps protocol, schemaString, and file.url, not delta wrappers. + Response parquet = query(endpoint, bearer, "{}", "responseformat=parquet"); + assertVersion("2", parquet); + assertParquetQuery(parquet.text()); + // Dual responseformat prefers delta over parquet. + Response both = query(endpoint, bearer, "{}", "responseformat=delta,parquet"); + assertVersion("2", both); + assertDeltaQuery(both.text()); + // Delta-only wraps Kernel add actions as deltaSingleAction. + Response delta = query(endpoint, bearer, "{}", "responseformat=delta"); + assertVersion("2", delta); + assertDeltaQuery(delta.text()); + + // Historical version 0, and as of the first commit time, return that snapshot. + assertVersion("0", query(endpoint, bearer, "{\"version\":0}", null)); + assertVersion( + "0", query(endpoint, bearer, "{\"timestamp\":\"2021-05-04T17:24:06Z\"}", null)); + // As of before version 0 is INVALID_PARAMETER_VALUE. + assertInvalid(query(endpoint, bearer, "{\"timestamp\":\"1970-01-01T00:00:00Z\"}", null)); + // version and timestamp together are mutually exclusive. + assertInvalid( + query(endpoint, bearer, "{\"version\":1,\"timestamp\":\"2021-05-04T17:24:06Z\"}", null)); + } + + @Test + @EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+") + @EnabledIfEnvironmentVariable(named = "AWS_SECRET_ACCESS_KEY", matches = ".+") + void queriesFilesThroughASchemaGrant() { + createShare("schema-query"); + addObject("schema-query", "SCHEMA", "main.sales", "sales"); + String bearer = createAndActivateRecipient("schema-query-partner"); + grant("schema-query", "schema-query-partner"); + + // A SCHEMA grant can Query Table for catalog children (table1). + Response response = + query( + PROTOCOL + "/shares/schema-query/schemas/SALES/tables/TABLE1/query", + bearer, + "{}", + null); + assertVersion("2", response); + assertParquetQuery(response.text()); + } + + @Test + @EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+") + @EnabledIfEnvironmentVariable(named = "AWS_SECRET_ACCESS_KEY", matches = ".+") + void queriesDeltaFilesForDeletionVectorsTable() { + String bearer = shareTable("dv-query", "main.sales.deletionvectors", "sales.deletionvectors"); + String endpoint = PROTOCOL + "/shares/dv-query/schemas/sales/tables/deletionvectors/query"; + + // Delta format signs the deletion vector object and returns its file id for cache keys. + Response delta = + query(endpoint, bearer, "{}", "responseformat=delta;readerfeatures=deletionvectors"); + assertEquals(200, delta.status()); + assertDeltaQuery(delta.text()); + assertTrue(delta.text().contains("\"deletionVectorFileId\""), delta.text()); + assertTrue(delta.text().contains("\"deletionVector\""), delta.text()); + assertTrue(delta.text().contains("X-Amz-Signature="), delta.text()); + + // Without the reader feature, or in parquet, the table cannot be read. + assertInvalid(query(endpoint, bearer, "{}", "responseformat=delta")); + assertInvalid(query(endpoint, bearer, "{}", null)); + } + + @Test + @EnabledIfEnvironmentVariable(named = "AZURE_TEST_ACCOUNT_KEY", matches = ".+") + void queriesAzureTableFiles() { + queryCloudTableFiles("azure-query", "main.sales.azure", "sales.azure", "sig="); + } + + @Test + @EnabledIfEnvironmentVariable(named = "GOOGLE_APPLICATION_CREDENTIALS", matches = ".+") + void queriesGoogleTableFiles() { + queryCloudTableFiles("gcs-query", "main.sales.gcs", "sales.gcs", "X-Goog-Signature="); + } + + // The cloud copies of table1 have one commit, version 0, with one file in partition c2=foo bar. + private static void queryCloudTableFiles( + String share, String catalogName, String sharedAs, String signature) { + String bearer = shareTable(share, catalogName, sharedAs); + String[] alias = sharedAs.split("\\."); + String endpoint = + PROTOCOL + "/shares/" + share + "/schemas/" + alias[0] + "/tables/" + alias[1] + "/query"; + + Response latest = query(endpoint, bearer, "{}", null); + assertVersion("0", latest); + assertParquetQuery(latest.text()); + assertEquals(1, count(latest.text(), "file"), latest.text()); + assertTrue(latest.text().contains(signature), latest.text()); + Response delta = query(endpoint, bearer, "{}", "responseformat=delta"); + assertVersion("0", delta); + assertDeltaQuery(delta.text()); + + // A historical query reports the version on metaData and on every file. + Response historical = query(endpoint, bearer, "{\"version\":0}", null); + assertVersion("0", historical); + assertEquals( + 2, historical.text().split("\"version\":0", -1).length - 1, historical.text()); + // A version after the latest is INVALID_PARAMETER_VALUE. + assertInvalid(query(endpoint, bearer, "{\"version\":1}", null)); + } + + private static void assertParquetQuery(String ndjson) { + assertTrue(ndjson.contains("\"protocol\""), ndjson); + assertTrue(ndjson.contains("\"metaData\""), ndjson); + assertTrue(ndjson.contains("\"schemaString\""), ndjson); + assertTrue(ndjson.contains("\"file\""), ndjson); + assertTrue(ndjson.contains("\"url\""), ndjson); + assertFalse(ndjson.contains("\"deltaProtocol\""), ndjson); + assertFalse(ndjson.contains("\"deltaSingleAction\""), ndjson); + } + + private static void assertDeltaQuery(String ndjson) { + assertTrue(ndjson.contains("\"deltaProtocol\""), ndjson); + assertTrue(ndjson.contains("\"deltaMetadata\""), ndjson); + assertTrue(ndjson.contains("\"deltaSingleAction\""), ndjson); + } + + /** Lines whose single key is {@code action}, such as file. */ + private static long count(String ndjson, String action) { + return Arrays.stream(ndjson.split("\n")) + .filter(line -> line.startsWith("{\"" + action + "\"")) + .count(); + } + + private static String shareTable(String share, String catalogName, String sharedAs) { + createShare(share); + addObject(share, "TABLE", catalogName, sharedAs); + String bearer = createAndActivateRecipient(share + "-partner"); + grant(share, share + "-partner"); + return bearer; + } + + private static Response query(String path, String bearer, String body, String capabilities) { + return capabilities == null + ? server.post(path, bearer, body) + : server.call( + HttpMethod.POST, path, bearer, body, "delta-sharing-capabilities", capabilities); + } + + private static void assertVersion(String expected, Response response) { + assertEquals(200, response.status(), response.text()); + assertEquals(expected, response.headers().get("Delta-Table-Version")); + } + + private static void assertInvalid(Response response) { + assertEquals(400, response.status(), response.text()); + assertEquals(ErrorCodes.INVALID_PARAMETER_VALUE, response.errorCode()); + } +} diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableQueryTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableQueryTest.java new file mode 100644 index 00000000..d2d15af8 --- /dev/null +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableQueryTest.java @@ -0,0 +1,125 @@ +package io.opensharing.protocol; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.linecorp.armeria.common.HttpMethod; +import com.linecorp.armeria.common.MediaType; +import io.opensharing.TestServer.Response; +import io.opensharing.http.ErrorCodes; +import java.util.ArrayList; +import java.util.List; +import org.junit.jupiter.api.Test; + +/** + * Query Table through the local catalog and Kernel, on a local Delta log with no data files; file + * URLs are signed only for cloud storage. + */ +class ProtocolTableQueryTest extends LocalDeltaTableSupport { + + private static final ObjectMapper JSON = new ObjectMapper(); + + @Test + void queriesGrantedTable() { + createShare("table-query"); + addObject("table-query", "TABLE", "main.ops.events", "ops.events"); + createShare("hidden-query"); + addObject("hidden-query", "TABLE", "main.sales.customers", "sales.customers"); + String bearer = createAndActivateRecipient("table-query-partner"); + grant("table-query", "table-query-partner"); + + // The latest snapshot defaults to parquet NDJSON: protocol, then metaData, then files. + String events = PROTOCOL + "/shares/TABLE-QUERY/schemas/OPS/tables/EVENTS/query"; + Response latest = query(events, bearer, "{}", null, null); + assertEquals(200, latest.status()); + assertTrue(latest.headers().contentType().is(MediaType.parse("application/x-ndjson"))); + assertEquals("1", latest.headers().get("Delta-Table-Version")); + assertEquals("responseformat=parquet", latest.headers().get("delta-sharing-capabilities")); + assertNull(latest.headers().get("fileidhash")); + List lines = lines(latest); + assertEquals(2, lines.size()); + assertEquals(1, lines.get(0).path("protocol").path("minReaderVersion").asInt()); + JsonNode metaData = lines.get(1).path("metaData"); + assertEquals("events", metaData.path("id").asText()); + assertEquals(location("events"), metaData.path("location").asText()); + assertTrue(metaData.path("version").isMissingNode()); + + // The body is optional. + assertEquals(200, query(events, bearer, null, null, null).status()); + + // fileidhash is echoed in lowercase; unsupported values are INVALID_PARAMETER_VALUE. + Response hashed = query(events, bearer, "{}", null, "PARQUET"); + assertEquals(200, hashed.status()); + assertEquals("parquet", hashed.headers().get("fileidhash")); + assertInvalid(query(events, bearer, "{}", null, "md5")); + + // responseformat=delta wraps Kernel protocol and metadata as deltaProtocol/deltaMetadata. + Response delta = query(events, bearer, "{}", "responseformat=delta", null); + assertEquals(200, delta.status()); + assertEquals("responseformat=delta", delta.headers().get("delta-sharing-capabilities")); + lines = lines(delta); + JsonNode deltaProtocol = lines.get(0).path("protocol").path("deltaProtocol"); + assertEquals(2, deltaProtocol.path("minWriterVersion").asInt()); + assertEquals("events", lines.get(1).path("metaData").path("deltaMetadata").path("id").asText()); + assertInvalid(query(events, bearer, "{}", "responseformat=json", null)); + + // A historical version or timestamp is echoed in metaData.version. + Response atVersion = query(events, bearer, "{\"version\":0}", null, null); + assertEquals(200, atVersion.status()); + assertEquals("0", atVersion.headers().get("Delta-Table-Version")); + assertEquals(0, lines(atVersion).get(1).path("metaData").path("version").asLong()); + Response atTimestamp = + query(events, bearer, "{\"timestamp\":\"2024-01-01T00:30:00Z\"}", null, null); + assertEquals(200, atTimestamp.status()); + assertEquals("0", atTimestamp.headers().get("Delta-Table-Version")); + + assertInvalid(query(events, bearer, "{\"version\":9}", null, null)); + String both = "{\"version\":1,\"timestamp\":\"2024-01-01T00:30:00Z\"}"; + assertInvalid(query(events, bearer, both, null, null)); + assertInvalid(query(events, bearer, "{\"timestamp\":\"not-a-timestamp\"}", null, null)); + + // A table in an ungranted share is 404, not 403. + Response hidden = + query( + PROTOCOL + "/shares/hidden-query/schemas/sales/tables/customers/query", + bearer, + "{}", + null, + null); + assertEquals(404, hidden.status()); + assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, hidden.errorCode()); + } + + private static Response query( + String path, String bearer, String body, String capabilities, String fileIdHash) { + List headers = new ArrayList<>(); + if (capabilities != null) { + headers.addAll(List.of("delta-sharing-capabilities", capabilities)); + } + if (fileIdHash != null) { + headers.addAll(List.of("fileidhash", fileIdHash)); + } + return server.call(HttpMethod.POST, path, bearer, body, headers.toArray(String[]::new)); + } + + private static List lines(Response response) { + List lines = new ArrayList<>(); + for (String line : response.text().split("\n")) { + try { + lines.add(JSON.readTree(line)); + } catch (JsonProcessingException e) { + throw new AssertionError("not an NDJSON line: " + line, e); + } + } + return lines; + } + + private static void assertInvalid(Response response) { + assertEquals(400, response.status()); + assertEquals(ErrorCodes.INVALID_PARAMETER_VALUE, response.errorCode()); + } +}