coord: fix propagation of negative multiplicities error

src/coord/coord.rs

@@ -1054,10 +1054,14 @@ where
                             match (memo, resp) {
                                 (PeekResponse::Rows(mut memo), PeekResponse::Rows(rows)) => {
                                     memo.extend(rows);
-                                    let out: Result<_, comm::Error> = Ok(PeekResponse::Rows(memo));
-                                    future::ready(out)
+                                    future::ok(PeekResponse::Rows(memo))
+                                }
+                                (PeekResponse::Error(e), _) | (_, PeekResponse::Error(e)) => {
+                                    future::ok(PeekResponse::Error(e))
+                                }
+                                (PeekResponse::Canceled, _) | (_, PeekResponse::Canceled) => {
+                                    future::ok(PeekResponse::Canceled)
                                 }
-                                _ => future::ok(PeekResponse::Canceled),
                             }
                         })
                         .map_ok(move |mut resp| {

test/testdrive/negative-multiplicities.td

@@ -0,0 +1,59 @@
+# Copyright Materialize, Inc. All rights reserved.
+#
+# Use of this software is governed by the Business Source License
+# included in the LICENSE file at the root of this repository.
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0.
+
+$ set schema={
+    "type": "record",
+    "name": "envelope",
+    "fields": [
+      {
+        "name": "before",
+        "type": [
+          {
+            "name": "row",
+            "type": "record",
+            "fields": [
+              {"name": "a", "type": "long"}
+            ]
+          },
+          "null"
+        ]
+      },
+      { "name": "after", "type": ["row", "null"] }
+    ]
+  }
+
+$ kafka-ingest format=avro topic=data schema=${schema} timestamp=1
+{"before": null, "after": {"a": 1}}
+
+> CREATE MATERIALIZED SOURCE data
+  FROM KAFKA BROKER '${testdrive.kafka-addr}' TOPIC 'testdrive-data-${testdrive.seed}'
+  FORMAT AVRO USING SCHEMA '${schema}'
+  ENVELOPE DEBEZIUM
+
+> SELECT * FROM data
+1
+
+$ kafka-ingest format=avro topic=data schema=${schema} timestamp=1
+{"before": {"a": 1}, "after": null}
+{"before": {"a": 1}, "after": null}
+
+> SELECT count(*) FROM data
+-1
+
+! SELECT * FROM data
+Negative multiplicity: -1
+
+$ kafka-ingest format=avro topic=data schema=${schema} timestamp=1
+{"before": {"a": 1}, "after": null}
+
+> SELECT count(*) FROM data
+-2
+
+! SELECT * FROM data
+Negative multiplicity: -2