examples: fix derive-patterns examples

Author: psFried

examples/derive-patterns/README.md

@@ -2,17 +2,18 @@
 
 ## Where to accumulate?
 
-When building a derived collection, the central question is where
-accumulation will happen: within derivation registers, or within a
-materialized database? Both approaches can produce equivalent results,
-but they do it in very different ways.
+When building a derived collection, the central question is where accumulation
+will happen: within derivation state, or within an external database that you
+materialize into? Both approaches can produce equivalent results, but they do it
+in very different ways.
 
-### Accumulate in the Database
+### Accumulate in the external database
 
-To accumulate in the database, you'll define a collection having a reducible
-schema with a derivation that uses only "publish" lambdas and no registers.
-The Flow runtime uses the provided annotations to reduce new documents into
-the collection, and ultimately keep the materialized table up to date.
+To accumulate in the external database, you'll define a collection having a
+reducible schema with a stateless derivation. The derivation can be written
+in either SQL or Typescript, but for these examples we use Typescript. The
+Flow runtime uses the provided annotations to reduce new documents into the
+collection, and ultimately keep the materialized table up to date.
 
 A key insight is that the database is the _only_ stateful system in this
 scenario, and that Flow is making use of reductions in two places:
@@ -45,42 +46,43 @@ When materializing into a pub/sub topic, there _is_ no store to hold final value
 and Flow will publish delta states: each a partial update of the (unknown)
 final value.
 
-### Accumulate in Registers
+### Accumulate in derivation state
 
-Accumulating in registers involves a derivation that defines a reducible
-register schema, and uses "update" lambdas.
-Registers are arbitrary documents that can be shared and updated by the various
-transformations of a derivation. The Flow runtime allocates, manages, and scales
-durable storage for registers; you don't have to.
+Accumulating in derivation state involves a `sqlite` derivation having one
+or more tables, which are created by `migrations`. These tables can be shared
+and updated by the various transforms of the derivation. The Flow runtime
+transactionally persists modifications to these tables.
 
-When using registers, the typical pattern is to use reduction annotations
-within updates of the register, and to then publish last-write-wins "snapshots"
-of the fully reduced value.
+When using a stateful derivation, the typical pattern is to use `INSERT ... ON
+CONFLICT ...` to accumulate state in your tables, and then `SELECT` from those
+tables to emit the documents.
 
 Returning to our summing example:
 
-| Time | Register | Lambdas                             | Derived Document |
-| ---- | -------- | ----------------------------------- | ---------------- |
-| T0   | **0**    | update(2, 1, 2), publish(register)  | **5**            |
-| T1   | **5**    | update(-2, 1), publish(register)    | **4**            |
-| T2   | **4**    | update(3, -2, 1), publish(register) | **6**            |
-| T3   | **6**    | update()                            |
+| Time | sum table | Lambdas                          | Derived Document |
+| ---- | --------- | -------------------------------- | ---------------- |
+| T0   | **0**     | update(2, 1, 2), select sum ...  | **5**            |
+| T1   | **5**     | update(-2, 1), select sum ...    | **4**            |
+| T2   | **4**     | update(3, -2, 1), select sum ... | **6**            |
+| T3   | **6**     | update()                         |
 
-Register derivations are a great solution for materializations into non-
+Stateful derivations are a great solution for materializations into non-
 transactional stores, because the documents they produce can be applied
 multiple times without breaking correctness.
 
 They're also well suited for materializations that publish into pub/sub,
 as they can produce stand-alone updates of a fully-reduced value.
 
-### Example: Summing in DB vs Register
+Additionally, stateful derivations are the best way to perform inner joins and time-windowed joins.
+
+### Example: Summing in a stateless vs a stateful derivation
 
 See [summer.flow.yaml](summer.flow.yaml) for a simple example
-of summing counts in the database, vs in registers.
+of summing counts using both approaches.
 
 ## Types of Joins
 
-### Outer Join accumulated in Database
+### Outer Join using a stateless derivation
 
 Example of an outer join, which is reduced within a target database table.
 This join is "fully reactive": it updates with either source collection,
@@ -89,11 +91,11 @@ and reflects the complete accumulation of their documents on both sides.
 The literal documents written to the collection are combined delta states,
 reflecting changes on one or both sides of the join. These delta states
 are then fully reduced into the database table, and no other storage _but_
-the table is required by this example.
+the table being materialized into is required.
 
 See [join-outer-flow.yaml](join-outer-flow.yaml).
 
-### Inner Join accumulated in Registers
+### Inner Join using a stateful derivation
 
 Example of an inner join, which is reduced within the derivation's registers.
 This join is also "fully reactive", updating with either source collection,
@@ -108,7 +110,7 @@ join are matched.
 
 See [join-inner.flow.yaml](join-inner.flow.yaml).
 
-### One-sided Join accumulated in Registers
+### One-sided join using a stateful derivation
 
 Example of a one-sided join, which publishes a current LHS joined
 with an accumulated RHS.
@@ -118,13 +120,3 @@ paired with a reduced snapshot of the RHS accumulator at that time.
 
 See [join-one-sided.flow.yaml](join-one-sided.yaml).
 
-### Comparing Registers
-
-Suppose we want to take action based on how a register is changing.
-
-For example, suppose we want to detect "zero crossings" of a running sum,
-and then filter the source collection to those documents which caused the
-sum to cross from positive to negative (or vice versa).
-
-We can use the `previous` register value to do so.
-See [zero-crossing.flow.yaml](zero-crossing.flow.yaml).

examples/derive-patterns/flow.yaml

@@ -4,4 +4,3 @@ import:
   - join-one-sided.flow.yaml
   - join-outer.flow.yaml
   - summer.flow.yaml
-  - zero-crossing.flow.yaml

examples/derive-patterns/join-inner.flow.ts

@@ -9,26 +9,50 @@ collections:
       required: [Key]
     key: [/Key]
 
-    derivation:
-      register:
-        schema: schema.yaml#Join
-        initial: {}
-
-      transform:
-        fromInts:
-          source: { name: patterns/ints }
+    derive:
+      using:
+        sqlite:
+          migrations:
+            - |
+              create table join_state (
+                key text not null primary key,
+                -- Stores the left hand side of the join
+                lhs integer,
+                -- Stores the right hand side of the join, using a JSON array
+                -- since this is a one-to-many join
+                rhs json
+              );
+      transforms:
+        - name: fromInts
+          source: patterns/ints
           shuffle: { key: [/Key] }
-          update: { lambda: typescript }
-          publish: { lambda: typescript }
+          lambda: |
+            insert into join_state (key, lhs) values ($Key, $Int)
+            on conflict (key) do update set lhs = lhs + $Int;
+            -- now emit the joined result
+            SELECT JSON_OBJECT(
+              'Key', $Key,
+              'LHS', lhs,
+              'RHS', JSON(rhs)
+            )
+            FROM join_state
+            WHERE key = $Key AND lhs IS NOT NULL AND rhs IS NOT NULL;
 
-        fromStrings:
-          source: { name: patterns/strings }
+        - name: fromStrings
+          source: patterns/strings
           shuffle: { key: [/Key] }
-          update: { lambda: typescript }
-          publish: { lambda: typescript }
-      typescript:
-        module: join-inner.flow.ts
-
+          lambda: |
+            INSERT INTO join_state (key, rhs) VALUES ($Key, JSON_ARRAY($String))
+            ON CONFLICT (key) DO UPDATE SET rhs = JSON_INSERT(COALESCE(rhs, '[]'), '$[#]', $String);
+            -- now emit the joined result
+            SELECT JSON_OBJECT(
+              'Key', $Key,
+              'LHS', lhs,
+              'RHS', JSON(rhs)
+            )
+            FROM join_state
+            WHERE key = $Key AND lhs IS NOT NULL AND rhs IS NOT NULL;
+   
 tests:
   patterns/test/inner-join:
     - ingest:

examples/derive-patterns/join-inner.flow.yaml

@@ -9,6 +9,42 @@ collections:
       required: [Key]
     key: [/Key]
 
+    derive:
+      using:
+        sqlite:
+          migrations:
+            - |
+              create table join_state (
+                key text not null primary key,
+                -- Stores the left hand side of the join
+                lhs integer,
+                -- Stores the right hand side of the join, using a JSON array
+                -- since this is a one-to-many join
+                rhs json
+              );
+      transforms:
+        - name: fromInts
+          source: patterns/ints
+          shuffle: { key: [/Key] }
+          lambda: |
+            insert into join_state (key, lhs) values ($Key, $Int)
+            on conflict (key) do update set lhs = $Int;
+            -- now emit the joined result
+            SELECT JSON_OBJECT(
+              'Key', $Key,
+              'LHS', lhs,
+              'RHS', JSON(COALESCE(rhs, '[]'))
+            )
+            FROM join_state
+            WHERE key = $Key;
+
+        - name: fromStrings
+          source: patterns/strings
+          shuffle: { key: [/Key] }
+          lambda: |
+            INSERT INTO join_state (key, rhs) VALUES ($Key, JSON_ARRAY($String))
+            ON CONFLICT (key) DO UPDATE SET rhs = JSON_INSERT(COALESCE(rhs, '[]'), '$[#]', $String);
+
     derivation:
       register:
         schema: schema.yaml#Join

examples/derive-patterns/join-one-sided.flow.ts

@@ -1,11 +1,11 @@
-import { IDerivation, Document, Register, FromIntsSource, FromStringsSource } from 'flow/patterns/outer-join';
+import { IDerivation, Document, SourceFromInts, SourceFromStrings } from 'flow/patterns/outer-join.ts';
 
 // Implementation for derivation examples/derive-patterns/join-outer.flow.yaml#/collections/patterns~1outer-join/derivation.
-export class Derivation implements IDerivation {
-    fromIntsPublish(source: FromIntsSource, _register: Register, _previous: Register): Document[] {
-        return [{ Key: source.Key, LHS: source.Int }];
+export class Derivation extends IDerivation {
+    fromInts(read: {doc: SourceFromInts}): Document[] {
+        return [{ Key: read.doc.Key, LHS: read.doc.Int }];
     }
-    fromStringsPublish(source: FromStringsSource, _register: Register, _previous: Register): Document[] {
-        return [{ Key: source.Key, RHS: [source.String] }];
+    fromStrings(read: {doc: SourceFromStrings}): Document[] {
+        return [{ Key: read.doc.Key, RHS: [read.doc.String] }];
     }
 }

examples/derive-patterns/join-one-sided.flow.yaml

@@ -8,19 +8,17 @@ collections:
       required: [Key]
     key: [/Key]
 
-    derivation:
-      transform:
-        fromInts:
+    derive:
+      using:
+        typescript: { module: join-outer.flow.ts }
+      transforms:
+        - name: fromInts
           source: { name: patterns/ints }
           shuffle: { key: [/Key] }
-          publish: { lambda: typescript }
 
-        fromStrings:
+        - name: fromStrings
           source: { name: patterns/strings }
           shuffle: { key: [/Key] }
-          publish: { lambda: typescript }
-      typescript:
-        module: join-outer.flow.ts
 
 tests:
   patterns/test/outer-join:

examples/derive-patterns/join-outer.flow.ts

@@ -1,8 +1,8 @@
-import { IDerivation, Document, Register, FromIntsSource } from 'flow/patterns/sums-db';
+import { IDerivation, Document, SourceFromInts } from 'flow/patterns/sums-reductions.ts';
 
 // Implementation for derivation examples/derive-patterns/summer.flow.yaml#/collections/patterns~1sums-db/derivation.
-export class Derivation implements IDerivation {
-    fromIntsPublish(source: FromIntsSource, _register: Register, _previous: Register): Document[] {
-        return [{ Key: source.Key, Sum: source.Int }];
+export class Derivation extends IDerivation {
+    fromInts(read: { doc: SourceFromInts }): Document[] {
+        return [{ Key: read.doc.Key, Sum: read.doc.Int }];
     }
 }