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 829436c1..56903f03 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 @@ -177,7 +177,8 @@ public TemporaryTableCredentialsResponse temporaryCredentials( * {@code version}, or an ISO-8601 {@code timestamp} in {@code request}, with presigned file * URLs. A latest-snapshot query can ask for a refresh token, and a later query can pass it back * to re-sign the same version. jsonPredicateHints and limitHint narrow the files returned. - * startingVersion queries are not implemented. + * {@code startingVersion}, with an optional {@code endingVersion}, returns the data change files + * committed in that range instead. */ public DeltaTableQueryReader.Result query( RecipientPrincipal recipient, @@ -195,7 +196,7 @@ public DeltaTableQueryReader.Result query( } QueryTableRequest body = request == null ? QueryTableRequest.EMPTY : request; if (body.startingVersion() != null || body.endingVersion() != null) { - throw ApiException.notImplemented("startingVersion queries are not supported"); + return queryChanges(body, resolved, granted, options); } boolean historical = body.version() != null || body.timestamp() != null; boolean includeRefreshToken = Boolean.TRUE.equals(body.includeRefreshToken()); @@ -222,6 +223,41 @@ public DeltaTableQueryReader.Result query( limitHint(body)); } + // Predicate and limit hints apply to snapshot files only, so they are ignored here. + private DeltaTableQueryReader.Result queryChanges( + QueryTableRequest body, + ResolvedAsset table, + ShareEntity share, + DeltaTableQueryReader.ResponseOptions options) { + Long startingVersion = body.startingVersion(); + Long endingVersion = body.endingVersion(); + if (startingVersion == null) { + throw ApiException.invalidParameter("endingVersion requires startingVersion"); + } + if (body.version() != null || body.timestamp() != null) { + throw ApiException.invalidParameter( + "Only one of version, timestamp, and startingVersion can be set"); + } + if (Boolean.TRUE.equals(body.includeRefreshToken()) || body.refreshToken() != null) { + throw ApiException.invalidParameter( + "includeRefreshToken and refreshToken cannot be used with startingVersion"); + } + if (startingVersion < 0) { + throw ApiException.invalidParameter("startingVersion must be non-negative"); + } + if (endingVersion != null && endingVersion < startingVersion) { + throw ApiException.invalidParameter( + "endingVersion must be greater than or equal to startingVersion"); + } + return queryReader.readChanges( + table, + owner(share), + options, + startingVersion, + endingVersion, + Boolean.TRUE.equals(body.includeHistoricalProtocol())); + } + // limitHint counts unfiltered file records, so it is dropped when any predicate hint is present. private static Long limitHint(QueryTableRequest body) { boolean hasPredicate = 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 index 05c65a86..fdf093ba 100644 --- 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 @@ -3,8 +3,8 @@ import java.util.List; /** - * Optional body for Query Table. Snapshot queries use version or timestamp; streaming is not - * implemented. + * Optional body for Query Table. Snapshot queries use version or timestamp; startingVersion, with + * an optional endingVersion, returns data change files instead. */ public record QueryTableRequest( // Deprecated by the protocol in favor of jsonPredicateHints; only logged, never used to filter. 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 902498ba..3be5baa8 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 @@ -99,6 +99,40 @@ public static String parquetFile( expirationTimestamp))); } + /** Data change file kinds, keyed {@code add} or {@code remove}. */ + public enum ChangeType { + ADD, + REMOVE + } + + /** A data change file for a startingVersion query. */ + public static String parquetChange( + ChangeType type, + String url, + String id, + Map partitionValues, + long size, + String stats, + long version, + long timestamp, + Long expirationTimestamp) { + ParquetFile file = + new ParquetFile( + url, + id, + partitionValues == null ? Map.of() : partitionValues, + size, + stats, + version, + timestamp, + expirationTimestamp); + return line( + switch (type) { + case ADD -> new AddLine(file); + case REMOVE -> new RemoveLine(file); + }); + } + public static String deltaFile( String id, String deletionVectorFileId, @@ -106,15 +140,47 @@ public static String deltaFile( Long version, Long timestamp, AddFile add) { + return deltaFile( + id, + deletionVectorFileId, + expirationTimestamp, + version, + timestamp, + new DeltaSingleAction(add.toRow(), null)); + } + + /** {@code action} is the add or remove row; {@code type} picks its action key. */ + public static String deltaChange( + String id, + String deletionVectorFileId, + Long expirationTimestamp, + long version, + long timestamp, + ChangeType type, + Row action) { + return deltaFile( + id, + deletionVectorFileId, + expirationTimestamp, + version, + timestamp, + switch (type) { + case ADD -> new DeltaSingleAction(action, null); + case REMOVE -> new DeltaSingleAction(null, action); + }); + } + + private static String deltaFile( + String id, + String deletionVectorFileId, + Long expirationTimestamp, + Long version, + Long timestamp, + DeltaSingleAction action) { return line( new FileLine( new DeltaFile( - id, - deletionVectorFileId, - version, - timestamp, - expirationTimestamp, - new DeltaSingleAction(add)))); + id, deletionVectorFileId, version, timestamp, expirationTimestamp, action))); } public static String endStreamAction(String refreshToken, Long minUrlExpirationTimestamp) { @@ -126,7 +192,11 @@ public static String endStreamAction(String refreshToken, Long minUrlExpirationT * 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(); + return new AddFile(withPath(add.toRow(), path, deletionVectorUrl)); + } + + /** {@link #withPath(AddFile, String, String)} for an add or remove action row. */ + public static Row withPath(Row row, String path, String deletionVectorUrl) { Map overrides = new HashMap<>(); overrides.put(row.getSchema().indexOf("path"), path); if (deletionVectorUrl != null) { @@ -142,7 +212,7 @@ public static AddFile withPath(AddFile add, String path, String deletionVectorUr dv.getSchema().indexOf("pathOrInlineDv"), deletionVectorUrl))); } - return new AddFile(new DelegateRow(row, overrides)); + return new DelegateRow(row, overrides); } private static String line(Object value) { @@ -177,6 +247,7 @@ public void serialize(Object value, JsonGenerator gen, SerializerProvider serial case Protocol protocol -> protocol.toRow(); case Metadata metadata -> metadata.toRow(); case AddFile add -> add.toRow(); + case Row action -> action; default -> throw new IllegalArgumentException( "unsupported Kernel action: " + value.getClass().getName()); @@ -257,8 +328,14 @@ public record DeltaFile( DeltaSingleAction deltaSingleAction) implements FileBody {} + @JsonInclude(JsonInclude.Include.NON_NULL) public record DeltaSingleAction( - @JsonSerialize(using = KernelActionSerializer.class) AddFile add) {} + @JsonSerialize(using = KernelActionSerializer.class) Row add, + @JsonSerialize(using = KernelActionSerializer.class) Row remove) {} + + public record AddLine(ParquetFile add) {} + + public record RemoveLine(ParquetFile remove) {} public record EndStreamLine(EndStreamAction endStreamAction) {} 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 index d17d86b6..5cc28e80 100644 --- 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 @@ -4,9 +4,13 @@ import io.delta.kernel.Scan; import io.delta.kernel.ScanBuilder; import io.delta.kernel.Snapshot; +import io.delta.kernel.data.ColumnarBatch; import io.delta.kernel.data.FilteredColumnarBatch; import io.delta.kernel.data.Row; +import io.delta.kernel.exceptions.KernelException; import io.delta.kernel.expressions.Predicate; +import io.delta.kernel.internal.DeltaLogActionUtils; +import io.delta.kernel.internal.DeltaLogActionUtils.DeltaAction; import io.delta.kernel.internal.InternalScanFileUtils; import io.delta.kernel.internal.ScanImpl; import io.delta.kernel.internal.SnapshotImpl; @@ -14,7 +18,14 @@ 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.actions.RemoveFile; import io.delta.kernel.internal.checksum.CRCInfo; +import io.delta.kernel.internal.fs.Path; +import io.delta.kernel.internal.util.FileNames; +import io.delta.kernel.internal.util.Utils; +import io.delta.kernel.internal.util.VectorUtils; +import io.delta.kernel.types.LongType; +import io.delta.kernel.types.StructType; import io.delta.kernel.utils.CloseableIterator; import io.delta.kernel.utils.FileStatus; import io.opensharing.asset.table.DeltaSharingCapabilities; @@ -22,6 +33,7 @@ import io.opensharing.asset.table.FileIdHash; import io.opensharing.asset.table.RefreshTokens; import io.opensharing.asset.table.TableActions; +import io.opensharing.asset.table.TableActions.ChangeType; import io.opensharing.asset.table.signer.SignedUrl; import io.opensharing.asset.table.signer.UrlSigners; import io.opensharing.auth.AuthContext; @@ -29,27 +41,43 @@ import io.opensharing.http.ApiException; import java.io.IOException; import java.io.UncheckedIOException; +import java.net.URI; import java.time.Duration; import java.time.Instant; +import java.util.ArrayList; +import java.util.List; 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, and an endStreamAction when requested. + * Reads Query Table files from a Delta log. The parquet or delta response is NDJSON: protocol, + * metaData, then one file action per data file, or one add or remove per data change file for a + * startingVersion query, and an endStreamAction when requested. * *

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. * - *

jsonPredicateHints and limitHint only decide which files are returned. Clients still filter - * rows and apply the limit themselves, so returning extra files is always correct. + *

jsonPredicateHints and limitHint only decide which snapshot files are returned. Clients still + * filter rows and apply the limit themselves, so returning extra files is always correct. */ public class DeltaTableQueryReader { private static final Duration URL_TTL = Duration.ofHours(1); private static final ObjectMapper JSON = new ObjectMapper(); + // Commit files are read directly because Kernel 4.0's getChanges requires commitInfo fields that + // older Delta writers omit; only the commitInfo fields used here are read. + private static final StructType CHANGE_SCHEMA = + new StructType() + .add(DeltaAction.PROTOCOL.colName, DeltaAction.PROTOCOL.schema) + .add(DeltaAction.METADATA.colName, DeltaAction.METADATA.schema) + .add(DeltaAction.ADD.colName, DeltaAction.ADD.schema) + .add(DeltaAction.REMOVE.colName, DeltaAction.REMOVE.schema) + .add( + DeltaAction.COMMITINFO.colName, + new StructType().add("inCommitTimestamp", LongType.LONG)); + private final DeltaKernel kernel; private final UrlSigners signers; @@ -139,6 +167,135 @@ public Result read( return new Result(snapshot.getVersion(), out.ndjson()); } + /** + * Reads data change files committed from {@code startingVersion} through {@code endingVersion}, + * or the latest version when it is null. Protocol and metaData lines come from {@code + * startingVersion}; later metaData changes follow with their version, and later protocol changes + * too when {@code includeHistoricalProtocol} is set on a delta response. Versions after the + * latest are rejected rather than capped. + */ + public Result readChanges( + ResolvedAsset table, + AuthContext auth, + ResponseOptions options, + long startingVersion, + Long endingVersion, + boolean includeHistoricalProtocol) { + DeltaKernel.Session session = kernel.open(table, auth, URL_TTL); + long latest = DeltaSnapshots.open(session, null, null).getVersion(); + requireAtMostLatest("startingVersion", startingVersion, latest); + if (endingVersion != null) { + requireAtMostLatest("endingVersion", endingVersion, latest); + } + return changes( + session, + table, + (SnapshotImpl) DeltaSnapshots.open(session, startingVersion, null), + new ChangeRange( + startingVersion, + endingVersion == null ? latest : endingVersion, + includeHistoricalProtocol), + options); + } + + /** + * Protocol and metaData from {@code header}, then each commit in {@code range} with its add and + * remove files that have dataChange. + */ + private Result changes( + DeltaKernel.Session session, + ResolvedAsset table, + SnapshotImpl header, + ChangeRange range, + ResponseOptions options) { + ResponseWriter out = new ResponseWriter(session, table, options); + out.requireReaderFeatures(header.getProtocol()); + // With includeHistoricalProtocol, the starting protocol carries its version like later ones. + out.protocol( + header.getProtocol(), + range.includeHistoricalProtocol() && out.format == ResponseFormat.DELTA + ? header.getVersion() + : null); + out.metadata(header.getMetadata(), header.getVersion(), crcFor(header, header.getVersion())); + + Path tablePath = new Path(session.table().getPath(session.engine())); + try { + for (FileStatus file : + DeltaLogActionUtils.getCommitFilesForVersionRange( + session.engine(), tablePath, range.start(), range.end())) { + write(out, range, readCommit(session, file)); + } + } catch (IOException e) { + throw new UncheckedIOException(e); + } catch (KernelException | IllegalArgumentException invalid) { + // Kernel fails when a commit file in the range is missing, such as after log cleanup. + throw ApiException.invalidParameter(invalid.getMessage()); + } + + if (options.includeEndStreamAction()) { + out.endStreamAction(null); + } + // The protocol reports the starting version as the table version of a change response. + return new Result(range.start(), out.ndjson()); + } + + /** + * Reads one commit file, which holds every action of its version. Kernel can split a large file + * across several batches, each with its own schema ordinals. + */ + private static Commit readCommit(DeltaKernel.Session session, FileStatus file) + throws IOException { + Commit commit = new Commit(FileNames.deltaVersion(file.getPath()), file.getModificationTime()); + try (CloseableIterator batches = + session + .engine() + .getJsonHandler() + .readJsonFiles( + Utils.singletonCloseableIterator(file), CHANGE_SCHEMA, Optional.empty())) { + while (batches.hasNext()) { + ColumnarBatch batch = batches.next(); + ChangeColumns columns = ChangeColumns.of(batch.getSchema()); + try (CloseableIterator rows = batch.getRows()) { + while (rows.hasNext()) { + commit.add(rows.next(), columns); + } + } + } + } + return commit; + } + + private static void write(ResponseWriter out, ChangeRange range, Commit commit) { + long version = commit.version; + // The header already covers the starting version's protocol and metaData. + boolean later = version > range.start(); + // A commit has a protocol only when it creates the table or upgrades the protocol. + for (Protocol protocol : commit.protocols) { + // A later protocol can need reader features the client did not declare. + out.requireReaderFeatures(protocol); + if (range.includeHistoricalProtocol() && out.format == ResponseFormat.DELTA && later) { + out.protocol(protocol, version); + } + } + // A commit has metaData only when it creates the table or changes its schema or properties. + for (Metadata metadata : commit.metadata) { + if (later) { + out.metadata(metadata, version, null); + } + } + // A commit has add and remove files only when it writes or rewrites data. + for (ChangeFile file : commit.files) { + out.changeFile(file, version, commit.timestamp); + } + } + + private static void requireAtMostLatest(String name, long version, long latest) { + if (version > latest) { + throw ApiException.invalidParameter( + name + " " + version + " is after the latest table version " + latest); + } + } + // Files without stats count as zero records, so limitHint never stops early on them. private static long numRecords(String stats) { if (stats == null || stats.isBlank()) { @@ -188,6 +345,7 @@ private final class ResponseWriter { private final String fileIdHash; private final StringBuilder ndjson = new StringBuilder(); private Long minUrlExpirationTimestamp; + private String tableRoot; ResponseWriter(DeltaKernel.Session session, ResolvedAsset table, ResponseOptions options) { this.format = DeltaSharingCapabilities.choose(options.capabilities()); @@ -255,6 +413,38 @@ long snapshotFile(Row scanFile, Long version, Long timestamp) { - add.getDeletionVector().map(DeletionVectorDescriptor::getCardinality).orElse(0L); } + /** Appends a data change file committed at {@code version}, unless it has no dataChange. */ + void changeFile(ChangeFile file, long version, long timestamp) { + // Add and remove files without dataChange, such as compaction rewrites, are skipped. + if (!file.dataChange()) { + return; + } + SignedFile signed = sign(absolutePath(file.path()), file.path(), file.deletionVector()); + if (format == ResponseFormat.DELTA) { + ndjson.append( + TableActions.deltaChange( + signed.id(), + signed.deletionVectorFileId(), + signed.expirationTimestamp(), + version, + timestamp, + file.type(), + TableActions.withPath(file.action(), signed.url(), signed.deletionVectorUrl()))); + } else { + ndjson.append( + TableActions.parquetChange( + file.type(), + signed.url(), + signed.id(), + file.partitionValues(), + file.size(), + file.stats(), + version, + timestamp, + signed.expirationTimestamp())); + } + } + /** {@code refreshToken} may be null. */ void endStreamAction(String refreshToken) { ndjson.append(TableActions.endStreamAction(refreshToken, minUrlExpirationTimestamp)); @@ -298,6 +488,15 @@ private SignedFile sign( deletionVectorFileId, expirationTimestamp); } + + // Log paths are URI-encoded and usually relative to the table root; resolving an absolute + // path against the root keeps it unchanged. + private String absolutePath(String path) { + if (tableRoot == null) { + tableRoot = session.table().getPath(session.engine()); + } + return new Path(new Path(URI.create(tableRoot)), new Path(URI.create(path))).toString(); + } } /** {@code expirationTimestamp} is the earlier of the file and deletion vector URL expiries. */ @@ -307,4 +506,108 @@ private record SignedFile( String deletionVectorUrl, String deletionVectorFileId, long expirationTimestamp) {} + + /** + * The versions and options of a startingVersion query, which also returns later metaData + * changes. + * + * @param includeHistoricalProtocol return protocol changes after {@code start} in delta responses + */ + private record ChangeRange(long start, long end, boolean includeHistoricalProtocol) {} + + /** Ordinals of a commit file batch. */ + private record ChangeColumns(int protocol, int metadata, int add, int remove, int commitInfo) { + + static ChangeColumns of(StructType schema) { + return new ChangeColumns( + schema.indexOf(DeltaAction.PROTOCOL.colName), + schema.indexOf(DeltaAction.METADATA.colName), + schema.indexOf(DeltaAction.ADD.colName), + schema.indexOf(DeltaAction.REMOVE.colName), + schema.indexOf(DeltaAction.COMMITINFO.colName)); + } + } + + /** The actions of one commit, buffered so its protocol and metaData go out before its files. */ + private static final class Commit { + final long version; + long timestamp; + final List protocols = new ArrayList<>(); + final List metadata = new ArrayList<>(); + final List files = new ArrayList<>(); + + /** + * The commit file's {@code modificationTime} is the timestamp unless commitInfo has an + * inCommitTimestamp, which tables with in-commit timestamps record because modification times + * change when the log is copied. + */ + Commit(long version, long modificationTime) { + this.version = version; + this.timestamp = modificationTime; + } + + // Each line of a commit file holds exactly one action, so one column of the row is set. + void add(Row row, ChangeColumns columns) { + if (!row.isNullAt(columns.protocol())) { + protocols.add(Protocol.fromRow(row.getStruct(columns.protocol()))); + } else if (!row.isNullAt(columns.metadata())) { + metadata.add(Metadata.fromRow(row.getStruct(columns.metadata()))); + } else if (!row.isNullAt(columns.add())) { + files.add(ChangeFile.of(ChangeType.ADD, row.getStruct(columns.add()))); + } else if (!row.isNullAt(columns.remove())) { + files.add(ChangeFile.of(ChangeType.REMOVE, row.getStruct(columns.remove()))); + } else if (!row.isNullAt(columns.commitInfo())) { + Row commitInfo = row.getStruct(columns.commitInfo()); + int inCommitTimestamp = commitInfo.getSchema().indexOf("inCommitTimestamp"); + if (!commitInfo.isNullAt(inCommitTimestamp)) { + timestamp = commitInfo.getLong(inCommitTimestamp); + } + } + } + } + + /** The fields a change line needs from an add or remove action in the log. */ + private record ChangeFile( + ChangeType type, + Row action, + String path, + boolean dataChange, + Map partitionValues, + long size, + String stats, + Optional deletionVector) { + + static ChangeFile of(ChangeType type, Row action) { + return switch (type) { + case ADD -> { + AddFile add = new AddFile(action); + yield new ChangeFile( + type, + action, + add.getPath(), + add.getDataChange(), + VectorUtils.toJavaMap(add.getPartitionValues()), + add.getSize(), + add.getStatsJson().orElse(null), + add.getDeletionVector()); + } + // A remove may omit partitionValues and size. + case REMOVE -> { + RemoveFile remove = new RemoveFile(action); + yield new ChangeFile( + type, + action, + remove.getPath(), + remove.getDataChange(), + remove + .getPartitionValues() + .>map(VectorUtils::toJavaMap) + .orElse(Map.of()), + remove.getSize().orElse(0L), + null, + remove.getDeletionVector()); + } + }; + } + } } diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/delta/DeltaTableQueryReaderTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/delta/DeltaTableQueryReaderTest.java index 6ad94586..8341e0a6 100644 --- a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/delta/DeltaTableQueryReaderTest.java +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/delta/DeltaTableQueryReaderTest.java @@ -1,8 +1,12 @@ package io.opensharing.asset.table.delta; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; import io.delta.kernel.Table; import io.delta.kernel.defaults.engine.DefaultEngine; import io.delta.kernel.engine.Engine; @@ -16,6 +20,7 @@ import io.opensharing.catalog.ResolvedAsset; import io.opensharing.catalog.StorageCredentials; import io.opensharing.catalog.TableProperties; +import io.opensharing.http.ApiException; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; @@ -128,6 +133,133 @@ void limitHintStopsOnceFilesCoverTheLimit() { assertEquals(3, files(null, 1000L).size()); } + @Test + void startingVersionReturnsDataChangeFilesAndHistoricalMetadata() throws IOException { + writeChanges(); + List lines = changes(1L, 2L, null, false); + assertEquals("{\"protocol\":{\"minReaderVersion\":1}}", lines.get(0)); + assertTrue(lines.get(1).contains("\"version\":1"), lines.get(1)); + assertEquals( + List.of("remove a 1", "add d 1", "metaData 2", "add e 2"), + describe(lines.subList(2, lines.size()))); + assertTrue(lines.get(2).contains("\"timestamp\":"), lines.get(2)); + assertFalse(lines.get(2).contains("\"stats\":"), lines.get(2)); + assertTrue(lines.get(3).contains("\"stats\":"), lines.get(3)); + + // Parquet cannot carry the deletion vector protocol committed at version 3. + assertThrows(ApiException.class, () -> changes(1L, null, null, false)); + } + + @Test + void includeHistoricalProtocolAddsLaterProtocolsToDeltaResponses() throws IOException { + writeChanges(); + String delta = "responseformat=delta;readerfeatures=deletionvectors"; + // The compaction at version 4 rewrites files without dataChange, so it adds nothing. + assertEquals( + List.of("remove a 1", "add d 1", "metaData 2", "add e 2"), + describe(changes(1L, null, delta, false).subList(2, 6))); + List withoutProtocol = changes(1L, null, delta, false); + assertEquals(6, withoutProtocol.size()); + assertFalse(withoutProtocol.get(0).contains("\"version\""), withoutProtocol.get(0)); + List withProtocol = changes(1L, null, delta, true); + assertEquals( + List.of("remove a 1", "add d 1", "metaData 2", "add e 2", "protocol 3"), + describe(withProtocol.subList(2, withProtocol.size()))); + assertTrue(withProtocol.get(0).contains("\"version\":1"), withProtocol.get(0)); + } + + @Test + void rejectsVersionsAfterTheLatestVersion() throws IOException { + writeChanges(); + assertThrows(ApiException.class, () -> changes(5L, null, null, false)); + assertThrows(ApiException.class, () -> changes(1L, 5L, null, false)); + } + + private List changes( + long startingVersion, Long endingVersion, String capabilities, boolean historicalProtocol) { + return List.of( + reader + .readChanges( + table, + null, + new DeltaTableQueryReader.ResponseOptions(capabilities, null, false), + startingVersion, + endingVersion, + historicalProtocol) + .ndjson() + .split("\n")); + } + + /** "add d 1": action, file name, and version; metaData and protocol lines show their version. */ + private static List describe(List lines) { + return lines.stream() + .map( + line -> { + JsonNode node = parse(line); + String key = node.fieldNames().next(); + JsonNode body = node.get(key); + if (key.equals("file")) { + JsonNode action = body.get("deltaSingleAction"); + key = action.fieldNames().next(); + String name = fileName(action.get(key).get("path")); + return key + " " + name + " " + body.get("version"); + } + if (key.equals("add") || key.equals("remove")) { + return key + " " + fileName(body.get("url")) + " " + body.get("version"); + } + return key + " " + body.get("version"); + }) + .toList(); + } + + private static String fileName(JsonNode url) { + return url.asText().replaceAll(".*https://signed/(\\w)\\.parquet.*", "$1"); + } + + private static JsonNode parse(String line) { + try { + return new ObjectMapper().readTree(line); + } catch (IOException e) { + throw new IllegalStateException(e); + } + } + + // 1: delete a, insert d. 2: change metaData, insert e. 3: enable deletion vectors. 4: compact. + private void writeChanges() throws IOException { + Path log = dir.resolve("_delta_log"); + Files.writeString( + log.resolve("00000000000000000001.json"), + remove("a", "2021-01-01", true) + "\n" + add("d", "2021-01-03", 31, 40) + "\n"); + Files.writeString( + log.resolve("00000000000000000002.json"), + "{\"metaData\":{\"id\":\"t\",\"format\":{\"provider\":\"parquet\",\"options\":{}}," + + "\"schemaString\":\"" + + SCHEMA + + "\",\"partitionColumns\":[\"date\"],\"configuration\":{\"k\":\"v\"}," + + "\"createdTime\":0}}\n" + + add("e", "2021-01-03", 41, 50) + + "\n"); + Files.writeString( + log.resolve("00000000000000000003.json"), + "{\"protocol\":{\"minReaderVersion\":3,\"minWriterVersion\":7," + + "\"readerFeatures\":[\"deletionVectors\"]," + + "\"writerFeatures\":[\"deletionVectors\"]}}\n"); + Files.writeString( + log.resolve("00000000000000000004.json"), + remove("b", "2021-01-02", false) + + "\n" + + remove("c", "2021-01-02", false) + + "\n" + + add("f", "2021-01-02", 11, 30).replace("\"dataChange\":true", "\"dataChange\":false") + + "\n"); + } + + private static String remove(String name, String date, boolean dataChange) { + return ("{\"remove\":{\"path\":\"date=%s/%s.parquet\",\"deletionTimestamp\":1," + + "\"dataChange\":%s,\"partitionValues\":{\"date\":\"%s\"},\"size\":1}}") + .formatted(date, name, dataChange, date); + } + private String ndjson(String jsonPredicateHints, Long limitHint) { return reader .read( diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableStartingVersionIntegrationTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableStartingVersionIntegrationTest.java new file mode 100644 index 00000000..4ab19193 --- /dev/null +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableStartingVersionIntegrationTest.java @@ -0,0 +1,104 @@ +package io.opensharing.protocol; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.linecorp.armeria.common.HttpMethod; +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; + +/** + * Query Table startingVersion and endingVersion on real cloud Delta logs, with presigned file + * URLs. Individual cloud cases are skipped unless their repository credentials are present. + */ +@Timeout(60) +class ProtocolTableStartingVersionIntegrationTest 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 queriesTable1Changes() { + queriesChanges("table1-changes", "main.sales.table1", "sales.table1", 1, 2); + } + + @Test + @EnabledIfEnvironmentVariable(named = "AZURE_TEST_ACCOUNT_KEY", matches = ".+") + void queriesAzureTableChanges() { + queriesChanges("azure-changes", "main.sales.azure", "sales.azure", 0, 0); + } + + @Test + @EnabledIfEnvironmentVariable(named = "GOOGLE_APPLICATION_CREDENTIALS", matches = ".+") + void queriesGoogleTableChanges() { + queriesChanges("gcs-changes", "main.sales.gcs", "sales.gcs", 0, 0); + } + + private static void queriesChanges( + String share, String catalogName, String sharedAs, long firstAdds, long latest) { + createShare(share); + addObject(share, "TABLE", catalogName, sharedAs); + String bearer = createAndActivateRecipient(share + "-partner"); + grant(share, share + "-partner"); + String[] alias = sharedAs.split("\\."); + String endpoint = + PROTOCOL + "/shares/" + share + "/schemas/" + alias[0] + "/tables/" + alias[1] + "/query"; + + // Each file added at firstAdds comes back as an add with its version. + String range = "{\"startingVersion\":" + firstAdds + ",\"endingVersion\":" + firstAdds + "}"; + String parquet = ok(query(endpoint, bearer, range, null), "" + firstAdds); + assertTrue(count(parquet, "add") > 0, parquet); + assertEquals(0, count(parquet, "file"), parquet); + assertTrue(parquet.contains("\"version\":" + firstAdds), parquet); + String delta = ok(query(endpoint, bearer, range, "responseformat=delta"), "" + firstAdds); + assertEquals(count(parquet, "add"), count(delta, "file"), delta); + assertTrue(delta.contains("\"add\""), delta); + // Without endingVersion the range runs through the latest version. + ok(query(endpoint, bearer, "{\"startingVersion\":" + latest + "}", null), "" + latest); + + // Versions after the latest, and a range that ends before it starts, are rejected. + for (String body : + new String[] { + "{\"startingVersion\":" + (latest + 1) + "}", + "{\"startingVersion\":0,\"endingVersion\":" + (latest + 1) + "}", + "{\"startingVersion\":1,\"endingVersion\":0}" + }) { + Response response = query(endpoint, bearer, body, null); + assertEquals(400, response.status(), body); + assertEquals(ErrorCodes.INVALID_PARAMETER_VALUE, response.errorCode(), body); + } + } + + /** Lines whose single key is {@code action}, such as add or file. */ + private static long count(String ndjson, String action) { + return Arrays.stream(ndjson.split("\n")) + .filter(line -> line.startsWith("{\"" + action + "\"")) + .count(); + } + + 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); + } + + /** Expects 200 with {@code version} as the Delta-Table-Version and returns the NDJSON body. */ + private static String ok(Response response, String version) { + assertEquals(200, response.status(), response.text()); + assertEquals(version, response.headers().get("Delta-Table-Version")); + return response.text(); + } +} diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableStartingVersionTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableStartingVersionTest.java new file mode 100644 index 00000000..80978d25 --- /dev/null +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableStartingVersionTest.java @@ -0,0 +1,62 @@ +package io.opensharing.protocol; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.linecorp.armeria.common.HttpMethod; +import io.opensharing.TestServer.Response; +import io.opensharing.http.ErrorCodes; +import java.util.List; +import org.junit.jupiter.api.Test; + +/** Query Table from a startingVersion, on a local Delta log with no data files. */ +class ProtocolTableStartingVersionTest extends LocalDeltaTableSupport { + + @Test + void queriesGrantedTableChangesFromStartingVersion() { + createShare("table-changes"); + addObject("table-changes", "TABLE", "main.ops.events", "ops.events"); + String bearer = createAndActivateRecipient("table-changes-partner"); + grant("table-changes", "table-changes-partner"); + String query = PROTOCOL + "/shares/table-changes/schemas/ops/tables/events/query"; + + // Delta-Table-Version is the startingVersion, and the protocol and metaData lead. + Response changes = + server.call( + HttpMethod.POST, + query, + bearer, + "{\"startingVersion\":0,\"endingVersion\":1,\"includeHistoricalProtocol\":true}", + "delta-sharing-capabilities", + "includeendstreamaction=true"); + assertEquals(200, changes.status(), changes.text()); + assertEquals("0", changes.headers().get("Delta-Table-Version")); + assertEquals( + "responseformat=parquet;includeendstreamaction=true", + changes.headers().get("delta-sharing-capabilities")); + String[] lines = changes.text().split("\n"); + assertTrue(lines[0].startsWith("{\"protocol\""), changes.text()); + assertTrue(lines[1].startsWith("{\"metaData\""), changes.text()); + assertTrue(lines[lines.length - 1].startsWith("{\"endStreamAction\""), changes.text()); + // Without endingVersion the range runs through the latest version. + Response latest = server.post(query, bearer, "{\"startingVersion\":1}"); + assertEquals(200, latest.status(), latest.text()); + assertEquals("1", latest.headers().get("Delta-Table-Version")); + + for (String invalid : + List.of( + "{\"endingVersion\":1}", + "{\"startingVersion\":-1}", + "{\"startingVersion\":1,\"endingVersion\":0}", + "{\"startingVersion\":2}", + "{\"startingVersion\":0,\"endingVersion\":2}", + "{\"startingVersion\":0,\"version\":0}", + "{\"startingVersion\":0,\"timestamp\":\"2024-01-01T00:00:00Z\"}", + "{\"startingVersion\":0,\"includeRefreshToken\":true}", + "{\"startingVersion\":0,\"refreshToken\":\"token\"}")) { + Response response = server.post(query, bearer, invalid); + assertEquals(400, response.status(), invalid); + assertEquals(ErrorCodes.INVALID_PARAMETER_VALUE, response.errorCode(), invalid); + } + } +}