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 5f97da6d..829436c1 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 @@ -31,6 +31,8 @@ import java.util.Comparator; import java.util.List; import java.util.function.Supplier; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * The tables a recipient reads through the sharing protocol in a share it holds SELECT on: the @@ -42,6 +44,7 @@ */ public class ProtocolTables { + private static final Logger log = LoggerFactory.getLogger(ProtocolTables.class); private static final int CATALOG_PAGE_SIZE = 1000; private final CatalogConnector catalog; @@ -173,7 +176,8 @@ public TemporaryTableCredentialsResponse temporaryCredentials( * 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. A latest-snapshot query can ask for a refresh token, and a later query can pass it back - * to re-sign the same version. Pushdown hints and startingVersion queries are not implemented. + * to re-sign the same version. jsonPredicateHints and limitHint narrow the files returned. + * startingVersion queries are not implemented. */ public DeltaTableQueryReader.Result query( RecipientPrincipal recipient, @@ -193,9 +197,6 @@ public DeltaTableQueryReader.Result query( 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"); - } boolean historical = body.version() != null || body.timestamp() != null; boolean includeRefreshToken = Boolean.TRUE.equals(body.includeRefreshToken()); if (includeRefreshToken && historical) { @@ -206,6 +207,9 @@ public DeltaTableQueryReader.Result query( throw ApiException.invalidParameter( "refreshToken cannot be used when querying a specific version."); } + if (body.predicateHints() != null && !body.predicateHints().isEmpty()) { + log.debug("Ignoring deprecated predicateHints {}", body.predicateHints()); + } return queryReader.read( resolved, owner(granted), @@ -213,13 +217,17 @@ public DeltaTableQueryReader.Result query( body.version(), parseTimestamp("timestamp", body.timestamp()), body.refreshToken(), - includeRefreshToken); + includeRefreshToken, + body.jsonPredicateHints(), + limitHint(body)); } - private static boolean hasPushdownHint(QueryTableRequest body) { - return (body.predicateHints() != null && !body.predicateHints().isEmpty()) - || (body.jsonPredicateHints() != null && !body.jsonPredicateHints().isBlank()) - || body.limitHint() != null; + // limitHint counts unfiltered file records, so it is dropped when any predicate hint is present. + private static Long limitHint(QueryTableRequest body) { + boolean hasPredicate = + (body.predicateHints() != null && !body.predicateHints().isEmpty()) + || (body.jsonPredicateHints() != null && !body.jsonPredicateHints().isBlank()); + return hasPredicate || body.limitHint() == null ? null : body.limitHint().longValue(); } private List inSchema(ShareEntity share, String 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 index 034b3190..05c65a86 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 @@ -7,7 +7,8 @@ * implemented. */ public record QueryTableRequest( - List predicateHints, + // Deprecated by the protocol in favor of jsonPredicateHints; only logged, never used to filter. + @Deprecated List predicateHints, String jsonPredicateHints, Integer limitHint, Long version, 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 872c2408..d17d86b6 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 @@ -1,10 +1,14 @@ package io.opensharing.asset.table.delta; +import com.fasterxml.jackson.databind.ObjectMapper; import io.delta.kernel.Scan; +import io.delta.kernel.ScanBuilder; import io.delta.kernel.Snapshot; import io.delta.kernel.data.FilteredColumnarBatch; import io.delta.kernel.data.Row; +import io.delta.kernel.expressions.Predicate; import io.delta.kernel.internal.InternalScanFileUtils; +import io.delta.kernel.internal.ScanImpl; import io.delta.kernel.internal.SnapshotImpl; import io.delta.kernel.internal.actions.AddFile; import io.delta.kernel.internal.actions.DeletionVectorDescriptor; @@ -37,10 +41,14 @@ *

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. */ public class DeltaTableQueryReader { private static final Duration URL_TTL = Duration.ofHours(1); + private static final ObjectMapper JSON = new ObjectMapper(); private final DeltaKernel kernel; private final UrlSigners signers; @@ -58,6 +66,8 @@ public DeltaTableQueryReader(DeltaKernel kernel, UrlSigners signers) { * @param timestamp read the latest version committed at or before this time * @param refreshToken a refresh token from an earlier response; reads the version it pins * @param includeRefreshToken put a refresh token for this snapshot in the endStreamAction + * @param jsonPredicateHints the request's hint; an unusable hint returns every file + * @param limitHint stop once the returned files hold at least this many records; null for all */ public Result read( ResolvedAsset table, @@ -66,7 +76,9 @@ public Result read( Long version, Instant timestamp, String refreshToken, - boolean includeRefreshToken) { + boolean includeRefreshToken, + String jsonPredicateHints, + Long limitHint) { if (version != null && timestamp != null) { throw ApiException.invalidParameter("version and timestamp are mutually exclusive"); } @@ -88,14 +100,27 @@ public Result read( 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())) { + // Live records in the files returned so far, compared against limitHint. + long numRecords = 0; + // Scan files are the add actions that are live in the snapshot after log replay. With a + // filter, Kernel also prunes partitions and skips files whose min/max stats cannot match. + ScanBuilder builder = snapshot.getScanBuilder(); + Optional filter = + JsonPredicateHints.toPredicate(jsonPredicateHints, snapshot.getSchema()); + Scan scan = filter.map(builder::withFilter).orElse(builder).build(); + // Stats are only read with includeStats; they feed file stats and limitHint record counts. + try (CloseableIterator batches = + ((ScanImpl) scan).getScanFiles(session.engine(), true)) { + files: 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); + // Checked before each file, so the last file returned may go past the limit. + if (limitHint != null && numRecords >= limitHint) { + break files; + } + numRecords += out.snapshotFile(rows.next(), fileVersion, fileTimestamp); } } } @@ -114,6 +139,18 @@ public Result read( return new Result(snapshot.getVersion(), out.ndjson()); } + // 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()) { + return 0; + } + try { + return JSON.readTree(stats).path("numRecords").asLong(0); + } catch (IOException unreadable) { + return 0; + } + } + /** * 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. @@ -181,8 +218,11 @@ void metadata(Metadata metadata, Long version, CRCInfo crc) { format)); } - /** Appends the file action for a snapshot scan file. */ - void snapshotFile(Row scanFile, Long version, Long timestamp) { + /** + * Appends the file action for a snapshot scan file and returns its live record count, which + * excludes rows a deletion vector marks deleted. + */ + long 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()); @@ -211,6 +251,8 @@ void snapshotFile(Row scanFile, Long version, Long timestamp) { timestamp, signed.expirationTimestamp())); } + return numRecords(stats) + - add.getDeletionVector().map(DeletionVectorDescriptor::getCardinality).orElse(0L); } /** {@code refreshToken} may be null. */ diff --git a/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/JsonPredicateHints.java b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/JsonPredicateHints.java new file mode 100644 index 00000000..8dc30c08 --- /dev/null +++ b/server/opensharing-server-core/src/main/java/io/opensharing/asset/table/delta/JsonPredicateHints.java @@ -0,0 +1,243 @@ +package io.opensharing.asset.table.delta; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.delta.kernel.expressions.AlwaysFalse; +import io.delta.kernel.expressions.AlwaysTrue; +import io.delta.kernel.expressions.And; +import io.delta.kernel.expressions.Column; +import io.delta.kernel.expressions.Expression; +import io.delta.kernel.expressions.Literal; +import io.delta.kernel.expressions.Or; +import io.delta.kernel.expressions.Predicate; +import io.delta.kernel.types.BooleanType; +import io.delta.kernel.types.DataType; +import io.delta.kernel.types.DateType; +import io.delta.kernel.types.DoubleType; +import io.delta.kernel.types.FloatType; +import io.delta.kernel.types.IntegerType; +import io.delta.kernel.types.LongType; +import io.delta.kernel.types.StringType; +import io.delta.kernel.types.StructType; +import io.delta.kernel.types.TimestampType; +import java.time.Instant; +import java.time.LocalDate; +import java.time.OffsetDateTime; +import java.time.format.DateTimeFormatter; +import java.time.temporal.ChronoUnit; +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; + +/** + * Converts Delta Sharing {@code jsonPredicateHints} into a Kernel scan filter. Hints are best + * effort: an oversized, malformed, or unsupported hint returns no filter, so every file is + * returned. + * + *

Kernel uses the filter to prune partitions and to skip files whose min/max stats cannot match. + * It never drops rows inside a file, so the client must still apply the predicate itself. Returning + * no filter is therefore always safe; it only costs extra files. + * + *

A hint is a tree of {@code {"op": ..., "children": [...]}} nodes. Leaves are {@code column} + * (top-level columns only, by {@code name}) and {@code literal} (a string {@code value}); both + * carry a {@code valueType}. Supported ops are {@code and}, {@code or}, {@code not}, {@code + * isNull}, {@code equal}, {@code lessThan}, {@code lessThanOrEqual}, {@code greaterThan} and + * {@code greaterThanOrEqual}. + */ +final class JsonPredicateHints { + + /** Largest hint accepted, in characters; larger hints are ignored rather than parsed. */ + static final int MAX_SIZE = 1024 * 1024; + + /** Deepest op tree accepted, which also bounds the recursion in {@link #convert}. */ + static final int MAX_DEPTH = 100; + + private static final ObjectMapper JSON = new ObjectMapper(); + + private JsonPredicateHints() {} + + /** + * Returns the Kernel filter for {@code json}, or empty when there is no usable hint. + * + * @param json the request's {@code jsonPredicateHints}; may be null or blank + * @param schema the snapshot schema; columns must exist in it with the hinted type + */ + static Optional toPredicate(String json, StructType schema) { + if (json == null || json.isBlank() || json.length() > MAX_SIZE) { + return Optional.empty(); + } + try { + JsonNode root = JSON.readTree(json); + if (depth(root) > MAX_DEPTH) { + return Optional.empty(); + } + return Optional.of(asPredicate(convert(root, schema), schema)); + } catch (Exception unsupported) { + // Bad JSON, unknown ops or types, schema mismatches and unparsable literals all land here. + return Optional.empty(); + } + } + + /** Converts one op node. Throws on anything unsupported, so the whole hint is dropped. */ + private static Expression convert(JsonNode node, StructType schema) { + String op = text(node, "op"); + return switch (op) { + case "column", "literal" -> leaf(node, schema); + case "isNull" -> new Predicate("IS_NULL", leaf(children(node, 1).get(0), schema)); + case "not" -> + new Predicate("NOT", asPredicate(convert(children(node, 1).get(0), schema), schema)); + case "equal" -> comparison("=", node, schema); + case "lessThan" -> comparison("<", node, schema); + case "lessThanOrEqual" -> comparison("<=", node, schema); + case "greaterThan" -> comparison(">", node, schema); + case "greaterThanOrEqual" -> comparison(">=", node, schema); + case "and", "or" -> { + // Kernel's And and Or are binary, so fold two or more children from the left. + List children = children(node, -1); + Predicate result = asPredicate(convert(children.get(0), schema), schema); + for (JsonNode child : children.subList(1, children.size())) { + Predicate next = asPredicate(convert(child, schema), schema); + result = op.equals("and") ? new And(result, next) : new Or(result, next); + } + yield result; + } + default -> throw new IllegalArgumentException("unsupported op " + op); + }; + } + + /** A binary comparison. Both sides must declare the same valueType; nothing is coerced. */ + private static Predicate comparison(String name, JsonNode node, StructType schema) { + List children = children(node, 2); + if (!text(children.get(0), "valueType").equals(text(children.get(1), "valueType"))) { + throw new IllegalArgumentException("type mismatch in " + node); + } + return new Predicate(name, leaf(children.get(0), schema), leaf(children.get(1), schema)); + } + + /** A column or literal leaf, typed by its valueType. */ + private static Expression leaf(JsonNode node, StructType schema) { + String op = text(node, "op"); + DataType type = type(text(node, "valueType")); + if (op.equals("column")) { + // The column must exist with the hinted type; otherwise drop the hint, not prune wrongly. + String name = text(node, "name"); + if (!schema.fieldNames().contains(name) || !schema.get(name).getDataType().equals(type)) { + throw new IllegalArgumentException("column " + name + " does not match the schema"); + } + return new Column(name); + } + if (op.equals("literal")) { + return literal(text(node, "value"), type); + } + throw new IllegalArgumentException("expected a column or literal but found " + op); + } + + /** + * Parses a literal's string value as {@code type}. Dates are ISO {@code yyyy-MM-dd}; timestamps + * are ISO-8601 with an offset and become microseconds since the epoch, as Kernel stores them. + */ + private static Literal literal(String value, DataType type) { + if (type instanceof BooleanType) { + if (!value.equalsIgnoreCase("true") && !value.equalsIgnoreCase("false")) { + throw new IllegalArgumentException("invalid bool " + value); + } + return Literal.ofBoolean(Boolean.parseBoolean(value)); + } + if (type instanceof IntegerType) { + return Literal.ofInt(Integer.parseInt(value)); + } + if (type instanceof LongType) { + return Literal.ofLong(Long.parseLong(value)); + } + if (type instanceof StringType) { + return Literal.ofString(value); + } + if (type instanceof DateType) { + // Kernel dates are days since the epoch. + return Literal.ofDate(Math.toIntExact(LocalDate.parse(value).toEpochDay())); + } + if (type instanceof FloatType) { + return Literal.ofFloat(Float.parseFloat(value)); + } + if (type instanceof DoubleType) { + return Literal.ofDouble(Double.parseDouble(value)); + } + // type() only yields the types above or TIMESTAMP, so this is a timestamp. + Instant instant = + OffsetDateTime.parse(value, DateTimeFormatter.ISO_OFFSET_DATE_TIME).toInstant(); + return Literal.ofTimestamp(ChronoUnit.MICROS.between(Instant.EPOCH, instant)); + } + + /** + * Kernel only filters on predicates, but a hint may put a bare bool column or literal where a + * condition goes: at the root or as a child of {@code and}, {@code or} or {@code not}. A bool + * column {@code c} becomes {@code c = true}, which also drops rows where {@code c} is null. A + * bool literal becomes {@code AlwaysTrue} or {@code AlwaysFalse}. Any other non-predicate fails, + * which drops the hint. + */ + private static Predicate asPredicate(Expression expression, StructType schema) { + if (expression instanceof Predicate predicate) { + return predicate; + } + if (expression instanceof Column column + && schema.get(column.getNames()[0]).getDataType() instanceof BooleanType) { + return new Predicate("=", column, Literal.ofBoolean(true)); + } + if (expression instanceof Literal literal && literal.getDataType() instanceof BooleanType) { + return Boolean.TRUE.equals(literal.getValue()) + ? AlwaysTrue.ALWAYS_TRUE + : AlwaysFalse.ALWAYS_FALSE; + } + throw new IllegalArgumentException("not a boolean expression: " + expression); + } + + /** Maps a protocol valueType to its Kernel type; any other valueType drops the hint. */ + private static DataType type(String valueType) { + return switch (valueType) { + case "bool" -> BooleanType.BOOLEAN; + case "int" -> IntegerType.INTEGER; + case "long" -> LongType.LONG; + case "string" -> StringType.STRING; + case "date" -> DateType.DATE; + case "float" -> FloatType.FLOAT; + case "double" -> DoubleType.DOUBLE; + case "timestamp" -> TimestampType.TIMESTAMP; + default -> throw new IllegalArgumentException("unsupported valueType " + valueType); + }; + } + + /** The node's children: exactly {@code expected} of them, or at least two when it is negative. */ + private static List children(JsonNode node, int expected) { + JsonNode children = node.get("children"); + if (children == null || !children.isArray()) { + throw new IllegalArgumentException("missing children in " + node); + } + List list = new ArrayList<>(); + children.forEach(list::add); + if (expected < 0 ? list.size() < 2 : list.size() != expected) { + throw new IllegalArgumentException("wrong number of children in " + node); + } + return list; + } + + /** A required string field; numbers and booleans are rejected, as the protocol sends strings. */ + private static String text(JsonNode node, String field) { + JsonNode value = node.get(field); + if (value == null || !value.isTextual()) { + throw new IllegalArgumentException("missing " + field + " in " + node); + } + return value.asText(); + } + + /** Depth of the op tree, counting a leaf as 1; checked before {@link #convert} recurses. */ + private static int depth(JsonNode node) { + int deepest = 0; + JsonNode children = node.get("children"); + if (children != null) { + for (JsonNode child : children) { + deepest = Math.max(deepest, depth(child)); + } + } + return deepest + 1; + } +} 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 new file mode 100644 index 00000000..6ad94586 --- /dev/null +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/delta/DeltaTableQueryReaderTest.java @@ -0,0 +1,162 @@ +package io.opensharing.asset.table.delta; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.delta.kernel.Table; +import io.delta.kernel.defaults.engine.DefaultEngine; +import io.delta.kernel.engine.Engine; +import io.opensharing.asset.table.signer.SignedUrl; +import io.opensharing.asset.table.signer.UrlSigners; +import io.opensharing.auth.AuthContext; +import io.opensharing.catalog.AssetLocation; +import io.opensharing.catalog.AssetType; +import io.opensharing.catalog.CloudProvider; +import io.opensharing.catalog.DataSourceFormat; +import io.opensharing.catalog.ResolvedAsset; +import io.opensharing.catalog.StorageCredentials; +import io.opensharing.catalog.TableProperties; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.time.Instant; +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Query Table file selection on a local Delta log. Kernel opens the log directly and every path + * signs to {@code https://signed/}, so no catalog or cloud storage is involved. + */ +class DeltaTableQueryReaderTest { + + private static final String SCHEMA = + "{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[" + + "{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"long\\\",\\\"nullable\\\":true," + + "\\\"metadata\\\":{}}," + + "{\\\"name\\\":\\\"date\\\",\\\"type\\\":\\\"date\\\",\\\"nullable\\\":true," + + "\\\"metadata\\\":{}}]}"; + + @TempDir Path dir; + + private DeltaTableQueryReader reader; + private ResolvedAsset table; + + @BeforeEach + void writeTable() throws IOException { + Path log = Files.createDirectories(dir.resolve("_delta_log")); + Files.writeString( + log.resolve("00000000000000000000.json"), + String.join( + "\n", + "{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":2}}", + "{\"metaData\":{\"id\":\"t\",\"format\":{\"provider\":\"parquet\",\"options\":{}}," + + "\"schemaString\":\"" + + SCHEMA + + "\",\"partitionColumns\":[\"date\"],\"configuration\":{},\"createdTime\":0}}", + add("a", "2021-01-01", 1, 10), + add("b", "2021-01-02", 11, 20), + add("c", "2021-01-02", 21, 30)) + + "\n"); + + Engine engine = DefaultEngine.create(new Configuration()); + String location = dir.toUri().toString(); + StorageCredentials credentials = + new StorageCredentials(location, CloudProvider.AWS, Map.of(), Instant.MAX); + DeltaKernel.Session session = + new DeltaKernel.Session( + engine, Table.forPath(engine, dir.toString()), location, credentials); + DeltaKernel kernel = + new DeltaKernel(null) { + @Override + public Session open(ResolvedAsset table, AuthContext auth, Duration ttl) { + return session; + } + }; + UrlSigners signers = + new UrlSigners() { + @Override + public SignedUrl sign(String path, StorageCredentials credentials, Duration ttl) { + return new SignedUrl( + "https://signed/" + Path.of(path).getFileName(), + Instant.ofEpochMilli(4102444800000L)); + } + }; + reader = new DeltaTableQueryReader(kernel, signers); + table = + ResolvedAsset.builder(AssetType.TABLE, "main.sales.events") + .location(new AssetLocation(location, null)) + .additionalProperties(new TableProperties(DataSourceFormat.DELTA, Map.of())) + .build(); + } + + @Test + void jsonPredicateHintsPrunePartitionsAndSkipFilesByStats() { + assertEquals(List.of("a", "b", "c"), files(null, null)); + // date = 2021-01-02 prunes the 2021-01-01 partition. + assertEquals( + List.of("b", "c"), + files( + "{\"op\":\"equal\",\"children\":[{\"op\":\"column\",\"name\":\"date\"," + + "\"valueType\":\"date\"},{\"op\":\"literal\",\"value\":\"2021-01-02\"," + + "\"valueType\":\"date\"}]}", + null)); + // id > 25 skips files whose maxValues.id is at most 25. + assertEquals( + List.of("c"), + files( + "{\"op\":\"greaterThan\",\"children\":[{\"op\":\"column\",\"name\":\"id\"," + + "\"valueType\":\"long\"},{\"op\":\"literal\",\"value\":\"25\"," + + "\"valueType\":\"long\"}]}", + null)); + // An unsupported hint returns every file. + assertEquals(List.of("a", "b", "c"), files("{\"op\":\"like\"}", null)); + } + + @Test + void limitHintStopsOnceFilesCoverTheLimit() { + // limitHint counts stats.numRecords, which file actions carry through. + assertTrue(ndjson(null, null).contains("\\\"numRecords\\\":10")); + assertEquals(0, files(null, 0L).size()); + assertEquals(1, files(null, 10L).size()); + assertEquals(2, files(null, 11L).size()); + assertEquals(3, files(null, 1000L).size()); + } + + private String ndjson(String jsonPredicateHints, Long limitHint) { + return reader + .read( + table, + null, + new DeltaTableQueryReader.ResponseOptions(null, null, false), + null, + null, + null, + false, + jsonPredicateHints, + limitHint) + .ndjson(); + } + + private List files(String jsonPredicateHints, Long limitHint) { + return Arrays.stream(ndjson(jsonPredicateHints, limitHint).split("\n")) + .filter(line -> line.startsWith("{\"file\"")) + .map(line -> line.replaceAll(".*https://signed/(\\w)\\.parquet.*", "$1")) + .sorted() + .toList(); + } + + private static String add(String name, String date, long minId, long maxId) { + return ("{\"add\":{\"path\":\"date=%s/%s.parquet\",\"partitionValues\":{\"date\":\"%s\"}," + + "\"size\":1,\"modificationTime\":0,\"dataChange\":true," + + "\"stats\":\"{\\\"numRecords\\\":10," + + "\\\"minValues\\\":{\\\"id\\\":%d},\\\"maxValues\\\":{\\\"id\\\":%d}," + + "\\\"nullCount\\\":{\\\"id\\\":0}}\"}}") + .formatted(date, name, date, minId, maxId); + } +} diff --git a/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/delta/JsonPredicateHintsTest.java b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/delta/JsonPredicateHintsTest.java new file mode 100644 index 00000000..f912cab6 --- /dev/null +++ b/server/opensharing-server-core/src/test/java/io/opensharing/asset/table/delta/JsonPredicateHintsTest.java @@ -0,0 +1,199 @@ +package io.opensharing.asset.table.delta; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import io.delta.kernel.expressions.Predicate; +import io.delta.kernel.types.BooleanType; +import io.delta.kernel.types.DateType; +import io.delta.kernel.types.DoubleType; +import io.delta.kernel.types.FloatType; +import io.delta.kernel.types.IntegerType; +import io.delta.kernel.types.LongType; +import io.delta.kernel.types.StringType; +import io.delta.kernel.types.StructType; +import io.delta.kernel.types.TimestampType; +import java.util.Optional; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +class JsonPredicateHintsTest { + + private static final StructType SCHEMA = + new StructType() + .add("active", BooleanType.BOOLEAN) + .add("age", IntegerType.INTEGER) + .add("id", LongType.LONG) + .add("name", StringType.STRING) + .add("hireDate", DateType.DATE) + .add("score", FloatType.FLOAT) + .add("salary", DoubleType.DOUBLE) + .add("updatedAt", TimestampType.TIMESTAMP); + + private static final String ID = column("id", "long"); + + @ParameterizedTest + @CsvSource({ + "equal, =", + "lessThan, <", + "lessThanOrEqual, <=", + "greaterThan, >", + "greaterThanOrEqual, >=" + }) + void convertsEachComparison(String op, String kernelOp) { + assertConverts("(column(`id`) " + kernelOp + " 10)", op(op, ID, literal("10", "long"))); + } + + @ParameterizedTest + @CsvSource({ + "active, bool, true, true", + "age, int, 42, 42", + "id, long, 9000000000, 9000000000", + "name, string, Alice, Alice", + // Days since the epoch. + "hireDate, date, 2021-04-29, 18746", + "score, float, 1.5, 1.5", + "salary, double, 2.25, 2.25", + // Microseconds since the epoch; the offset is applied. + "updatedAt, timestamp, 2021-04-29T01:00:00+01:00, 1619654400000000" + }) + void parsesEachLiteralType(String column, String type, String value, String kernelValue) { + assertConverts( + "(column(`" + column + "`) = " + kernelValue + ")", + op("equal", column(column, type), literal(value, type))); + } + + @Test + void convertsIsNullAndNot() { + assertConverts( + "NOT(IS_NULL(column(`name`)))", op("not", op("isNull", column("name", "string")))); + } + + @Test + void foldsAndFromTheLeft() { + assertConverts( + "(((column(`id`) >= 10) AND (column(`id`) < 20)) AND (column(`age`) = 30))", + op( + "and", + op("greaterThanOrEqual", ID, literal("10", "long")), + op("lessThan", ID, literal("20", "long")), + op("equal", column("age", "int"), literal("30", "int")))); + } + + @Test + void foldsOrFromTheLeft() { + assertConverts( + "(((column(`id`) = 1) OR (column(`id`) = 2)) OR (column(`id`) = 3))", + op( + "or", + op("equal", ID, literal("1", "long")), + op("equal", ID, literal("2", "long")), + op("equal", ID, literal("3", "long")))); + } + + @Test + void treatsABoolColumnAsColumnEqualsTrue() { + assertConverts("(column(`active`) = true)", column("active", "bool")); + assertConverts("NOT((column(`active`) = true))", op("not", column("active", "bool"))); + assertConverts( + "((column(`active`) = true) AND (column(`id`) = 1))", + op("and", column("active", "bool"), op("equal", ID, literal("1", "long")))); + } + + @Test + void treatsABoolLiteralAsAlwaysTrueOrAlwaysFalse() { + assertConverts("ALWAYS_TRUE()", literal("true", "bool")); + assertConverts("ALWAYS_FALSE()", literal("FALSE", "bool")); + } + + @Test + void skipsEmptyOrOversizedInput() { + assertSkipped(null); + assertSkipped(" "); + assertSkipped("not json"); + // Valid JSON, but past the size limit. + assertSkipped(literal("true", "bool") + " ".repeat(JsonPredicateHints.MAX_SIZE)); + } + + @Test + void skipsUnsupportedOpsAndTypes() { + assertSkipped(op("like", ID, ID)); + assertSkipped(op("equal", column("id", "binary"), literal("1", "binary"))); + // op must be a string. + assertSkipped("{\"op\":1}"); + // A column of a supported type, but without op. + assertSkipped("{\"name\":\"id\",\"valueType\":\"long\"}"); + } + + @Test + void skipsColumnsThatDoNotMatchTheSchema() { + assertSkipped(op("equal", column("missing", "long"), literal("1", "long"))); + // id is a long in the schema. + assertSkipped(op("equal", column("id", "int"), literal("1", "int"))); + } + + @Test + void skipsComparisonsWhoseSidesDiffer() { + assertSkipped(op("equal", ID, literal("1", "string"))); + assertSkipped(op("lessThan", ID, op("isNull", ID))); + } + + @Test + void skipsUnparsableLiterals() { + assertSkipped(op("equal", ID, literal("x", "long"))); + assertSkipped(op("equal", column("active", "bool"), literal("yes", "bool"))); + assertSkipped(op("equal", column("hireDate", "date"), literal("04/29/2021", "date"))); + // Timestamps need an offset. + assertSkipped( + op("equal", column("updatedAt", "timestamp"), literal("2021-04-29T01:00:00", "timestamp"))); + } + + @Test + void skipsWrongChildCounts() { + assertSkipped("{\"op\":\"not\"}"); + assertSkipped(op("equal", ID)); + assertSkipped(op("isNull", ID, ID)); + assertSkipped(op("and", literal("true", "bool"))); + assertSkipped(op("or", literal("true", "bool"))); + } + + @Test + void skipsNonBooleanConditions() { + assertSkipped(ID); + assertSkipped(op("not", ID)); + assertSkipped(op("and", ID, literal("true", "bool"))); + } + + @Test + void skipsHintsOverTheDepthLimit() { + String deep = literal("true", "bool"); + for (int depth = 1; depth < JsonPredicateHints.MAX_DEPTH; depth++) { + deep = op("not", deep); + } + assertTrue(JsonPredicateHints.toPredicate(deep, SCHEMA).isPresent()); + assertSkipped(op("not", deep)); + } + + private static String column(String name, String valueType) { + return "{\"op\":\"column\",\"name\":\"" + name + "\",\"valueType\":\"" + valueType + "\"}"; + } + + private static String literal(String value, String valueType) { + return "{\"op\":\"literal\",\"value\":\"" + value + "\",\"valueType\":\"" + valueType + "\"}"; + } + + private static String op(String op, String... children) { + return "{\"op\":\"" + op + "\",\"children\":[" + String.join(",", children) + "]}"; + } + + private static void assertConverts(String expected, String json) { + Optional predicate = JsonPredicateHints.toPredicate(json, SCHEMA); + assertTrue(predicate.isPresent(), json); + assertEquals(expected, predicate.get().toString(), json); + } + + private static void assertSkipped(String json) { + assertEquals(Optional.empty(), JsonPredicateHints.toPredicate(json, SCHEMA), json); + } +} diff --git a/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTablePushdownIntegrationTest.java b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTablePushdownIntegrationTest.java new file mode 100644 index 00000000..d47f7ac3 --- /dev/null +++ b/server/opensharing-server/src/test/java/io/opensharing/protocol/ProtocolTablePushdownIntegrationTest.java @@ -0,0 +1,105 @@ +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 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 predicate and limit hints on real cloud Delta logs. Individual cloud cases are + * skipped unless their repository credentials are present. + */ +@Timeout(60) +class ProtocolTablePushdownIntegrationTest 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 limitsTable1Files() { + String bearer = shareTable("table1-hints", "main.sales.table1", "sales.table1"); + String endpoint = endpoint("table1-hints", "sales.table1"); + + long all = files(endpoint, bearer, "{}", "2"); + assertTrue(all > 0, "table1 has no files"); + // The limit is checked before each file, so at least one file is returned. + long limited = files(endpoint, bearer, "{\"limitHint\":1}", "2"); + assertTrue(limited >= 1 && limited <= all, limited + " of " + all); + // A hint that does not parse, or names a missing column, returns every file. + assertEquals(all, files(endpoint, bearer, "{\"jsonPredicateHints\":\"not json\"}", "2")); + assertEquals(all, files(endpoint, bearer, equalHint("missing", "x"), "2")); + } + + @Test + @EnabledIfEnvironmentVariable(named = "AZURE_TEST_ACCOUNT_KEY", matches = ".+") + void filtersAzureTableFiles() { + filtersFiles("azure-hints", "main.sales.azure", "sales.azure"); + } + + @Test + @EnabledIfEnvironmentVariable(named = "GOOGLE_APPLICATION_CREDENTIALS", matches = ".+") + void filtersGoogleTableFiles() { + filtersFiles("gcs-hints", "main.sales.gcs", "sales.gcs"); + } + + // The cloud copies of table1 have one file, in partition c2=foo bar, holding one c1=foo bar row. + private static void filtersFiles(String share, String catalogName, String sharedAs) { + String bearer = shareTable(share, catalogName, sharedAs); + String endpoint = endpoint(share, sharedAs); + + assertEquals(1, files(endpoint, bearer, equalHint("c2", "foo bar"), "0")); + // A partition value no file has prunes the file. + assertEquals(0, files(endpoint, bearer, equalHint("c2", "baz"), "0")); + // c1 is not a partition column, so its min/max stats skip the file. + assertEquals(0, files(endpoint, bearer, equalHint("c1", "zzz"), "0")); + // A hint that does not parse returns every file. + assertEquals(1, files(endpoint, bearer, "{\"jsonPredicateHints\":\"not json\"}", "0")); + // The limit is checked before each file, so the first file is always returned. + assertEquals(1, files(endpoint, bearer, "{\"limitHint\":1}", "0")); + } + + 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 query 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] + + "/query"; + } + + private static long files(String endpoint, String bearer, String body, String version) { + Response response = server.post(endpoint, bearer, body); + assertEquals(200, response.status(), response.text()); + assertEquals(version, response.headers().get("Delta-Table-Version")); + return Arrays.stream(response.text().split("\n")) + .filter(line -> line.startsWith("{\"file\"")) + .count(); + } + + /** A request body whose jsonPredicateHints is {@code column = value} on string values. */ + private static String equalHint(String column, String value) { + String hint = + "{\"op\":\"equal\",\"children\":[" + + "{\"op\":\"column\",\"name\":\"" + column + "\",\"valueType\":\"string\"}," + + "{\"op\":\"literal\",\"value\":\"" + value + "\",\"valueType\":\"string\"}]}"; + return "{\"jsonPredicateHints\":\"" + hint.replace("\"", "\\\"") + "\"}"; + } +}