From 52078eea0c0ce27299401c2002b01f7159d4decc Mon Sep 17 00:00:00 2001 From: Sun Cao Date: Wed, 14 Oct 2026 10:42:38 -0700 Subject: [PATCH] Serve the Change Data Feed of shared Delta tables, on Armeria. --- .../asset/table/ProtocolTables.java | 42 +++- .../opensharing/asset/table/TableActions.java | 22 +- .../asset/table/TableChangesRequest.java | 14 ++ .../table/delta/DeltaTableQueryReader.java | 194 ++++++++++++++++-- .../protocol/DeltaSharingProtocolService.java | 62 +++++- .../delta/DeltaTableQueryReaderTest.java | 111 +++++++++- .../ProtocolTableChangesIntegrationTest.java | 128 ++++++++++++ .../protocol/ProtocolTableChangesTest.java | 44 ++++ .../test/resources/local-catalog-cloud.yml | 7 + 9 files changed, 585 insertions(+), 39 deletions(-) create mode 100644 server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableChangesRequest.java create mode 100644 server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableChangesIntegrationTest.java create mode 100644 server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableChangesTest.java 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 56903f03..039a6665 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 @@ -188,12 +188,7 @@ public DeltaTableQueryReader.Result query( 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"); - } + ResolvedAsset resolved = requireUrlAccess(resolve(granted, schema, table), schema, table); QueryTableRequest body = request == null ? QueryTableRequest.EMPTY : request; if (body.startingVersion() != null || body.endingVersion() != null) { return queryChanges(body, resolved, granted, options); @@ -223,6 +218,41 @@ public DeltaTableQueryReader.Result query( limitHint(body)); } + /** + * The Change Data Feed of a table in the share that supports url access and records it, from + * {@code startingVersion} or an ISO-8601 {@code startingTimestamp} through {@code endingVersion}, + * {@code endingTimestamp}, or the latest version, with presigned file URLs. + */ + public DeltaTableQueryReader.Result changes( + RecipientPrincipal recipient, + String share, + String schema, + String table, + TableChangesRequest request, + DeltaTableQueryReader.ResponseOptions options) { + ShareEntity granted = shares.requireGranted(recipient, share); + ResolvedAsset resolved = requireUrlAccess(resolve(granted, schema, table), schema, table); + return queryReader.readChangeDataFeed( + resolved, + owner(granted), + options, + request.startingVersion(), + parseTimestamp("startingTimestamp", request.startingTimestamp()), + request.endingVersion(), + parseTimestamp("endingTimestamp", request.endingTimestamp()), + Boolean.TRUE.equals(request.includeHistoricalMetadata()), + Boolean.TRUE.equals(request.includeHistoricalProtocol())); + } + + private static ResolvedAsset requireUrlAccess(ResolvedAsset table, String schema, String name) { + List accessModes = TableAccessModes.forTable(table); + if (accessModes == null || !accessModes.contains("url")) { + throw ApiException.invalidParameter( + "table '" + schema + "." + name + "' does not support url access"); + } + return table; + } + // Predicate and limit hints apply to snapshot files only, so they are ignored here. private DeltaTableQueryReader.Result queryChanges( QueryTableRequest body, 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 3be5baa8..c2075f81 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,13 +99,14 @@ public static String parquetFile( expirationTimestamp))); } - /** Data change file kinds, keyed {@code add} or {@code remove}. */ + /** Data change file kinds, keyed {@code add}, {@code remove}, or {@code cdf} in parquet. */ public enum ChangeType { ADD, - REMOVE + REMOVE, + CDC } - /** A data change file for a startingVersion query. */ + /** A data change file for a startingVersion or Change Data Feed query. */ public static String parquetChange( ChangeType type, String url, @@ -130,6 +131,7 @@ public static String parquetChange( switch (type) { case ADD -> new AddLine(file); case REMOVE -> new RemoveLine(file); + case CDC -> new CdfLine(file); }); } @@ -146,10 +148,10 @@ public static String deltaFile( expirationTimestamp, version, timestamp, - new DeltaSingleAction(add.toRow(), null)); + new DeltaSingleAction(add.toRow(), null, null)); } - /** {@code action} is the add or remove row; {@code type} picks its action key. */ + /** {@code action} is the add, remove, or cdc row; {@code type} picks its action key. */ public static String deltaChange( String id, String deletionVectorFileId, @@ -165,8 +167,9 @@ public static String deltaChange( version, timestamp, switch (type) { - case ADD -> new DeltaSingleAction(action, null); - case REMOVE -> new DeltaSingleAction(null, action); + case ADD -> new DeltaSingleAction(action, null, null); + case REMOVE -> new DeltaSingleAction(null, action, null); + case CDC -> new DeltaSingleAction(null, null, action); }); } @@ -331,12 +334,15 @@ public record DeltaFile( @JsonInclude(JsonInclude.Include.NON_NULL) public record DeltaSingleAction( @JsonSerialize(using = KernelActionSerializer.class) Row add, - @JsonSerialize(using = KernelActionSerializer.class) Row remove) {} + @JsonSerialize(using = KernelActionSerializer.class) Row remove, + @JsonSerialize(using = KernelActionSerializer.class) Row cdc) {} public record AddLine(ParquetFile add) {} public record RemoveLine(ParquetFile remove) {} + public record CdfLine(ParquetFile cdf) {} + public record EndStreamLine(EndStreamAction endStreamAction) {} @JsonInclude(JsonInclude.Include.NON_NULL) diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableChangesRequest.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableChangesRequest.java new file mode 100644 index 00000000..feef7258 --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/TableChangesRequest.java @@ -0,0 +1,14 @@ +package io.opensharing.asset.table; + +/** + * Query parameters of Query Table Changes. The range starts at {@code startingVersion} or {@code + * startingTimestamp}, and ends at {@code endingVersion}, {@code endingTimestamp}, or the latest + * version; timestamps are ISO-8601. + */ +public record TableChangesRequest( + Long startingVersion, + String startingTimestamp, + Long endingVersion, + String endingTimestamp, + Boolean includeHistoricalMetadata, + Boolean includeHistoricalProtocol) {} 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 5cc28e80..182f5624 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 @@ -14,6 +14,8 @@ import io.delta.kernel.internal.InternalScanFileUtils; import io.delta.kernel.internal.ScanImpl; import io.delta.kernel.internal.SnapshotImpl; +import io.delta.kernel.internal.TableConfig; +import io.delta.kernel.internal.TableImpl; import io.delta.kernel.internal.actions.AddFile; import io.delta.kernel.internal.actions.DeletionVectorDescriptor; import io.delta.kernel.internal.actions.Metadata; @@ -25,6 +27,8 @@ 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.MapType; +import io.delta.kernel.types.StringType; import io.delta.kernel.types.StructType; import io.delta.kernel.utils.CloseableIterator; import io.delta.kernel.utils.FileStatus; @@ -48,11 +52,13 @@ import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.stream.Stream; /** - * 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. + * Reads Query Table and Change Data Feed files from a Delta log. The parquet or delta response is + * NDJSON: protocol, metaData, then one file action per data file, or one add, remove, or cdc per + * data change file for a startingVersion or Change Data Feed 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 @@ -74,9 +80,15 @@ public class DeltaTableQueryReader { .add(DeltaAction.METADATA.colName, DeltaAction.METADATA.schema) .add(DeltaAction.ADD.colName, DeltaAction.ADD.schema) .add(DeltaAction.REMOVE.colName, DeltaAction.REMOVE.schema) + .add(DeltaAction.CDC.colName, DeltaAction.CDC.schema) .add( DeltaAction.COMMITINFO.colName, - new StructType().add("inCommitTimestamp", LongType.LONG)); + new StructType() + .add("inCommitTimestamp", LongType.LONG) + .add("operation", StringType.STRING) + .add( + "operationMetrics", + new MapType(StringType.STRING, StringType.STRING, true))); private final DeltaKernel kernel; private final UrlSigners signers; @@ -191,7 +203,7 @@ public Result readChanges( session, table, (SnapshotImpl) DeltaSnapshots.open(session, startingVersion, null), - new ChangeRange( + ChangeRange.dataChanges( startingVersion, endingVersion == null ? latest : endingVersion, includeHistoricalProtocol), @@ -199,8 +211,72 @@ public Result readChanges( } /** - * Protocol and metaData from {@code header}, then each commit in {@code range} with its add and - * remove files that have dataChange. + * Reads the Change Data Feed from the start through the end version or timestamp, or the latest + * version. Versions take precedence over timestamps, and an end after the latest version is + * capped, as in the OSS server. Protocol and metaData lines come from the start version with + * {@code includeHistoricalMetadata}, which also returns later metaData changes, and from the + * latest version otherwise. + */ + public Result readChangeDataFeed( + ResolvedAsset table, + AuthContext auth, + ResponseOptions options, + Long startingVersion, + Instant startingTimestamp, + Long endingVersion, + Instant endingTimestamp, + boolean includeHistoricalMetadata, + boolean includeHistoricalProtocol) { + DeltaKernel.Session session = kernel.open(table, auth, URL_TTL); + SnapshotImpl latest = (SnapshotImpl) DeltaSnapshots.open(session, null, null); + TableImpl delta = (TableImpl) session.table(); + long start; + Long end; + try { + if (startingVersion != null) { + start = startingVersion; + } else if (startingTimestamp != null) { + start = + delta.getVersionAtOrAfterTimestamp(session.engine(), startingTimestamp.toEpochMilli()); + } else { + throw ApiException.invalidParameter("startingVersion or startingTimestamp is required"); + } + end = + endingVersion != null + ? endingVersion + : endingTimestamp != null + ? delta.getVersionBeforeOrAtTimestamp( + session.engine(), endingTimestamp.toEpochMilli()) + : null; + } catch (KernelException | IllegalArgumentException invalid) { + throw ApiException.invalidParameter(invalid.getMessage()); + } + if (start < 0) { + throw ApiException.invalidParameter("startingVersion must be non-negative"); + } + requireAtMostLatest("startingVersion", start, latest.getVersion()); + if (end != null && end < start) { + throw ApiException.invalidParameter( + "endingVersion " + end + " is before startingVersion " + start); + } + long resolvedEnd = end == null ? latest.getVersion() : Math.min(end, latest.getVersion()); + SnapshotImpl first = (SnapshotImpl) DeltaSnapshots.open(session, start, null); + if (!changeDataFeedEnabled(first.getMetadata())) { + throw changeDataNotRecorded(start, start, resolvedEnd); + } + return changes( + session, + table, + includeHistoricalMetadata ? first : latest, + ChangeRange.changeDataFeed( + start, resolvedEnd, includeHistoricalMetadata, includeHistoricalProtocol), + options); + } + + /** + * Protocol and metaData from {@code header}, then each commit in {@code range}. A Change Data + * Feed commit with cdc files returns only those, and a no-op MERGE returns no files; otherwise + * commits return their add and remove files with dataChange. */ private Result changes( DeltaKernel.Session session, @@ -279,16 +355,59 @@ private static void write(ResponseWriter out, ChangeRange range, Commit commit) } // A commit has metaData only when it creates the table or changes its schema or properties. for (Metadata metadata : commit.metadata) { - if (later) { + if (range.changeDataFeed() && !changeDataFeedEnabled(metadata)) { + throw changeDataNotRecorded(version, range.start(), range.end()); + } + if (range.includeHistoricalMetadata() && 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) { + List files = commit.files; + if (range.changeDataFeed() && !commit.cdcFiles.isEmpty()) { + files = commit.cdcFiles; + } else if (range.changeDataFeed() && noOpMerge(commit.commitInfo)) { + files = List.of(); + } + for (ChangeFile file : files) { out.changeFile(file, version, commit.timestamp); } } + private static boolean changeDataFeedEnabled(Metadata metadata) { + return TableConfig.CHANGE_DATA_FEED_ENABLED.fromMetadata(metadata); + } + + private static ApiException changeDataNotRecorded(long version, long start, long end) { + return ApiException.invalidParameter( + "Error getting change data for range [" + + start + + ", " + + end + + "] as change data was not recorded for version [" + + version + + "]"); + } + + // MERGE can rewrite files with dataChange but change no rows; its metrics say so, and no cdc + // is written. Missing metrics do not skip. + private static boolean noOpMerge(Row commitInfo) { + if (commitInfo == null) { + return false; + } + StructType schema = commitInfo.getSchema(); + int operation = schema.indexOf("operation"); + int metrics = schema.indexOf("operationMetrics"); + if (commitInfo.isNullAt(operation) + || !"MERGE".equals(commitInfo.getString(operation)) + || commitInfo.isNullAt(metrics)) { + return false; + } + Map values = VectorUtils.toJavaMap(commitInfo.getMap(metrics)); + return Stream.of("numTargetRowsInserted", "numTargetRowsUpdated", "numTargetRowsDeleted") + .allMatch(metric -> "0".equals(values.get(metric))); + } + private static void requireAtMostLatest(String name, long version, long latest) { if (version > latest) { throw ApiException.invalidParameter( @@ -508,15 +627,37 @@ private record SignedFile( long expirationTimestamp) {} /** - * The versions and options of a startingVersion query, which also returns later metaData - * changes. + * The versions and options of a change query. * + * @param changeDataFeed whether this is a Change Data Feed query, which prefers cdc files + * @param includeHistoricalMetadata return metaData changes after {@code start} * @param includeHistoricalProtocol return protocol changes after {@code start} in delta responses */ - private record ChangeRange(long start, long end, boolean includeHistoricalProtocol) {} + private record ChangeRange( + long start, + long end, + boolean changeDataFeed, + boolean includeHistoricalMetadata, + boolean includeHistoricalProtocol) { + + // startingVersion queries always return later metaData changes. + static ChangeRange dataChanges(long start, long end, boolean includeHistoricalProtocol) { + return new ChangeRange(start, end, false, true, includeHistoricalProtocol); + } + + static ChangeRange changeDataFeed( + long start, + long end, + boolean includeHistoricalMetadata, + boolean includeHistoricalProtocol) { + return new ChangeRange( + start, end, true, includeHistoricalMetadata, includeHistoricalProtocol); + } + } /** Ordinals of a commit file batch. */ - private record ChangeColumns(int protocol, int metadata, int add, int remove, int commitInfo) { + private record ChangeColumns( + int protocol, int metadata, int add, int remove, int cdc, int commitInfo) { static ChangeColumns of(StructType schema) { return new ChangeColumns( @@ -524,17 +665,23 @@ static ChangeColumns of(StructType schema) { schema.indexOf(DeltaAction.METADATA.colName), schema.indexOf(DeltaAction.ADD.colName), schema.indexOf(DeltaAction.REMOVE.colName), + schema.indexOf(DeltaAction.CDC.colName), schema.indexOf(DeltaAction.COMMITINFO.colName)); } } - /** The actions of one commit, buffered so its protocol and metaData go out before its files. */ + /** + * The actions of one commit, buffered so its protocol and metaData go out before its files and + * its cdc files can take precedence over add and remove. + */ 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<>(); + final List cdcFiles = new ArrayList<>(); + Row commitInfo; /** * The commit file's {@code modificationTime} is the timestamp unless commitInfo has an @@ -556,8 +703,10 @@ void add(Row row, ChangeColumns columns) { 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.cdc())) { + cdcFiles.add(ChangeFile.of(ChangeType.CDC, row.getStruct(columns.cdc()))); } else if (!row.isNullAt(columns.commitInfo())) { - Row commitInfo = row.getStruct(columns.commitInfo()); + commitInfo = row.getStruct(columns.commitInfo()); int inCommitTimestamp = commitInfo.getSchema().indexOf("inCommitTimestamp"); if (!commitInfo.isNullAt(inCommitTimestamp)) { timestamp = commitInfo.getLong(inCommitTimestamp); @@ -566,7 +715,7 @@ void add(Row row, ChangeColumns columns) { } } - /** The fields a change line needs from an add or remove action in the log. */ + /** The fields a change line needs from an add, remove, or cdc action in the log. */ private record ChangeFile( ChangeType type, Row action, @@ -607,6 +756,19 @@ yield new ChangeFile( null, remove.getDeletionVector()); } + // cdc files are always data changes and carry no stats or deletion vector. + case CDC -> { + StructType schema = action.getSchema(); + yield new ChangeFile( + type, + action, + action.getString(schema.indexOf("path")), + true, + VectorUtils.toJavaMap(action.getMap(schema.indexOf("partitionValues"))), + action.getLong(schema.indexOf("size")), + null, + Optional.empty()); + } }; } } 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 d26c3ab3..9885cee7 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 @@ -21,6 +21,7 @@ import io.opensharing.asset.table.FileIdHash; import io.opensharing.asset.table.ProtocolTables; import io.opensharing.asset.table.QueryTableRequest; +import io.opensharing.asset.table.TableChangesRequest; import io.opensharing.asset.table.TemporaryTableCredentialsRequest; import io.opensharing.asset.table.TemporaryTableCredentialsResponse; import io.opensharing.asset.table.delta.DeltaTableMetadataReader; @@ -134,20 +135,65 @@ public HttpResponse queryTable( 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), - DeltaSharingCapabilities.includeEndStreamAction(capabilities)); - DeltaTableQueryReader.Result result = - tables.query(RecipientTokenDecorator.recipient(ctx), share, schema, table, query, options); + DeltaTableQueryReader.ResponseOptions options = responseOptions(capabilities, fileIdHash); + return files( + tables.query(RecipientTokenDecorator.recipient(ctx), share, schema, table, query, options), + options); + } + + /** + * {@code GET /shares/{share}/schemas/{schema}/tables/{table}/changes}: the table's Change Data + * Feed over a version or timestamp range as NDJSON with presigned file URLs, and the starting + * version in the {@code Delta-Table-Version} header. + */ + @Get("/shares/{share}/schemas/{schema}/tables/{table}/changes") + @Produces("application/x-ndjson") + public HttpResponse queryTableChanges( + ServiceRequestContext ctx, + @Param String share, + @Param String schema, + @Param String table, + @Param @Nullable Long startingVersion, + @Param @Nullable String startingTimestamp, + @Param @Nullable Long endingVersion, + @Param @Nullable String endingTimestamp, + @Param @Nullable Boolean includeHistoricalMetadata, + @Param @Nullable Boolean includeHistoricalProtocol, + @Header(DeltaSharingCapabilities.HEADER) @Nullable String capabilities, + @Header(FileIdHash.HEADER) @Nullable String fileIdHash) { + TableChangesRequest request = + new TableChangesRequest( + startingVersion, + startingTimestamp, + endingVersion, + endingTimestamp, + includeHistoricalMetadata, + includeHistoricalProtocol); + DeltaTableQueryReader.ResponseOptions options = responseOptions(capabilities, fileIdHash); + return files( + tables.changes( + RecipientTokenDecorator.recipient(ctx), share, schema, table, request, options), + options); + } + + private static DeltaTableQueryReader.ResponseOptions responseOptions( + String capabilities, String fileIdHash) { + return new DeltaTableQueryReader.ResponseOptions( + capabilities, + FileIdHash.parse(fileIdHash), + DeltaSharingCapabilities.includeEndStreamAction(capabilities)); + } + + private static HttpResponse files( + DeltaTableQueryReader.Result result, DeltaTableQueryReader.ResponseOptions options) { ResponseHeadersBuilder headers = ResponseHeaders.builder(HttpStatus.OK) .contentType(NDJSON) .add(DELTA_TABLE_VERSION, Long.toString(result.version())) .add( DeltaSharingCapabilities.HEADER, - DeltaSharingCapabilities.responded(capabilities, options.includeEndStreamAction())); + DeltaSharingCapabilities.responded( + options.capabilities(), options.includeEndStreamAction())); if (options.fileIdHash() != null) { headers.add(FileIdHash.HEADER, options.fileIdHash()); } 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 8341e0a6..89e79582 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 @@ -24,6 +24,7 @@ import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.attribute.FileTime; import java.time.Duration; import java.time.Instant; import java.util.Arrays; @@ -175,6 +176,114 @@ void rejectsVersionsAfterTheLatestVersion() throws IOException { assertThrows(ApiException.class, () -> changes(1L, 5L, null, false)); } + @Test + void changeDataFeedPrefersCdcFilesAndSkipsNoOpMerges() throws IOException { + writeChangeDataFeed(); + List lines = changeDataFeed(0L, null, 3L, null, null, false); + // Without includeHistoricalMetadata, protocol and metaData come from the latest version. + assertTrue(lines.get(1).contains("\"version\":4"), lines.get(1)); + assertEquals( + List.of("add a 0", "add b 0", "cdf c 1", "remove b 2"), + describe(lines.subList(2, lines.size()))); + assertTrue(lines.get(4).startsWith("{\"cdf\":"), lines.get(4)); + + List delta = changeDataFeed(1L, null, 2L, null, "responseformat=delta", false); + assertEquals(List.of("cdc c 1", "remove b 2"), describe(delta.subList(2, delta.size()))); + } + + @Test + void changeDataFeedResolvesTimestampsAndHistoricalMetadata() throws IOException { + writeChangeDataFeed(); + List lines = + changeDataFeed( + null, + Instant.ofEpochMilli(1500), + null, + Instant.ofEpochMilli(3500), + null, + true); + assertTrue(lines.get(1).contains("\"version\":1"), lines.get(1)); + assertEquals(List.of("cdf c 1", "remove b 2"), describe(lines.subList(2, lines.size()))); + // An end after the latest version is capped; the range then reaches CDF being disabled. + assertThrows(ApiException.class, () -> changeDataFeed(0L, null, 9L, null, null, false)); + } + + @Test + void changeDataFeedRejectsInvalidRanges() throws IOException { + // The table written in setup never enabled the Change Data Feed. + assertThrows(ApiException.class, () -> changeDataFeed(0L, null, null, null, null, false)); + writeChangeDataFeed(); + assertThrows(ApiException.class, () -> changeDataFeed(null, null, null, null, null, false)); + assertThrows(ApiException.class, () -> changeDataFeed(2L, null, 1L, null, null, false)); + assertThrows(ApiException.class, () -> changeDataFeed(5L, null, null, null, null, false)); + } + + private List changeDataFeed( + Long startingVersion, + Instant startingTimestamp, + Long endingVersion, + Instant endingTimestamp, + String capabilities, + boolean includeHistoricalMetadata) { + return List.of( + reader + .readChangeDataFeed( + table, + null, + new DeltaTableQueryReader.ResponseOptions(capabilities, null, false), + startingVersion, + startingTimestamp, + endingVersion, + endingTimestamp, + includeHistoricalMetadata, + false) + .ndjson() + .split("\n")); + } + + // 0: insert a, b. 1: update a to d with a cdc file. 2: delete b without cdc. 3: no-op MERGE. + // 4: disable the Change Data Feed. Commit n is written at (n + 1) seconds. + private void writeChangeDataFeed() throws IOException { + Path log = dir.resolve("_delta_log"); + String metadata = + "{\"metaData\":{\"id\":\"t\",\"format\":{\"provider\":\"parquet\",\"options\":{}}," + + "\"schemaString\":\"" + + SCHEMA + + "\",\"partitionColumns\":[\"date\"]," + + "\"configuration\":{\"delta.enableChangeDataFeed\":\"%s\"},\"createdTime\":0}}"; + List commits = + List.of( + String.join( + "\n", + "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":4}}", + metadata.formatted("true"), + add("a", "2021-01-01", 1, 10), + add("b", "2021-01-02", 11, 20)), + String.join( + "\n", + remove("a", "2021-01-01", true), + add("d", "2021-01-01", 1, 10), + "{\"cdc\":{\"path\":\"_change_data/date=2021-01-01/c.parquet\"," + + "\"partitionValues\":{\"date\":\"2021-01-01\"},\"size\":1," + + "\"dataChange\":false}}"), + remove("b", "2021-01-02", true), + String.join( + "\n", + "{\"commitInfo\":{\"timestamp\":4000,\"engineInfo\":\"test\"," + + "\"operation\":\"MERGE\",\"operationParameters\":{}," + + "\"isBlindAppend\":false,\"txnId\":\"t\"," + + "\"operationMetrics\":{\"numTargetRowsInserted\":\"0\"," + + "\"numTargetRowsUpdated\":\"0\",\"numTargetRowsDeleted\":\"0\"}}}", + remove("d", "2021-01-01", true), + add("e", "2021-01-01", 1, 10)), + metadata.formatted("false")); + for (int version = 0; version < commits.size(); version++) { + Path commit = log.resolve("%020d.json".formatted(version)); + Files.writeString(commit, commits.get(version) + "\n"); + Files.setLastModifiedTime(commit, FileTime.fromMillis((version + 1) * 1000L)); + } + } + private List changes( long startingVersion, Long endingVersion, String capabilities, boolean historicalProtocol) { return List.of( @@ -204,7 +313,7 @@ private static List describe(List lines) { String name = fileName(action.get(key).get("path")); return key + " " + name + " " + body.get("version"); } - if (key.equals("add") || key.equals("remove")) { + if (key.equals("add") || key.equals("remove") || key.equals("cdf")) { return key + " " + fileName(body.get("url")) + " " + body.get("version"); } return key + " " + body.get("version"); diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableChangesIntegrationTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableChangesIntegrationTest.java new file mode 100644 index 00000000..11acdf00 --- /dev/null +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableChangesIntegrationTest.java @@ -0,0 +1,128 @@ +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 Changes on real cloud Delta logs, with presigned file URLs. Individual cloud cases + * are skipped unless their repository credentials are present. + */ +@Timeout(60) +class ProtocolTableChangesIntegrationTest 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 queriesChangeDataFeedOfCdfTable() { + String bearer = shareTable("cdf-changes", "main.sales.cdf", "sales.cdf"); + String endpoint = endpoint("cdf-changes", "sales.cdf"); + + // Version 4 turns the Change Data Feed off and version 5 turns it back on. + String range = "?startingVersion=0&endingVersion=3"; + String parquet = ok(changes(endpoint + range, bearer, null), "0"); + assertTrue(parquet.contains("\"protocol\""), parquet); + assertTrue(parquet.contains("\"metaData\""), parquet); + long changes = count(parquet, "add") + count(parquet, "remove") + count(parquet, "cdf"); + assertTrue(changes > 0, parquet); + String delta = ok(changes(endpoint + range, bearer, "responseformat=delta"), "0"); + assertEquals(changes, count(delta, "file"), delta); + + // A range through version 4 is rejected. + Response off = changes(endpoint + "?startingVersion=0", bearer, null); + assertInvalid(off); + assertTrue(off.text().contains("not recorded for version [4]"), off.text()); + // An end after the latest version is capped rather than rejected. + ok(changes(endpoint + "?startingVersion=5&endingVersion=1000", bearer, null), "5"); + // A range that ends before it starts is rejected. + assertInvalid(changes(endpoint + "?startingVersion=1&endingVersion=0", bearer, null)); + } + + @Test + @EnabledIfEnvironmentVariable(named = "AWS_ACCESS_KEY_ID", matches = ".+") + @EnabledIfEnvironmentVariable(named = "AWS_SECRET_ACCESS_KEY", matches = ".+") + void rejectsChangeDataFeedOfTable1() { + rejectsChangeDataFeed("table1-cdf", "main.sales.table1", "sales.table1"); + } + + @Test + @EnabledIfEnvironmentVariable(named = "AZURE_TEST_ACCOUNT_KEY", matches = ".+") + void rejectsChangeDataFeedOfAzureTable() { + rejectsChangeDataFeed("azure-cdf", "main.sales.azure", "sales.azure"); + } + + @Test + @EnabledIfEnvironmentVariable(named = "GOOGLE_APPLICATION_CREDENTIALS", matches = ".+") + void rejectsChangeDataFeedOfGoogleTable() { + rejectsChangeDataFeed("gcs-cdf", "main.sales.gcs", "sales.gcs"); + } + + // Change Data Feed is not enabled on table1 or its cloud copies. + private static void rejectsChangeDataFeed(String share, String catalogName, String sharedAs) { + String bearer = shareTable(share, catalogName, sharedAs); + String endpoint = endpoint(share, sharedAs); + + Response unrecorded = changes(endpoint + "?startingVersion=0", bearer, null); + assertInvalid(unrecorded); + assertTrue(unrecorded.text().contains("change data was not recorded"), unrecorded.text()); + // A starting version or timestamp is required. + assertInvalid(changes(endpoint, bearer, null)); + } + + 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; + } + + /** The changes endpoint of {@code sharedAs}, which is {@code schema.table}. */ + private static String endpoint(String share, String sharedAs) { + String[] alias = sharedAs.split("\\."); + return PROTOCOL + "/shares/" + share + "/schemas/" + alias[0] + "/tables/" + alias[1] + + "/changes"; + } + + private static Response changes(String path, String bearer, String capabilities) { + return capabilities == null + ? server.get(path, bearer) + : server.call( + HttpMethod.GET, path, bearer, null, "delta-sharing-capabilities", capabilities); + } + + /** Lines whose single key is {@code action}, such as add, remove, cdf, or file. */ + private static long count(String ndjson, String action) { + return Arrays.stream(ndjson.split("\n")) + .filter(line -> line.startsWith("{\"" + action + "\"")) + .count(); + } + + /** 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(); + } + + 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/ProtocolTableChangesTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableChangesTest.java new file mode 100644 index 00000000..6c1c1464 --- /dev/null +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTableChangesTest.java @@ -0,0 +1,44 @@ +package io.opensharing.protocol; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.opensharing.TestServer.Response; +import io.opensharing.http.ErrorCodes; +import java.util.List; +import org.junit.jupiter.api.Test; + +/** + * Query Table Changes requests on local Delta logs, which do not record a Change Data Feed. Feeds + * themselves are read in {@code DeltaTableQueryReaderTest} and on the cloud test tables. + */ +class ProtocolTableChangesTest extends LocalDeltaTableSupport { + + @Test + void validatesChangeDataFeedRequests() { + createShare("table-cdf"); + addObject("table-cdf", "TABLE", "main.ops.events", "ops.events"); + createShare("hidden-cdf"); + addObject("hidden-cdf", "TABLE", "main.sales.customers", "sales.customers"); + String bearer = createAndActivateRecipient("table-cdf-partner"); + grant("table-cdf", "table-cdf-partner"); + String changes = PROTOCOL + "/shares/TABLE-CDF/schemas/OPS/tables/EVENTS/changes"; + + // A table that does not record its Change Data Feed is INVALID_PARAMETER_VALUE. + Response unrecorded = server.get(changes + "?startingVersion=0", bearer); + assertEquals(400, unrecorded.status(), unrecorded.text()); + assertEquals(ErrorCodes.INVALID_PARAMETER_VALUE, unrecorded.errorCode()); + assertTrue(unrecorded.text().contains("change data was not recorded"), unrecorded.text()); + + // A start is required, and must be a version or an ISO-8601 timestamp. + for (String invalid : List.of("", "?startingTimestamp=yesterday", "?startingVersion=one")) { + assertEquals(400, server.get(changes + invalid, bearer).status(), invalid); + } + + // A table in an ungranted share is 404, not 403. + String hiddenChanges = PROTOCOL + "/shares/hidden-cdf/schemas/sales/tables/customers/changes"; + Response hidden = server.get(hiddenChanges + "?startingVersion=0", bearer); + assertEquals(404, hidden.status()); + assertEquals(ErrorCodes.RESOURCE_DOES_NOT_EXIST, hidden.errorCode()); + } +} diff --git a/server/opensharing-server/src/test/resources/local-catalog-cloud.yml b/server/opensharing-server/src/test/resources/local-catalog-cloud.yml index e93b2205..980f1565 100644 --- a/server/opensharing-server/src/test/resources/local-catalog-cloud.yml +++ b/server/opensharing-server/src/test/resources/local-catalog-cloud.yml @@ -35,6 +35,13 @@ assets: storageLocation: s3://delta-exchange-test/delta-exchange-test/deletion_vectors_with_dvs_dv_property_on format: delta + - identifier: main.sales.cdf + type: TABLE + attributes: + subtype: MANAGED + storageLocation: s3://delta-exchange-test/delta-exchange-test/cdf_table_cdf_enabled/ + format: delta + - identifier: main.sales.azure type: TABLE attributes: