feat: Add opt-in OffsetPageInfo metadata for paginated GraphQL queries

documentation/docs/configuration-default.md

@@ -10,7 +10,7 @@ The following is the [default configuration file](https://raw.githubusercontent.
     "logger": "print",
     "compile-flink-plan": true,
     "extended-scalar-types": true,
-    "predicate-pushdown-rules": "LIMITED_TABLE_SOURCE_RULES",
+    "predicate-pushdown-rules": "LIMITED_RULES_NO_SOURCE",
     "cost-model": "DEFAULT",
     "explain": {
       "sql": false,
@@ -23,15 +23,17 @@ The following is the [default configuration file](https://raw.githubusercontent.
       "endpoints": "FULL",
       "add-prefix": true,
       "max-result-depth": 3,
-      "default-limit": 10
+      "default-limit": 10,
+      "paginated-results": false
     }
   },
   "engines": {
     "flink": {
       "config": {
         "execution.runtime-mode": "STREAMING",
         "security.delegation.tokens.enabled": false,
-        "state.backend.type": "rocksdb"
+        "state.backend.type": "rocksdb",
+        "table.exec.sink.require-on-conflict": false
       }
     },
     "duckdb": {

documentation/docs/configuration.md

@@ -162,7 +162,8 @@ Configuration options that control the compiler, such as where logging output is
       "endpoints": "FULL",         // endpoint generation strategy ("FULL", "GRAPHQL", "OPS_ONLY")
       "add-prefix": true,          // add an operation-type prefix to function names to ensure uniqueness
       "max-result-depth": 3,       // maximum depth of graph traversal when generating operations from a schema
-      "default-limit": 10          // default query result limit
+      "default-limit": 10,         // default query result limit
+      "paginated-results": false   // wrap generated query results in a page with pagination metadata
     }
   }
 }

documentation/docs/interface.md

@@ -67,6 +67,53 @@ You can customize the GraphQL schema by:
 The compiler raises errors when the provided GraphQL schema is not compatible with the object-relationship model.
 :::
 
+#### Pagination
+
+Every generated query endpoint takes `limit` and `offset` arguments to page through the result (`limit` defaults to the configured `default-limit`). By default the endpoint returns the rows directly and the client tracks the offsets itself.
+
+Set `paginated-results` to `true` in the [`api` compiler configuration](configuration.md#compiler-compiler) to get pagination metadata alongside the rows. The generated schema then wraps every multi-row query result in a page type:
+
+```graphql
+type PersonPage {
+  results: [Person!]
+  pagination: OffsetPageInfo
+}
+
+type OffsetPageInfo {
+  pageSize: Int!
+  currentPage: Int!
+  hasNextPage: Boolean!
+  hasPreviousPage: Boolean!
+  nextOffset: Int
+  prevOffset: Int
+  firstEventTime: DateTime
+  lastEventTime: DateTime
+}
+```
+
+A query then selects the rows and the metadata it needs:
+
+```graphql
+query GetPeople {
+  Person(limit: 10, offset: 20) {
+    results { name email }
+    pagination { currentPage hasNextPage nextOffset }
+  }
+}
+```
+
+`firstEventTime` and `lastEventTime` are the earliest and latest event time of the *entire* result set, not of the returned page. They are `null` when the query result has no event time (rowtime) column.
+
+If you provide your own GraphQL schema, pagination is opt-in per query: give the query a result type with exactly two fields — a list of the result type, and a field of type `OffsetPageInfo` — and declare `OffsetPageInfo` exactly as shown above. The field names of the wrapper type are up to you. The compiler validates that:
+* the `OffsetPageInfo` type is declared and matches the definition above
+* the paginated query declares both a `limit` and an `offset` argument
+* the query returns multiple rows (i.e. it is not restricted to a single row) and is a query, not a subscription
+
+The server computes only the metadata a request actually selects, so paginated queries cost no more than unpaginated ones unless you ask for more:
+* `pageSize`, `currentPage`, `hasPreviousPage`, and `prevOffset` are derived from the request arguments and cost nothing.
+* `hasNextPage` and `nextOffset` make the query fetch one extra row, which is discarded before the results are returned. Without a `limit` argument the page holds every remaining row and `hasNextPage` is `false`.
+* `firstEventTime` and `lastEventTime` run a second `MIN`/`MAX` query over the event time column. The compiler adds an index on that column for paginated queries.
+
 #### Authoritative Model
 
 DataSQRL uses the GraphQL schemas the authoritative model for all API protocols. It serves as the foundational model on which operations, endpoints, and access patterns are defined. This simplifies the conceptual model and server execution since any API operation maps to a GraphQL query which is executed by a centralized and optimized GraphQL engine.

sqrl-cli/src/main/java/com/datasqrl/compile/CompilationProcess.java

@@ -70,8 +70,6 @@ public Pair<PhysicalPlan, TestPlan> executeCompilation(Optional<Path> testsPath)
     var dagBuilder = planner.getDagBuilder();
     var dag = dagPlanner.optimize(dagBuilder.getDag());
     var physicalPlan = dagPlanner.assemble(dag, environment);
-    var rewriters = ServiceLoaderDiscovery.getAll(PhysicalPlanRewriter.class);
-    physicalPlan = physicalPlan.applyRewriting(rewriters, environment);
     var mutationDatabase = physicalPlan.getMutationDatabase();
     writeDeploymentArtifactsHook.run(dag, planner.getCompleteScript().toString(), mutationDatabase);
 
@@ -120,6 +118,9 @@ public Pair<PhysicalPlan, TestPlan> executeCompilation(Optional<Path> testsPath)
       }
     }
 
+    var rewriters = ServiceLoaderDiscovery.getAll(PhysicalPlanRewriter.class);
+    physicalPlan = physicalPlan.applyRewriting(rewriters, environment);
+
     // Read database file if configured and check compatibility
     mainScript
         .getMutationDatabase()

sqrl-planner/src/main/java/com/datasqrl/config/CompilerApiConfigImpl.java

@@ -57,6 +57,11 @@ public int getDefaultLimit() {
     return sqrlConfig.asInt("default-limit").get();
   }
 
+  @Override
+  public boolean generatePaginatedResults() {
+    return sqrlConfig.asBool("paginated-results").get();
+  }
+
   public enum Endpoints {
     OPS_ONLY, // only support the pre-defined operations in the GraphQL API, do not support flexible
     // GraphQL queries TODO: not yet implemented

sqrl-planner/src/main/java/com/datasqrl/config/PackageJson.java

@@ -79,6 +79,8 @@ interface CompilerApiConfig {
     int getMaxResultDepth();
 
     int getDefaultLimit();
+
+    boolean generatePaginatedResults();
   }
 
   interface ExplainConfig {

sqrl-planner/src/main/java/com/datasqrl/engine/PhysicalPlan.java

@@ -55,8 +55,8 @@ public PhysicalPlan applyRewriting(
     for (PhysicalStagePlan stagePlan : stagePlans) {
       var enginePlan = stagePlan.plan;
       for (PhysicalPlanRewriter rewriter : rewriters) {
-        if (rewriter.appliesTo(enginePlan)) {
-          enginePlan = rewriter.rewrite(enginePlan, sqrlEnv);
+        if (rewriter.satisfied(this) && rewriter.appliesTo(enginePlan)) {
+          enginePlan = rewriter.rewrite(this, enginePlan, sqrlEnv);
         }
       }
       builder.stagePlan(new PhysicalStagePlan(stagePlan.stage, enginePlan));

sqrl-planner/src/main/java/com/datasqrl/engine/server/ServerPhysicalPlan.java

@@ -16,13 +16,16 @@
 package com.datasqrl.engine.server;
 
 import com.datasqrl.engine.EnginePhysicalPlan;
+import com.datasqrl.planner.analyzer.TableAnalysis;
 import com.datasqrl.planner.dag.plan.MutationTable;
 import com.datasqrl.planner.tables.SqrlTableFunction;
 import com.datasqrl.server.graphql.RootGraphQLModel;
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 
@@ -44,4 +47,10 @@ public class ServerPhysicalPlan implements EnginePhysicalPlan {
    * plan later.
    */
   final Map<String, RootGraphQLModel> models = new LinkedHashMap<>();
+
+  /**
+   * Base tables of paginated queries, collected during model generation so a rowtime index can be
+   * added to their physical tables afterwards.
+   */
+  @JsonIgnore final Set<TableAnalysis> pagedRowTimeTables = new LinkedHashSet<>();
 }

sqrl-planner/src/main/java/com/datasqrl/plan/global/JdbcIndexOptimization.java

@@ -16,6 +16,7 @@
 package com.datasqrl.plan.global;
 
 import com.datasqrl.engine.EnginePhysicalPlan;
+import com.datasqrl.engine.PhysicalPlan;
 import com.datasqrl.engine.database.relational.AbstractJDBCDatabaseEngine;
 import com.datasqrl.engine.database.relational.JdbcPhysicalPlan;
 import com.datasqrl.planner.Sqrl2FlinkSQLTranslator;
@@ -32,14 +33,16 @@
 public class JdbcIndexOptimization implements PhysicalPlanRewriter {
 
   @Override
-  public boolean appliesTo(EnginePhysicalPlan plan) {
-    return plan instanceof JdbcPhysicalPlan jpp
+  public boolean appliesTo(EnginePhysicalPlan enginePlan) {
+    return enginePlan instanceof JdbcPhysicalPlan jpp
         && jpp.stage().engine() instanceof AbstractJDBCDatabaseEngine;
   }
 
   @Override
-  public JdbcPhysicalPlan rewrite(EnginePhysicalPlan plan, Sqrl2FlinkSQLTranslator sqrlEnv) {
-    var jdbcPlan = (JdbcPhysicalPlan) plan;
+  public JdbcPhysicalPlan rewrite(
+      PhysicalPlan fullPlan, EnginePhysicalPlan enginePlan, Sqrl2FlinkSQLTranslator sqrlEnv) {
+
+    var jdbcPlan = (JdbcPhysicalPlan) enginePlan;
     var engine = (AbstractJDBCDatabaseEngine) jdbcPlan.stage().engine();
     /*TODO: optimize the order of primary key columns (unless primary key is explicitly defined by hint)
        - partition keys come first

sqrl-planner/src/main/java/com/datasqrl/plan/global/PagedRowTimeIndexRewriter.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright © 2021 DataSQRL (contact@datasqrl.com)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.datasqrl.plan.global;
+
+import com.datasqrl.engine.EnginePhysicalPlan;
+import com.datasqrl.engine.PhysicalPlan;
+import com.datasqrl.engine.database.relational.AbstractJDBCDatabaseEngine;
+import com.datasqrl.engine.database.relational.JdbcPhysicalPlan;
+import com.datasqrl.engine.database.relational.JdbcStatement;
+import com.datasqrl.engine.server.ServerPhysicalPlan;
+import com.datasqrl.planner.Sqrl2FlinkSQLTranslator;
+import com.google.auto.service.AutoService;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/**
+ * Adds a btree index on the rowtime column of every physical table that backs a paginated ({@code
+ * OffsetPageInfo}) query. Those queries run a companion {@code MIN/MAX(rowtime)} aggregate for
+ * their {@code first/lastEventTime}; the btree lets the database answer it from the index endpoints
+ * instead of scanning the table.
+ */
+@AutoService(PhysicalPlanRewriter.class)
+public class PagedRowTimeIndexRewriter implements PhysicalPlanRewriter {
+
+  @Override
+  public boolean appliesTo(EnginePhysicalPlan enginePlan) {
+    return enginePlan instanceof JdbcPhysicalPlan jpp
+        && jpp.stage().engine() instanceof AbstractJDBCDatabaseEngine;
+  }
+
+  @Override
+  public boolean satisfied(PhysicalPlan fullPlan) {
+    return getServerPlan(fullPlan).isPresent();
+  }
+
+  @Override
+  public JdbcPhysicalPlan rewrite(
+      PhysicalPlan fullPlan, EnginePhysicalPlan enginePlan, Sqrl2FlinkSQLTranslator sqrlEnv) {
+    var jdbcPlan = (JdbcPhysicalPlan) enginePlan;
+    var serverPlan =
+        getServerPlan(fullPlan)
+            .orElseThrow(() -> new IllegalStateException("Server physical plan is missing"));
+
+    var pagedTables = serverPlan.getPagedRowTimeTables();
+    if (pagedTables.isEmpty()) {
+      return jdbcPlan;
+    }
+
+    var engine = (AbstractJDBCDatabaseEngine) jdbcPlan.stage().engine();
+    if (!engine.getIndexSelectorConfig().supportedIndexTypes().contains(IndexType.BTREE)) {
+      return jdbcPlan;
+    }
+    var stmtFactory = engine.getStatementFactory();
+    // Existing indexes (e.g. from JdbcIndexOptimization) may already cover the rowtime column.
+    var existingIndexNames =
+        jdbcPlan.getStatementsForType(JdbcStatement.Type.INDEX).stream()
+            .map(JdbcStatement::getName)
+            .collect(Collectors.toSet());
+
+    var builder = jdbcPlan.toBuilder();
+    for (var createTbl : jdbcPlan.tableIdMap().values()) {
+      var engineTable = createTbl.getEngineTable();
+      var tableAnalysis = engineTable.tableAnalysis();
+      if (!pagedTables.contains(tableAnalysis)
+          && !pagedTables.contains(tableAnalysis.getBaseTable())) {
+        continue;
+      }
+
+      var rowTime = tableAnalysis.getRowTime();
+      if (rowTime.isEmpty()) {
+        continue;
+      }
+
+      var index =
+          new IndexDefinition(
+              engineTable.tableName(),
+              List.of(rowTime.get()),
+              tableAnalysis.getRowType().getFieldNames(),
+              -1,
+              IndexType.BTREE);
+
+      if (existingIndexNames.add(index.getName())) {
+        builder.statement(stmtFactory.addIndex(index));
+      }
+    }
+
+    return builder.build();
+  }
+
+  private Optional<ServerPhysicalPlan> getServerPlan(PhysicalPlan fullPlan) {
+    return fullPlan.getPlans(ServerPhysicalPlan.class).findAny();
+  }
+}

sqrl-planner/src/main/java/com/datasqrl/plan/global/PhysicalPlanRewriter.java

@@ -16,11 +16,17 @@
 package com.datasqrl.plan.global;
 
 import com.datasqrl.engine.EnginePhysicalPlan;
+import com.datasqrl.engine.PhysicalPlan;
 import com.datasqrl.planner.Sqrl2FlinkSQLTranslator;
 
 public interface PhysicalPlanRewriter {
 
-  boolean appliesTo(EnginePhysicalPlan plan);
+  boolean appliesTo(EnginePhysicalPlan enginePlan);
 
-  EnginePhysicalPlan rewrite(EnginePhysicalPlan plan, Sqrl2FlinkSQLTranslator sqrlEnv);
+  EnginePhysicalPlan rewrite(
+      PhysicalPlan fullPlan, EnginePhysicalPlan enginePlan, Sqrl2FlinkSQLTranslator sqrlEnv);
+
+  default boolean satisfied(PhysicalPlan fullPlan) {
+    return true;
+  }
 }

sqrl-planner/src/main/java/com/datasqrl/server/GenerateServerModel.java

@@ -48,6 +48,7 @@ public RootGraphQLModel generateGraphQLModel(ApiSources api, ServerPhysicalPlan
         new GraphqlModelGenerator(
             serverPlan.getFunctions(), serverPlan.getMutations(), errorCollector);
     graphqlModelGenerator.walkAPISource(api.schema());
+    serverPlan.getPagedRowTimeTables().addAll(graphqlModelGenerator.getPagedRowTimeTables());
     var schema = StringSchema.builder().schema(api.schema().getDefinition()).build();
     var graphSchema = converter.getSchema(schema.getSchema());
     var apiConfig = configuration.getCompilerConfig().getApiConfig();