feat(engine): size the checkpoint threshold from the store and set it with CHECKPOINT AT

database/src/main/java/io/hstore/db/query/Ast.java

@@ -314,6 +314,9 @@ record ShowViews() implements Statement {
         record Checkpoint() implements Statement {
         }
 
+        record CheckpointAt(OptionalLong bytes) implements Statement {
+        }
+
         record Compact() implements Statement {
         }
     }

database/src/main/java/io/hstore/db/query/Executor.java

@@ -37,6 +37,7 @@
 import io.hstore.db.value.Value;
 import io.hstore.db.view.MaterializedViews;
 import io.hstore.db.view.ViewCell;
+import io.hstore.engine.CheckpointThreshold;
 import io.hstore.engine.EngineStats;
 import io.hstore.engine.HStoreException;
 import io.hstore.engine.catalog.Branch;
@@ -316,6 +317,7 @@ QueryResult run(Statement statement) {
                 var result = database().engine().checkpoint();
                 yield QueryResult.message("checkpoint at generation " + result.generation() + ", lsn " + result.lsn());
             }
+            case Statement.CheckpointAt(OptionalLong bytes) -> QueryResult.message(describe(database().engine().checkpointAt(bytes)));
             case Statement.Compact _ -> {
                 var report = database().engine().compact();
                 yield QueryResult.message("compacted segments " + report.compacted() + " at generation " + report.generation());
@@ -330,14 +332,22 @@ private static boolean administrative(Statement statement) {
         return switch (statement) {
             case Statement.CreateBranch _, Statement.DropBranch _, Statement.MergeBranch _, Statement.Sample _,
                  Statement.DiffGenerations _, Statement.DiffBranches _, Statement.History _, Statement.Stats _,
-                 Statement.Checkpoint _, Statement.Compact _, Statement.CreateTenant _, Statement.AlterTenant _,
+                 Statement.Checkpoint _, Statement.CheckpointAt _, Statement.Compact _, Statement.CreateTenant _, Statement.AlterTenant _,
                  Statement.CreateUser _, Statement.AlterUser _, Statement.DropUser _, Statement.ShowTenants _, Statement.ShowUsers _,
                  Statement.CreateView _, Statement.RefreshView _, Statement.ShowView _, Statement.CreateType _,
                  Statement.CreateIndex _, Statement.CreateJsonIndex _ -> true;
             default -> false;
         };
     }
 
+    private static String describe(CheckpointThreshold threshold) {
+        return "checkpoint after " + threshold.bytes() + " bytes of log, " + switch (threshold.source()) {
+            case AUTOMATIC -> "sized from the store";
+            case DATABASE -> "set by CHECKPOINT AT";
+            case CONFIGURATION -> "set by configuration";
+        };
+    }
+
     private static Cell limit(long value) {
         return value == Long.MAX_VALUE ? QueryResult.text("unlimited") : QueryResult.number(value);
     }
@@ -612,6 +622,9 @@ private QueryResult stats() {
         metrics.put("data bytes written", QueryResult.number(engine.dataBytesWritten()));
         metrics.put("wal bytes", QueryResult.number(engine.walBytes()));
         metrics.put("wal segments", QueryResult.number(engine.walSegments()));
+        CheckpointThreshold threshold = database().engine().checkpointThreshold();
+        metrics.put("checkpoint after wal bytes", QueryResult.number(threshold.bytes()));
+        metrics.put("checkpoint threshold", QueryResult.text(threshold.source().name().toLowerCase()));
         metrics.put("cache hit rate", QueryResult.number(Math.round(engine.cacheHitRate() * 1000) / 1000.0));
         metrics.put("data segments", QueryResult.number(engine.segments().size()));
         metrics.put("feed bytes", QueryResult.number(engine.feedBytes()));

database/src/main/java/io/hstore/db/query/Parser.java

@@ -393,6 +393,9 @@ private Statement statement() {
             return new Statement.ShowViews();
         }
         if (accept("CHECKPOINT")) {
+            if (accept("AT")) {
+                return new Statement.CheckpointAt(accept("AUTO") ? OptionalLong.empty() : OptionalLong.of(size()));
+            }
             return new Statement.Checkpoint();
         }
         if (accept("COMPACT")) {
@@ -931,6 +934,32 @@ private BigDecimal number() {
         }
     }
 
+    private long size() {
+        BigDecimal amount = number();
+        Token unit = next();
+        long scale = unit.kind() != Token.Kind.WORD ? 0 : switch (unit.text().toUpperCase()) {
+            case "B" -> 1;
+            case "KB", "KIB" -> 1L << 10;
+            case "MB", "MIB" -> 1L << 20;
+            case "GB", "GIB" -> 1L << 30;
+            case "TB", "TIB" -> 1L << 40;
+            default -> 0;
+        };
+        if (scale == 0) {
+            throw error("expected a size unit (B, KB, MB, GB or TB) but found " + unit);
+        }
+        long bytes;
+        try {
+            bytes = amount.multiply(BigDecimal.valueOf(scale)).longValueExact();
+        } catch (ArithmeticException e) {
+            throw error(amount.toPlainString() + " " + unit.text() + " is not a whole number of bytes");
+        }
+        if (bytes < 1) {
+            throw error("a size must be positive, not " + amount.toPlainString() + " " + unit.text());
+        }
+        return bytes;
+    }
+
     private long longValue() {
         try {
             return number().longValueExact();

database/src/test/java/io/hstore/db/SecurityTest.java

@@ -5,6 +5,7 @@
 
 import io.hstore.db.query.QueryResult;
 import io.hstore.db.query.Session;
+import io.hstore.engine.CheckpointThreshold;
 import io.hstore.engine.EngineOptions;
 import io.hstore.engine.HStoreException;
 
@@ -126,6 +127,9 @@ void rolesAndAuthenticationAreEnforced() {
             assertThrows(HStoreException.InvalidSchema.class, () -> auditor.execute("INSERT NODE Patient 'x' {mrn: 'x'}"));
             assertThrows(HStoreException.InvalidSchema.class, () -> alice.execute("CREATE NODE TYPE Forbidden"));
             assertThrows(HStoreException.InvalidSchema.class, () -> alice.execute("USE TENANT globex"));
+            HStoreException.InvalidSchema denied = assertThrows(HStoreException.InvalidSchema.class, () -> alice.execute("CHECKPOINT AT 64MB"));
+            assertTrue(denied.getMessage().contains("ADMIN"), denied.getMessage());
+            assertEquals(CheckpointThreshold.Source.AUTOMATIC, database.engine().checkpointThreshold().source());
             assertEquals("acme", alice.execute("WHOAMI").rows().getFirst().get(1).render());
         }
         assertTrue(database.requiresAuthentication());

database/src/test/java/io/hstore/db/query/QueryLanguageTest.java

@@ -17,6 +17,8 @@
 import java.nio.file.Path;
 import java.util.List;
 import java.util.Map;
+import java.util.OptionalLong;
+import java.util.stream.Collectors;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -253,4 +255,39 @@ void syntaxErrorsAreReportedWithOffsets() {
         HStoreException.InvalidSchema error = assertThrows(HStoreException.InvalidSchema.class, () -> session.execute("MATCH EDGE c:Claim WHERE"));
         assertTrue(error.getMessage().contains("offset"), error.getMessage());
     }
+
+    @Test
+    void checkpointAtTakesSizesInAnyCaseOrAuto() {
+        Map<String, Long> sizes = Map.of("CHECKPOINT AT 64MB", 64L << 20, "checkpoint at 64 mb", 64L << 20, "CHECKPOINT AT 1GB", 1L << 30,
+                "CHECKPOINT AT 512KB", 512L << 10, "CHECKPOINT AT 1.5 GiB", 3L << 29, "Checkpoint At 4096 b", 4096L);
+        sizes.forEach((text, bytes) -> assertEquals(new Ast.Statement.CheckpointAt(OptionalLong.of(bytes)), Parser.parse(text), text));
+        assertEquals(new Ast.Statement.CheckpointAt(OptionalLong.empty()), Parser.parse("CHECKPOINT AT AUTO"));
+        assertEquals(new Ast.Statement.CheckpointAt(OptionalLong.empty()), Parser.parse("checkpoint at auto"));
+        assertEquals(new Ast.Statement.Checkpoint(), Parser.parse("CHECKPOINT"));
+        for (String bad : List.of("CHECKPOINT AT", "CHECKPOINT AT 64", "CHECKPOINT AT 64 PB", "CHECKPOINT AT 0MB", "CHECKPOINT AT -1MB",
+                "CHECKPOINT AT 0.5B", "CHECKPOINT AT MB", "CHECKPOINT AT 99999999999TB", "CHECKPOINT AT 'auto'", "CHECKPOINT AT 64MB 2")) {
+            HStoreException.InvalidSchema error = assertThrows(HStoreException.InvalidSchema.class, () -> session.execute(bad), bad);
+            assertTrue(error.getMessage().contains("syntax error"), error.getMessage());
+        }
+    }
+
+    @Test
+    void checkpointAtIsKeptAcrossReopenAndAutoClearsIt() {
+        assertTrue(session.execute("CHECKPOINT AT 64MB").message().contains("67108864 bytes of log, set by CHECKPOINT AT"));
+        assertEquals(List.of("67108864", "database"), checkpointStats());
+        session.close();
+        database.close();
+        database = HypergraphDatabase.open(directory, DatabaseOptions.defaults()
+                .withEngine(EngineOptions.defaults().withPageSize(4096).withHistoryLimit(500)));
+        session = new Session(database);
+        assertEquals(List.of("67108864", "database"), checkpointStats());
+        assertTrue(session.execute("CHECKPOINT AT AUTO").message().contains("sized from the store"));
+        assertEquals(List.of(String.valueOf(2L << 20), "automatic"), checkpointStats());
+    }
+
+    private List<String> checkpointStats() {
+        Map<String, String> metrics = session.execute("STATS").rows().stream()
+                .collect(Collectors.toMap(row -> row.get(0).render(), row -> row.get(1).render()));
+        return List.of(metrics.get("checkpoint after wal bytes"), metrics.get("checkpoint threshold"));
+    }
 }

docs/architecture/overview.md

@@ -210,7 +210,7 @@ The commit's tree nodes stay in memory. A checkpoint later writes every node rea
 |---|---|---|---|
 | Caller threads | any (the server uses virtual threads) | the caller | Run reads without locks. Run a transaction's operations without locks. Before committing, wait if the write buffer is full, then take `commitLock` only for append (encoding the key changes, WAL append, and page writes for the rare tree past 4,096 changes). They then block in `GroupCommitter.await` outside the lock. |
 | (none; commit leader) | the committing thread | `GroupCommitter.await` | There is no group-commit thread. The first committer waiting on an unpublished commit leads: it takes every queued `Pending` as one batch and calls `makeDurable(batchSize, wrotePages)` once (data sync in `PAGE_REFERENCES` mode if a commit in the batch wrote pages, then WAL sync, both only under `Durability.SYNC`). It then runs each publication in generation order and wakes the waiters. One fsync pair covers the whole batch (`averageGroupCommit` in the statistics). A failure poisons the pipeline, and every later commit fails until restart. |
-| `hstore-maintenance` | virtual | `StorageEngine` constructor | Parks for `maintenanceInterval` (default 500 ms), or until a commit that finds the write buffer full unparks it. Then, under the `maintenance` lock (`tryLock`): checkpoint once the write buffer passes half of `write_buffer_mb` or the WAL or the pages written since the last checkpoint exceed `checkpoint_wal_mb`, compact one segment if any has fallen below `compaction_live_ratio`, then reclaim retired segments that no pinned reader can still see. A background checkpoint writes the pages, counts references and publishes the catalog image on this thread. |
+| `hstore-maintenance` | virtual | `StorageEngine` constructor | Parks for `maintenanceInterval` (default 500 ms), or until a commit that finds the write buffer full, or that passes the checkpoint threshold, unparks it. Then, under the `maintenance` lock (`tryLock`): checkpoint once the write buffer passes half of `write_buffer_mb` or the WAL or the pages written since the last checkpoint exceed the checkpoint threshold (`CHECKPOINT AT`, else `checkpoint_wal_mb`, else a quarter of the live data between 2 MiB and 256 MiB), compact one segment if any has fallen below `compaction_live_ratio`, then reclaim retired segments that no pinned reader can still see. A background checkpoint writes the pages, counts references and publishes the catalog image on this thread. |
 | `hstore-references` | platform, daemon | `StorageEngine` constructor | Marks dead the pages a transaction spilled and then gave up (abort, conflict, page budget, duplicate request). The queue holds up to 4 096 entries; past that, the caller waits. Counting itself happens at checkpoints, on the checkpointing thread. |
 | `feed-subscriber` | virtual, one per subscription | `ChangeFeed.subscribe` | Tails the feed segments, waiting on a condition signalled by `append` (with a 250 ms timeout). Subscribers are `SemanticPlane`, `MaterializedViews` and `Statistics`. |
 | Server connections | virtual, one per socket | `Server` (`Executors.newVirtualThreadPerTaskExecutor`) | Each connection runs a `Session`. |

docs/database/hql.md

@@ -83,7 +83,7 @@ Values and their storage are described in [data model](data-model.md#values).
 | [Views](#materialized-views) | `CREATE VIEW`, `REFRESH VIEW`, `VIEW`, `SHOW VIEWS` |
 | [Tenancy](#tenancy-and-users) | `CREATE TENANT`, `ALTER TENANT`, `SHOW TENANTS`, `CREATE USER`, `ALTER USER`, `DROP USER`, `SHOW USERS`, `USE TENANT`, `AUTHENTICATE` |
 | [Session](#session-and-output) | `WHOAMI`, `FORMAT` |
-| [Maintenance](#maintenance) | `CHECKPOINT`, `COMPACT`, `STATS` |
+| [Maintenance](#maintenance) | `CHECKPOINT`, `CHECKPOINT AT`, `COMPACT`, `STATS` |
 
 Statements marked **ADMIN** are rejected for non-admin principals before execution
 (`Executor.administrative`). Write statements additionally require the WRITER role (`Principal.requireWrite`).
@@ -702,20 +702,38 @@ Atom cells render as `{"id", "label"}`, list cells as arrays, `NULL` as `null`,
 ## Maintenance
 
 ```ebnf
-checkpoint = "CHECKPOINT" ;   (* ADMIN *)
-compact    = "COMPACT" ;      (* ADMIN *)
-stats      = "STATS" ;        (* ADMIN *)
+checkpoint    = "CHECKPOINT" ;                                     (* ADMIN *)
+checkpoint_at = "CHECKPOINT" "AT" ( "AUTO" | number unit ) ;       (* ADMIN *)
+unit          = "B" | "KB" | "MB" | "GB" | "TB" | "KiB" | "MiB" | "GiB" | "TiB" ;   (* any case *)
+compact       = "COMPACT" ;                                        (* ADMIN *)
+stats         = "STATS" ;                                          (* ADMIN *)
 ```
 
 ```sql
-CHECKPOINT;   -- checkpoint at generation 246, lsn 222166
-COMPACT;      -- compacted segments [] at generation 246
+CHECKPOINT;              -- checkpoint at generation 246, lsn 222166
+CHECKPOINT AT 64MB;      -- checkpoint after 67108864 bytes of log, set by CHECKPOINT AT
+CHECKPOINT AT 1 gb;      -- checkpoint after 1073741824 bytes of log, set by CHECKPOINT AT
+CHECKPOINT AT AUTO;      -- checkpoint after 2097152 bytes of log, sized from the store
+COMPACT;                 -- compacted segments [] at generation 246
 STATS;
 ```
 
+`CHECKPOINT AT` sets how much WAL, or how many bytes of pages, the background maintenance thread lets build up
+before it takes a checkpoint. Units are powers of 1024 (`1KB` is 1,024 bytes) and the space before the unit is
+optional. The size must be a whole, positive number of bytes; a size without a unit is a syntax error. The value
+is stored in the database and committed like any other change, so it survives a restart or a crash, and it takes
+effect on the next maintenance pass. It wins over the `checkpoint_wal_mb` setting. `CHECKPOINT AT AUTO` removes
+it, and the threshold goes back to `checkpoint_wal_mb`, or to the automatic rule when that is `auto` (see
+[configuration](../operations/configuration.md#reference)). The statement commits on its own, outside any open
+transaction, and needs the ADMIN role.
+
+The value is kept in an engine slot that builds without `CHECKPOINT AT` don't know. Once a database has run
+`CHECKPOINT AT <size>`, older builds can no longer open it: they fail with `no slot registered with id 8`.
+
 `STATS` columns `metric, value`: catalog statistics (see [statistics](views-signals-stats.md#statistics)),
 engine counters (`commits`, `rebased commits`, `conflicts`, `pages read`, `pages written`, `data bytes written`,
-`wal bytes`, `wal segments`, `cache hit rate`, `data segments`, `feed bytes`), the `semantic index generation`
+`wal bytes`, `wal segments`, `checkpoint after wal bytes`, `checkpoint threshold` (`automatic`, `database` or
+`configuration`), `cache hit rate`, `data segments`, `feed bytes`), the `semantic index generation`
 and one `planner feedback <path>` row per access-path kind with a learned correction factor.
 
 ## Errors

docs/database/security.md

@@ -17,7 +17,7 @@ enum Role { ADMIN, WRITER, READER }
 | read atoms of its tenant | yes | yes | yes |
 | insert, update, delete atoms, members, properties, embeddings, evidence, state, signals | yes | yes | no |
 | DDL (`CREATE … TYPE`, `CREATE INDEX`, `CREATE JSON INDEX`) | yes | no | no |
-| branches (`CREATE/DROP/MERGE BRANCH`, `SAMPLE`, `DIFF`), `HISTORY`, `STATS`, `CHECKPOINT`, `COMPACT` | yes | no | no |
+| branches (`CREATE/DROP/MERGE BRANCH`, `SAMPLE`, `DIFF`), `HISTORY`, `STATS`, `CHECKPOINT`, `CHECKPOINT AT`, `COMPACT` | yes | no | no |
 | tenants and users | yes | no | no |
 | create, refresh and read materialized views (`SHOW VIEWS` lists names for every role) | yes | no | no |
 | switch tenant (`USE TENANT`, `Principal.inTenant`) | yes | no | no |