update all tests for new shuffle: any semantics

Fix a logic bug in the grouped-transfers example, which needs to shuffle
on /sender (and not transfer /id).

crates/agent/src/publications/specs.rs

@@ -688,6 +688,7 @@ mod test {
                                     "transforms": Array [
                                         Object {
                                             "name": String("my-name"),
+                                            "shuffle": String("any"),
                                             "source": String("usageB/CollectionA"),
                                         },
                                     ],

crates/agent/src/publications/test_resources/happy_path.sql

@@ -24,7 +24,8 @@ p4 as (
             "transforms":[
                 {
                   "name": "my-name",
-                  "source": "usageB/CollectionA"
+                  "source": "usageB/CollectionA",
+                  "shuffle": "any"
                 }
             ]
         }

examples/acmeBank.flow.yaml

@@ -119,6 +119,7 @@ collections:
             partitions:
               include:
                 outcome: ["approve"]
+          shuffle: any
           lambda: |
             -- Map approved transfers to a credit for the recipient and a debit for the sender.
             select $recipient as account, $amount as balance
@@ -193,4 +194,4 @@ tests:
         documents:
           - { account: alice, balance: 35 }
           - { account: bob, balance: 5 }
-          - { account: deposit, balance: -40 }
+          - { account: deposit, balance: -40 }

examples/citi-bike/last-seen.flow.yaml

@@ -18,4 +18,5 @@ collections:
       transforms:
         - name: locationFromRide
           source: { name: examples/citi-bike/rides }
+          shuffle: any
           lambda: select $bike_id, $end as last;

examples/citi-bike/rides.flow.yaml

@@ -10,3 +10,4 @@ collections:
       transforms:
         - name: fromCsvRides
           source: { name: examples/citi-bike/csv-rides }
+          shuffle: any

examples/citi-bike/stations.flow.yaml

@@ -14,5 +14,5 @@ collections:
           module: stations.flow.ts
       transforms:
         - name: ridesAndMoves
-          source:
-            name: examples/citi-bike/rides-and-relocations
+          source: examples/citi-bike/rides-and-relocations
+          shuffle: any

examples/stock-stats/flow.yaml

@@ -27,6 +27,7 @@ collections:
           source:
             name: stock/ticks
             partitions: { include: { exchange: [NYSE, NASDAQ] } }
+          shuffle: any
 
 tests:
   # TODO this is a pretty silly test, and is the result of my messing around

go/bindings/.snapshots/TestBuildCatalog-one-derivation

@@ -1 +1 @@
-(*flow.CollectionSpec)(name:"a/derivation" write_schema_json:"{\"$id\":\"file:///build.flow.yaml?ptr=/collections/a~1derivation/schema\",\"properties\":{\"a_key\":{\"type\":\"string\"}},\"required\":[\"a_key\"],\"type\":\"object\"}" key:"/a_key" uuid_ptr:"/_meta/uuid" projections:<ptr:"/a_key" field:"a_key" is_primary_key:true inference:<types:"string" string:<> exists:MUST > > projections:<field:"flow_document" inference:<types:"object" exists:MUST > > ack_template_json:"{\"_meta\":{\"ack\":true,\"uuid\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\"}}" partition_template:<name:"a/derivation" replication:3 labels:<labels:<name:"app.gazette.dev/managed-by" value:"estuary.dev/flow" > labels:<name:"content-type" value:"application/x-ndjson" > labels:<name:"estuary.dev/build" value:"fixture" > labels:<name:"estuary.dev/collection" value:"a/derivation" > > fragment:<length:536870912 compression_codec:GZIP stores:"s3://a-bucket/" refresh_interval:<seconds:300 > path_postfix_template:"utc_date={{.Spool.FirstAppendTime.Format \"2006-01-02\"}}/utc_hour={{.Spool.FirstAppendTime.Format \"15\"}}" > flags:4 max_append_rate:4194304 > derivation:<connector_type:TYPESCRIPT config_json:"{\"module\":\"import { IDerivation, Document, SourceSwizzle } from 'flow/a/derivation.ts';\\n\\nexport class Derivation extends IDerivation {\\n    swizzle(source: { doc: SourceSwizzle }): Document[] {\\n      const doc = source.doc;\\n      return [{a_key: doc.a_key.repeat(doc.a_val ? doc.a_val : 1)}];\\n    }\\n}\\n\"}" transforms:<name:"swizzle" collection:<name:"a/collection" write_schema_json:"{\"$id\":\"file:///build.flow.yaml?ptr=/collections/a~1collection/schema\",\"properties\":{\"a_key\":{\"type\":\"string\"},\"a_val\":{\"type\":\"integer\"}},\"required\":[\"a_key\"],\"type\":\"object\"}" key:"/a_key" uuid_ptr:"/_meta/uuid" projections:<ptr:"/a_key" field:"a_key" is_primary_key:true inference:<types:"string" string:<> exists:MUST > > projections:<ptr:"/a_val" field:"a_val" inference:<types:"integer" exists:MAY > > projections:<field:"flow_document" inference:<types:"object" exists:MUST > > ack_template_json:"{\"_meta\":{\"ack\":true,\"uuid\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\"}}" partition_template:<name:"a/collection" replication:3 labels:<labels:<name:"app.gazette.dev/managed-by" value:"estuary.dev/flow" > labels:<name:"content-type" value:"application/x-ndjson" > labels:<name:"estuary.dev/build" value:"fixture" > labels:<name:"estuary.dev/collection" value:"a/collection" > > fragment:<length:536870912 compression_codec:GZIP stores:"s3://a-bucket/" refresh_interval:<seconds:300 > path_postfix_template:"utc_date={{.Spool.FirstAppendTime.Format \"2006-01-02\"}}/utc_hour={{.Spool.FirstAppendTime.Format \"15\"}}" > flags:4 max_append_rate:4194304 > > partition_selector:<include:<labels:<name:"estuary.dev/collection" value:"a/collection" > > exclude:<> > lambda_config_json:"null" journal_read_suffix:"derive/a/derivation/swizzle" > shuffle_key_types:STRING shard_template:<id:"derivation/a/derivation" recovery_log_prefix:"recovery" hint_prefix:"/estuary/flow/hints" hint_backups:2 max_txn_duration:<seconds:1 > labels:<labels:<name:"app.gazette.dev/managed-by" value:"estuary.dev/flow" > labels:<name:"estuary.dev/build" value:"fixture" > labels:<name:"estuary.dev/log-level" value:"info" > labels:<name:"estuary.dev/task-name" value:"a/derivation" > labels:<name:"estuary.dev/task-type" value:"derivation" > > ring_buffer_size:65536 read_channel_size:131072 > recovery_log_template:<name:"recovery/derivation/a/derivation" replication:3 labels:<labels:<name:"app.gazette.dev/managed-by" value:"estuary.dev/flow" > labels:<name:"content-type" value:"application/x-gazette-recoverylog" > labels:<name:"estuary.dev/build" value:"fixture" > labels:<name:"estuary.dev/task-name" value:"a/derivation" > labels:<name:"estuary.dev/task-type" value:"derivation" > > fragment:<length:268435456 compression_codec:SNAPPY stores:"s3://a-bucket/" refresh_interval:<seconds:300 > > flags:4 max_append_rate:4194304 > > )
+(*flow.CollectionSpec)(name:"a/derivation" write_schema_json:"{\"$id\":\"file:///build.flow.yaml?ptr=/collections/a~1derivation/schema\",\"properties\":{\"a_key\":{\"type\":\"string\"}},\"required\":[\"a_key\"],\"type\":\"object\"}" key:"/a_key" uuid_ptr:"/_meta/uuid" projections:<ptr:"/a_key" field:"a_key" is_primary_key:true inference:<types:"string" string:<> exists:MUST > > projections:<field:"flow_document" inference:<types:"object" exists:MUST > > ack_template_json:"{\"_meta\":{\"ack\":true,\"uuid\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\"}}" partition_template:<name:"a/derivation" replication:3 labels:<labels:<name:"app.gazette.dev/managed-by" value:"estuary.dev/flow" > labels:<name:"content-type" value:"application/x-ndjson" > labels:<name:"estuary.dev/build" value:"fixture" > labels:<name:"estuary.dev/collection" value:"a/derivation" > > fragment:<length:536870912 compression_codec:GZIP stores:"s3://a-bucket/" refresh_interval:<seconds:300 > path_postfix_template:"utc_date={{.Spool.FirstAppendTime.Format \"2006-01-02\"}}/utc_hour={{.Spool.FirstAppendTime.Format \"15\"}}" > flags:4 max_append_rate:4194304 > derivation:<connector_type:TYPESCRIPT config_json:"{\"module\":\"import { IDerivation, Document, SourceSwizzle } from 'flow/a/derivation.ts';\\n\\nexport class Derivation extends IDerivation {\\n    swizzle(source: { doc: SourceSwizzle }): Document[] {\\n      const doc = source.doc;\\n      return [{a_key: doc.a_key.repeat(doc.a_val ? doc.a_val : 1)}];\\n    }\\n}\\n\"}" transforms:<name:"swizzle" collection:<name:"a/collection" write_schema_json:"{\"$id\":\"file:///build.flow.yaml?ptr=/collections/a~1collection/schema\",\"properties\":{\"a_key\":{\"type\":\"string\"},\"a_val\":{\"type\":\"integer\"}},\"required\":[\"a_key\"],\"type\":\"object\"}" key:"/a_key" uuid_ptr:"/_meta/uuid" projections:<ptr:"/a_key" field:"a_key" is_primary_key:true inference:<types:"string" string:<> exists:MUST > > projections:<ptr:"/a_val" field:"a_val" inference:<types:"integer" exists:MAY > > projections:<field:"flow_document" inference:<types:"object" exists:MUST > > ack_template_json:"{\"_meta\":{\"ack\":true,\"uuid\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\"}}" partition_template:<name:"a/collection" replication:3 labels:<labels:<name:"app.gazette.dev/managed-by" value:"estuary.dev/flow" > labels:<name:"content-type" value:"application/x-ndjson" > labels:<name:"estuary.dev/build" value:"fixture" > labels:<name:"estuary.dev/collection" value:"a/collection" > > fragment:<length:536870912 compression_codec:GZIP stores:"s3://a-bucket/" refresh_interval:<seconds:300 > path_postfix_template:"utc_date={{.Spool.FirstAppendTime.Format \"2006-01-02\"}}/utc_hour={{.Spool.FirstAppendTime.Format \"15\"}}" > flags:4 max_append_rate:4194304 > > partition_selector:<include:<labels:<name:"estuary.dev/collection" value:"a/collection" > > exclude:<> > lambda_config_json:"null" journal_read_suffix:"derive/a/derivation/swizzle" > shard_template:<id:"derivation/a/derivation" recovery_log_prefix:"recovery" hint_prefix:"/estuary/flow/hints" hint_backups:2 max_txn_duration:<seconds:1 > labels:<labels:<name:"app.gazette.dev/managed-by" value:"estuary.dev/flow" > labels:<name:"estuary.dev/build" value:"fixture" > labels:<name:"estuary.dev/log-level" value:"info" > labels:<name:"estuary.dev/task-name" value:"a/derivation" > labels:<name:"estuary.dev/task-type" value:"derivation" > > ring_buffer_size:65536 read_channel_size:131072 > recovery_log_template:<name:"recovery/derivation/a/derivation" replication:3 labels:<labels:<name:"app.gazette.dev/managed-by" value:"estuary.dev/flow" > labels:<name:"content-type" value:"application/x-gazette-recoverylog" > labels:<name:"estuary.dev/build" value:"fixture" > labels:<name:"estuary.dev/task-name" value:"a/derivation" > labels:<name:"estuary.dev/task-type" value:"derivation" > > fragment:<length:268435456 compression_codec:SNAPPY stores:"s3://a-bucket/" refresh_interval:<seconds:300 > > flags:4 max_append_rate:4194304 > > )

go/bindings/testdata/build.flow.yaml

@@ -30,8 +30,8 @@ collections:
             }
       transforms:
         - name: swizzle
-          source:
-            name: a/collection
+          source: a/collection
+          shuffle: any
 
 captures:
   example/capture:

go/flow/testdata/specs_test.flow.yaml

@@ -37,7 +37,8 @@ collections:
             }
       transforms:
         - name: swizzle
-          source: { name: example/collection }
+          source: example/collection
+          shuffle: any
       shards:
         # Expect this is reflected in shard specs.
         maxTxnDuration: 60s

go/shuffle/testdata/ab.flow.yaml

@@ -19,12 +19,12 @@ collections:
 
     derive:
       using:
-        sqlite:
-          migrations: []
+        sqlite: {}
       transforms:
         # Note that our test depends on this transform being validated as `readOnly`.
         - name: swizzle
           source: a/collection
+          shuffle: any
           lambda: SELECT 1;
 
 storageMappings:

ops-catalog/template-common.flow.yaml

@@ -77,5 +77,5 @@ collections:
           module: catalog-stats-rollup.ts
       transforms:
         - name: fromOps.us-central1.v1
-          source:
-            name: ops.us-central1.v1/catalog-stats-L1
+          source: ops.us-central1.v1/catalog-stats-L1
+          shuffle: any

site/docs/concepts/bank/balances.flow.yaml

@@ -14,3 +14,4 @@ collections:
             partitions:
               include:
                 outcome: [approve]
+          shuffle: any

site/docs/concepts/bank/grouped.flow.yaml

@@ -22,9 +22,11 @@ collections:
       transforms:
         - name: enrichAndAddToWindow
           source: acmeBank/transfers
+          shuffle: { key: [/sender] }
           lambda: enrichAndAddToWindow.sql
 
         - name: removeFromWindow
           source: acmeBank/transfers
+          shuffle: { key: [/sender] }
           readDelay: 24h
           lambda: DELETE FROM transfers WHERE id = $id;

site/docs/concepts/bank/last-large-send.flow.yaml

@@ -9,4 +9,5 @@ collections:
       transforms:
         - name: filterTransfers
           source: acmeBank/transfers
+          shuffle: any
           lambda: SELECT $id, $sender, $recipient, $amount WHERE $amount > 100;