API index and search · Build metadata

Source snapshot

packages/normalized/src/agg-table.ts

1// Synthetic aggregate tables (AGGREGATE-SYNC-DESIGN.md §3.1/§3.3) — the client twin of2// `rindle-replica/src/normalize.rs`. A relationship `count` is never recomputed locally3// (the client lacks the child rows); the server ships the reduce's `(group_key, count)`4// output as a SYNTHETIC base table, and the client:5//   1. registers that table on its local engine + the `NormalizedSync` refcount layer,6//   2. rewrites the local AST so the relationship reads the table with a plain singular7//      join + the same scalar projection (`aggregatePrecomputed`) instead of a reduce.8//9// The table NAME is a content hash of the aggregate's definition, so client and server10// agree without coordinating. `aggTableName` is the **byte-exact** twin of Rust11// `agg_table_name` — the FNV-1a-64 byte protocol documented there; cross-language vectors12// pin it (Rust `agg_table_name_cross_language_vector` ↔ this module's test).1314import type {15  Aggregate,16  Ast,17  Condition,18  CorrelatedSubquery,19  Correlation,20  LitValue,21  NormalizedTableSchema,22  ValuePosition,23} from "@rindle/client";2425// FNV-1a-64, the same constants as `protocol.ts`'s `Fnv` (a local copy so `@rindle/normalized`26// keeps no dependency on `@rindle/remote`). The byte methods mirror Rust `normalize::Fnv`.27const FNV_OFFSET = 0xcbf29ce484222325n;28const FNV_PRIME = 0x00000100000001b3n;29const MASK = 0xffffffffffffffffn;30const enc = new TextEncoder();3132class Fnv {33  h = FNV_OFFSET;34  byte(b: number): void {35    this.h = ((this.h ^ BigInt(b & 0xff)) * FNV_PRIME) & MASK;36  }37  /** A `u32` little-endian (Rust `v.to_le_bytes()`). */38  u32(v: number): void {39    this.byte(v & 0xff);40    this.byte((v >>> 8) & 0xff);41    this.byte((v >>> 16) & 0xff);42    this.byte((v >>> 24) & 0xff);43  }44  /** A length-prefixed string: `u32(byteLen)` then UTF-8 bytes. */45  s(str: string): void {46    const bytes = enc.encode(str);47    this.u32(bytes.length);48    for (const b of bytes) this.byte(b);49  }50  /** An `f64` little-endian (Rust `n.to_le_bytes()` — IEEE-754 binary64). */51  f64(v: number): void {52    const buf = new ArrayBuffer(8);53    new DataView(buf).setFloat64(0, v, true);54    for (const b of new Uint8Array(buf)) this.byte(b);55  }56  /** An `i64` little-endian (Rust `i.to_le_bytes()` — two's complement). */57  i64(v: bigint): void {58    const u = BigInt.asUintN(64, v);59    for (let shift = 0n; shift < 64n; shift += 8n) this.byte(Number((u >> shift) & 0xffn));60  }61  hex(): string {62    return this.h.toString(16).padStart(16, "0");63  }64}6566/** The synthetic base-table name for a relationship `count` aggregate — the byte-exact twin67 *  of Rust `agg_table_name` (`normalize.rs` §3.1). A content hash of the aggregate definition68 *  (child table, kind, group key = correlation **child** fields, child `where`); the parent69 *  correlation field is excluded. Same definition ⇒ same name (cross-query sharing); a70 *  different filter ⇒ a different name (no `(table, pk)` collision). */71export function aggTableName(csq: CorrelatedSubquery): string {72  const f = new Fnv();73  f.u32(csq.correlation.childField.length);74  for (const cf of csq.correlation.childField) f.s(cf);75  hashAggSubquery(f, csq.subquery);76  return `__agg_${f.hex()}`;77}7879function hashAggSubquery(f: Fnv, ast: Ast): void {80  f.s(ast.table);81  f.byte(ast.aggregate === "count" ? 1 : 0); // Some(Count)=1, None=082  if (ast.where === undefined) f.byte(0);83  else {84    f.byte(1);85    hashCondition(f, ast.where);86  }87  const related = ast.related ?? [];88  f.u32(related.length);89  for (const r of related) {90    hashCorrelation(f, r.correlation);91    hashAggSubquery(f, r.subquery);92  }93}9495function hashCorrelation(f: Fnv, c: Correlation): void {96  f.u32(c.parentField.length);97  for (const p of c.parentField) f.s(p);98  f.u32(c.childField.length);99  for (const x of c.childField) f.s(x);100}101102function hashCondition(f: Fnv, cond: Condition): void {103  switch (cond.type) {104    case "simple":105      f.byte(0);106      f.s(cond.op); // SimpleOp is the wire string ("=", "!=", …) === Rust `op_str`107      hashValuePosition(f, cond.left);108      hashValuePosition(f, cond.right);109      break;110    case "and":111      f.byte(1);112      f.u32(cond.conditions.length);113      for (const c of cond.conditions) hashCondition(f, c);114      break;115    case "or":116      f.byte(2);117      f.u32(cond.conditions.length);118      for (const c of cond.conditions) hashCondition(f, c);119      break;120    case "correlatedSubquery":121      f.byte(3);122      f.byte(cond.op === "EXISTS" ? 1 : 0); // Exists=1, NotExists=0123      f.byte(systemByte(cond.related.system));124      hashCorrelation(f, cond.related.correlation);125      hashAggSubquery(f, cond.related.subquery);126      break;127  }128}129130function systemByte(sys: CorrelatedSubquery["system"]): number {131  switch (sys) {132    case "permissions":133      return 1; // System::Permissions134    case "client":135      return 2; // System::Client136    case "test":137      return 3; // System::Test138    default:139      return 0; // None140  }141}142143function hashValuePosition(f: Fnv, vp: ValuePosition): void {144  if (vp.type === "column") {145    f.byte(0);146    f.s(vp.name);147  } else {148    f.byte(1);149    hashLit(f, vp.value);150  }151}152153// The serde boundary for a number literal (Rust `Lit`, untagged): an INTEGER-form JSON154// token in i64 range deserializes as `Lit::Int`; every other number falls through to155// `Number(f64)`.156const INTEGER_TOKEN = /^-?[0-9]+$/;157const I64_MIN = -(1n << 63n);158const I64_MAX = (1n << 63n) - 1n;159160function hashLit(f: Fnv, lit: LitValue): void {161  if (lit === null) f.byte(0);162  else if (typeof lit === "boolean") {163    f.byte(1);164    f.byte(lit ? 1 : 0);165  } else if (typeof lit === "number") {166    // The exact-int plane, mirroring serde's untagged rule for Rust `Lit` (design 226167    // Stage B): what serde sees is the WIRE TOKEN, so hash the token, not the binary168    // value. `String(lit)` is exactly the token `JSON.stringify` emits, and above 2^53169    // the two diverge: JS serializes the SHORTEST decimal that round-trips (`2 ** 60` →170    // "1152921504606847000", not ...846976), and serde parses that token as the i64.171    // An integer-form token in i64 range → `Lit::Int` → tag 5 + exact i64 LE bytes;172    // everything else (non-integral, exponent-form ≥ 1e21, out of i64 range) → tag 2 +173    // f64 — where serde's parse also lands back on `lit`'s exact bits. Hashing either174    // side differently would name a different synthetic table from the server's, and175    // the aggregate rows would never route.176    if (!Number.isFinite(lit)) {177      // NaN/±Infinity have NO wire token — `JSON.stringify` emits `null`, so the178      // server parses `Lit::Null` and hashes the null tag. Mirror that, not the179      // binary f64 bits String() would suggest ("NaN" is not a JSON token).180      f.byte(0);181      return;182    }183    const token = String(lit);184    let int: bigint | null = null;185    if (INTEGER_TOKEN.test(token)) {186      const parsed = BigInt(token);187      if (parsed >= I64_MIN && parsed <= I64_MAX) int = parsed;188    }189    if (int !== null) {190      f.byte(5);191      f.i64(int);192    } else {193      f.byte(2);194      f.f64(lit);195    }196  } else if (typeof lit === "string") {197    f.byte(3);198    f.s(lit);199  } else if (typeof lit === "bigint") {200    // A bigint literal cannot reach the server at all — `JSON.stringify` throws on201    // bigint, so no wire token exists to mirror. Refuse informatively here rather202    // than fall into the array branch's `lit.length`/iteration TypeError.203    throw new Error(204      "bigint literals are not supported on the live-query wire until the browser " +205        "bigint lane ships (design 226) — pass a number, or read exact int64 cells " +206        "through the SQL plane",207    );208  } else {209    f.byte(4);210    f.u32(lit.length);211    for (const x of lit) hashLit(f, x);212  }213}214215/** Every synthetic aggregate table `ast` surfaces (recursively, a nested aggregate under a216 *  materialized relationship included), as flat table schemas — the twin of Rust217 *  `agg_table_schemas`. Columns `[childField…, "count"]`; PK = the leading group columns. */218export function aggTableSchemas(ast: Ast, isLocal?: (table: string) => boolean): NormalizedTableSchema[] {219  const out: NormalizedTableSchema[] = [];220  collectAggTables(ast, out, isLocal);221  return out;222}223224function collectAggTables(ast: Ast, out: NormalizedTableSchema[], isLocal?: (table: string) => boolean): void {225  for (const r of ast.related ?? []) {226    if (r.subquery.aggregate) {227      // L1 (`201-LOCAL-ONLY-TABLES-DESIGN.md` §5.2): a count over a LOCAL child is a native IVM228      // reduce — it has no server-authoritative `__agg_*` base, so emit no synthetic schema for it.229      if (isLocal?.(r.subquery.table)) continue;230      const k = r.correlation.childField.length;231      out.push({232        name: aggTableName(r),233        columns: [...r.correlation.childField, "count"],234        primaryKey: Array.from({ length: k }, (_, i) => i),235      });236    } else {237      collectAggTables(r.subquery, out, isLocal);238    }239  }240}241242/** Rewrite an AST for the LOCAL engine: each relationship `count` becomes a precomputed,243 *  source-backed singular relationship over its synthetic table — read the server's count244 *  with a plain join + the same scalar projection (`aggregatePrecomputed`), never a `reduce`245 *  (which would recount the already-aggregated rows). Non-aggregate relationships recurse246 *  (a nested aggregate is rewritten too); the parent's frame is otherwise untouched, so the247 *  view-schema slot order is preserved. Returns a new AST; the input is not mutated.248 *249 *  The `where` tree is rewritten the SAME way (`PARENT-AGGREGATE-FILTER-DESIGN.md` §3): a250 *  `having_count` parent gate lowers to an EXISTS whose subquery CLONES the display `count_as`251 *  and adds a post-aggregation `HAVING`. That is a relationship `count` like any other, so §3's252 *  premise — the client never recomputes a count, it lacks the child rows — applies to it too;253 *  left un-rewritten the gate would reduce over child rows the server (rightly) does not sync254 *  and every parent would fail it. `agg_table_name` hashes neither `alias` nor `having`, so the255 *  gate resolves to the SAME `__agg_*` table the display `count_as` already registers and syncs:256 *  the rewrite costs no extra table, no extra rows, and reads a count the server already sent.257 *  Its counterpart is the server pruning the gate's witnesses from the footprint (Rust258 *  `table_tree`) — that prune is only sound BECAUSE of this rewrite, so ship this side first. */259export function rewriteAggregates(ast: Ast, isLocal?: (table: string) => boolean): Ast {260  const where = ast.where && rewriteConditionAggregates(ast.where, isLocal);261  const related = ast.related?.length262    ? ast.related.map((r) => rewriteRelationship(r, isLocal))263    : ast.related;264  if (where === ast.where && related === ast.related) return ast;265  const out: Ast = { ...ast };266  if (where) out.where = where;267  if (related) out.related = related;268  return out;269}270271/** The `where`-tree half of {@link rewriteAggregates}: an EXISTS gate whose subquery carries an272 *  aggregate (the `having_count` lowering) is rewritten by the very same273 *  {@link rewriteRelationship}; every other condition recurses structurally. Returns `cond`274 *  itself when nothing below it rewrote, so an untouched frame keeps object identity and the275 *  caller can skip the copy. */276function rewriteConditionAggregates(cond: Condition, isLocal?: (table: string) => boolean): Condition {277  switch (cond.type) {278    case "and":279    case "or": {280      let changed = false;281      const conditions = cond.conditions.map((c) => {282        const next = rewriteConditionAggregates(c, isLocal);283        changed ||= next !== c;284        return next;285      });286      return changed ? { ...cond, conditions } : cond;287    }288    case "correlatedSubquery": {289      const related = rewriteRelationship(cond.related, isLocal);290      return related === cond.related ? cond : { ...cond, related };291    }292    default:293      return cond; // a `simple` leaf holds no subquery294  }295}296297function rewriteRelationship(csq: CorrelatedSubquery, isLocal?: (table: string) => boolean): CorrelatedSubquery {298  // L1 (`201-LOCAL-ONLY-TABLES-DESIGN.md` §5.2): a count over a LOCAL child is left as the native299  // IVM reduce (the pre-agg-sync path). The server never ships a count for it, so the precomputed300  // rewrite would point the relationship at an `__agg_*` table that is fed nothing → empty forever.301  if (csq.subquery.aggregate && !isLocal?.(csq.subquery.table)) {302    const subquery: Ast = {303      table: aggTableName(csq),304      aggregate: csq.subquery.aggregate as Aggregate,305      aggregatePrecomputed: true,306    };307    // The correlation is unchanged: the synthetic table's group columns are named after the308    // child correlation fields, so `childField` still resolves against `[childField…, count]`.309    if (csq.subquery.alias !== undefined) subquery.alias = csq.subquery.alias;310    // The post-aggregation `HAVING` SURVIVES the rewrite — it is the gate's whole predicate311    // (`count > n`), and it addresses the reduce's output column, which is exactly what the312    // synthetic row carries. A display `count_as` never has one, so this is inert there.313    if (csq.subquery.having !== undefined) subquery.having = csq.subquery.having;314    const out: CorrelatedSubquery = { correlation: csq.correlation, subquery };315    if (csq.system !== undefined) out.system = csq.system;316    return out;317  }318  const subquery = rewriteAggregates(csq.subquery, isLocal);319  return subquery === csq.subquery ? csq : { ...csq, subquery };320}321