Add changesets

Author: simolus3

.changeset/early-hounds-marry.md

@@ -0,0 +1,5 @@
+---
+'@powersync/service-sync-rules': patch
+---
+
+Prepare new sync streams compiler.

packages/sync-rules/src/compiler/bucket_resolver.ts

@@ -5,6 +5,9 @@ import { PointLookup, RowEvaluator, SourceRowProcessor } from './rows.js';
 import { RequestTableValuedResultSet } from './table.js';
 import { StreamOptions } from '../sync_plan/plan.js';
 
+/**
+ * Describes how to resolve a subscription to buckets.
+ */
 export class StreamResolver {
   constructor(
     readonly options: StreamOptions,

packages/sync-rules/src/compiler/compiler.ts

@@ -1,5 +1,5 @@
 import { NodeLocation, parse, PGNode } from 'pgsql-ast-parser';
-import { HashSet, StableHasher } from './equality.js';
+import { HashSet } from './equality.js';
 import { PointLookup, RowEvaluator } from './rows.js';
 import { StreamResolver } from './bucket_resolver.js';
 import { StreamOptions, SyncPlan } from '../sync_plan/plan.js';
@@ -9,6 +9,16 @@ import { StreamQueryParser } from './parser.js';
 
 /**
  * State for compiling sync streams.
+ *
+ * The output of compiling all sync streams is a {@link SyncPlan}, a declarative description of the sync process that
+ * can be serialized to bucket storage. The compiler stores a mutable intermediate representation that is essentially a
+ * copy of the sync plan, except that we're using JavaScript classes with methods to compute hash codes and equality
+ * relations. This allows the compiler to efficiently de-duplicate parameters and buckets.
+ *
+ * Overall, the compilation process is as follows: Each data query for a stream is first parsed by
+ * {@link StreamQueryParser} into a canonicalized intermediate representation (see that class for details).
+ * Then, {@link QuerierGraphBuilder} analyzes a chain of `AND` expressions to identify parameters (as partition keys)
+ * and their instantiation, as well as static filters that need to be added to reach row.
  */
 export class SyncStreamsCompiler {
   readonly output = new CompiledStreamQueries();

packages/sync-rules/src/compiler/ir_to_sync_plan.ts

@@ -1,6 +1,5 @@
 import * as plan from '../sync_plan/plan.js';
 import * as resolver from './bucket_resolver.js';
-import { equalsIgnoringResultSet } from './compatibility.js';
 import { CompiledStreamQueries } from './compiler.js';
 import { Equality, HashMap, StableHasher, unorderedEquality } from './equality.js';
 import { ColumnInRow, SyncExpression } from './expression.js';

packages/sync-rules/src/compiler/querier_graph.ts

@@ -88,6 +88,31 @@ export class QuerierGraphBuilder {
  * This works by first processing subterms related to the source set selected from. If the subterm is a row expression,
  * we add it as a filter to the row processor or parameter lookup. If it is a match expression, we create a partition
  * key and obtain the value by recursively applying this algorithm to resolve the other side.
+ *
+ * To visualize this algorithm, consider the following example query:
+ *
+ * ```SQL
+ * SELECT * FROM comments c, issues i, users u
+ *  WHERE u.user_id = auth.user_id() AND c.issue_id = i.id AND u.id = i.owner_id AND u.is_admin
+ * ```
+ *
+ * First, {@link resolvePrimaryInput} is called, which calls {@link resolveResultSet} on `comments c`. While resolving a
+ * result set, we extract conditions mentioning that result set. In this case, the only such expression is
+ * `c.issue_id = i.id`. Because this is an {@link EqualsClause}, we know we need to introduce a parameter (in this case,
+ * `c.issue_id` because that's the half depending on the current result set). We then look at the other half and
+ * recursively resolve `issues i` (via {@link resolvePointLookup}). When we're done resolving that, we add `i.id` as to
+ * {@link PendingExpandingLookup.usedOutputs}.
+ *
+ * To resolve `issues i`, we extract the only remaining expression mentioning it, `u.id = i.owner_id`. We once again
+ * recursve to resolve `users u` and will add `u.id` as a used output.
+ *
+ * Finally, we find `u.user_id = auth.user_id()` and `u.is_admin`. The first expression creates a parameter, but doesn't
+ * need to resolve any further result sets since the input depends on connection data. The second expression only
+ * depends on the row itself, so we add it as a static condition to only create parameter lookups for rows matching that
+ * condition.
+ *
+ * This algorithm gives us the bucket creator as well as parameter lookups with their partition keys and values, which
+ * is the sync plan.
  */
 class PendingQuerierPath {
   // Terms in the And that have not yet been handled.

packages/sync-rules/src/sync_plan/serialize.ts

@@ -15,6 +15,13 @@ import {
   SyncPlan
 } from './plan.js';
 
+/**
+ * Serializes a sync plan into a simple JSON object.
+ *
+ * While {@link SyncPlan}s are already serializable for the most part, it contains a graph of references from e.g.
+ * queriers to bucket creators. To represent this efficiently, we assign numbers to referenced elements while
+ * serializing instead of duplicating definitions.
+ */
 export function serializeSyncPlan(plan: SyncPlan): SerializedSyncPlanUnstable {
   const dataSourceIndex = new Map<StreamDataSource, number>();
   const bucketIndex = new Map<StreamBucketDataSource, number>();

packages/sync-rules/test/src/compiler/errors.test.ts

@@ -2,6 +2,15 @@ import { describe, expect, test } from 'vitest';
 import { compilationErrorsForSingleStream } from './utils.js';
 
 describe('compilation errors', () => {
+  test('not a select statement', () => {
+    expect(compilationErrorsForSingleStream("INSERT INTO users (id) VALUES ('foo')")).toStrictEqual([
+      {
+        message: 'Expected a SELECT statement',
+        source: "INSERT INTO users (id) VALUES ('foo'"
+      }
+    ]);
+  });
+
   test('IN operator with static left clause', () => {
     expect(
       compilationErrorsForSingleStream("SELECT * FROM issues WHERE 'static' IN (SELECT id FROM users WHERE is_admin)")

packages/sync-rules/test/src/compiler/utils.ts

@@ -47,10 +47,6 @@ export function compilationErrorsForSingleStream(...sql: string[]): TranslationE
   ])[0];
 }
 
-export function compilationErrors(inputs: SyncStreamInput[]): TranslationError[] {
-  return compileToSyncPlan(inputs)[0];
-}
-
 export function compileToSyncPlan(inputs: SyncStreamInput[]): [TranslationError[], SyncPlan] {
   const compiler = new SyncStreamsCompiler();
   const errors: TranslationError[] = [];