Delta Sharing protocol: Query Table predicate and limit pushdown

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,20 +207,27 @@ 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),
         options,
         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<SharedDataObjectEntity> inSchema(ShareEntity share, String schema) {

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

@@ -7,7 +7,8 @@
  * implemented.
  */
 public record QueryTableRequest(
-    List<String> predicateHints,
+    // Deprecated by the protocol in favor of jsonPredicateHints; only logged, never used to filter.
+    @Deprecated List<String> predicateHints,
     String jsonPredicateHints,
     Integer limitHint,
     Long version,

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 @@
  * <p>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.
+ *
+ * <p>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<FilteredColumnarBatch> 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<Predicate> 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<FilteredColumnarBatch> 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<Row> 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. */