Merge branch 'master' into wihl-time-travel-docs

crates/agent/src/discovers/specs.rs

@@ -19,6 +19,10 @@ pub fn parse_response(
     for binding in &mut bindings {
         binding.recommended_name = normalize_recommended_name(&binding.recommended_name);
     }
+    if bindings.iter().any(|b| b.resource_path.is_empty()) {
+        tracing::warn!(%image_name, %image_tag,
+            "connector discovered response omits resource_path, this is OK for now but will become an error in a future release");
+    }
 
     Ok((
         models::CaptureEndpoint::Connector(models::ConnectorConfig {

crates/derive/src/combine_api.rs

@@ -139,6 +139,15 @@ impl cgo::Service for API {
                 };
 
                 let spec = doc::combine::Spec::with_one_binding(
+                    // Piggy-back on the `schema_inference` knob to understand
+                    // if we should use full or partial reductions.
+                    // TODO(johnny): This is a hack to avoid churning this API further,
+                    // while we transition over to the `runtime` crate which is replacing this code.
+                    if infer_schema_json.is_empty() {
+                        true // This is a materialization (full reductions).
+                    } else {
+                        false // This is a capture (partial reductions).
+                    },
                     key_ex.clone(),
                     None,
                     new_validator(&schema_json)?,
@@ -302,11 +311,7 @@ pub fn drain_chunk(
     let target_length = target_length + arena.len();
 
     while let Some(drained) = drainer.next() {
-        let doc::combine::DrainedDoc {
-            binding: _, // Always zero.
-            reduced,
-            root,
-        } = drained?;
+        let doc::combine::DrainedDoc { meta, root } = drained?;
 
         if let Some(ref mut shape) = shape {
             if shape.widen_owned(&root) {
@@ -322,7 +327,7 @@ pub fn drain_chunk(
 
         // Only now do we know the actual length of the document in its serialized form.
         stats.increment(arena.len() - begin);
-        if reduced {
+        if meta.front() {
             cgo::send_bytes(Code::DrainedReducedDocument as u32, begin, arena, out);
         } else {
             cgo::send_bytes(Code::DrainedCombinedDocument as u32, begin, arena, out);
@@ -424,6 +429,9 @@ pub mod test {
                 false,
                 json!({"key": "really really really really really really really really long", "min": 10, "max": 10}),
             ),
+            (false, json!({"key": "one", "min": 7, "max": 7.7})),
+            (false, json!({"key": "one", "min": 8, "max": 8.8})),
+            (false, json!({"key": "one", "min": 9, "max": 9.9})),
         ] {
             svc.invoke(
                 if *left {
@@ -453,16 +461,16 @@ pub mod test {
 
         assert_eq!(out.len(), 1 * 3);
 
-        // Poll again to drain the final three, plus stats.
+        // Poll again to drain the final five, plus stats.
         svc.invoke(
             Code::DrainChunk as u32,
-            &(1024 as u32).to_be_bytes(),
+            &(4096 as u32).to_be_bytes(),
             &mut arena,
             &mut out,
         )
         .unwrap();
 
-        assert_eq!(out.len(), 4 * 3 + 1);
+        assert_eq!(out.len(), 6 * 3 + 1);
 
         // The last message in out should be stats
         let stats_out = out.pop().expect("missing stats");
@@ -473,12 +481,12 @@ pub mod test {
         let expected_stats = Stats {
             left: Some(DocsAndBytes { docs: 2, bytes: 62 }),
             right: Some(DocsAndBytes {
-                docs: 4,
-                bytes: 183,
+                docs: 7,
+                bytes: 276,
             }),
             out: Some(DocsAndBytes {
-                docs: 4,
-                bytes: 339,
+                docs: 6,
+                bytes: 491,
             }),
         };
         assert_eq!(expected_stats, stats_message);

crates/derive/src/snapshots/derive__combine_api__test__combine_api.snap

@@ -3,7 +3,7 @@ source: crates/derive/src/combine_api.rs
 expression: "(String::from_utf8_lossy(&arena[..stats_out.begin as usize]), out)"
 ---
 (
-    "{\"foo\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\",\"key\":\"one\",\"max\":5.5,\"min\":3}\u{2}one\0\u{15}\u{3}!�\u{16}\0\0\0\0\0\0{\"foo\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\",\"key\":\"really really really really really r\",\"max\":10,\"min\":10}\u{2}really really really really really r\0\u{15}\n\u{15}\n{\"foo\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\",\"key\":\"three\",\"max\":6.6,\"min\":6}\u{2}three\0\u{15}\u{6}!�\u{1a}ffffff{\"foo\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\",\"key\":\"two\",\"max\":4.4,\"min\":2}\u{2}two\0\u{15}\u{2}!�\u{11}������",
+    "{\"foo\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\",\"key\":\"one\",\"max\":3.3,\"min\":3}\u{2}one\0\u{15}\u{3}!�\nffffff{\"foo\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\",\"key\":\"one\",\"max\":9.9,\"min\":5}\u{2}one\0\u{15}\u{5}!�#������{\"foo\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\",\"key\":\"really really really really really r\",\"max\":10,\"min\":10}\u{2}really really really really really r\0\u{15}\n\u{15}\n{\"foo\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\",\"key\":\"three\",\"max\":6.6,\"min\":6}\u{2}three\0\u{15}\u{6}!�\u{1a}ffffff{\"foo\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\",\"key\":\"two\",\"max\":2.2,\"min\":2}\u{2}two\0\u{15}\u{2}!�\u{1}������{\"foo\":\"DocUUIDPlaceholder-329Bb50aa48EAa9ef\",\"key\":\"two\",\"max\":4.4,\"min\":4}\u{2}two\0\u{15}\u{4}!�\u{11}������",
     [
         Out {
             code: 202,
@@ -23,47 +23,77 @@ expression: "(String::from_utf8_lossy(&arena[..stats_out.begin as usize]), out)"
         Out {
             code: 201,
             begin: 92,
-            end: 201,
+            end: 168,
         },
         Out {
             code: 203,
-            begin: 201,
-            end: 239,
+            begin: 168,
+            end: 173,
         },
         Out {
             code: 204,
-            begin: 239,
-            end: 243,
+            begin: 173,
+            end: 184,
         },
         Out {
             code: 201,
-            begin: 243,
-            end: 321,
+            begin: 184,
+            end: 293,
         },
         Out {
             code: 203,
-            begin: 321,
-            end: 328,
+            begin: 293,
+            end: 331,
         },
         Out {
             code: 204,
-            begin: 328,
-            end: 339,
+            begin: 331,
+            end: 335,
         },
         Out {
-            code: 202,
-            begin: 339,
-            end: 415,
+            code: 201,
+            begin: 335,
+            end: 413,
         },
         Out {
             code: 203,
-            begin: 415,
+            begin: 413,
             end: 420,
         },
         Out {
             code: 204,
             begin: 420,
             end: 431,
         },
+        Out {
+            code: 202,
+            begin: 431,
+            end: 507,
+        },
+        Out {
+            code: 203,
+            begin: 507,
+            end: 512,
+        },
+        Out {
+            code: 204,
+            begin: 512,
+            end: 523,
+        },
+        Out {
+            code: 201,
+            begin: 523,
+            end: 599,
+        },
+        Out {
+            code: 203,
+            begin: 599,
+            end: 604,
+        },
+        Out {
+            code: 204,
+            begin: 604,
+            end: 615,
+        },
     ],
 )

crates/doc/Cargo.toml

@@ -36,6 +36,7 @@ allocator = { path = "../allocator" }
 
 hexdump = { workspace = true }
 insta = { workspace = true }
+json-patch = { workspace = true }
 pretty_assertions = { workspace = true }
 quickcheck = { workspace = true }
 quickcheck_macros = { workspace = true }