Unify negative-multiplicity errors into EvalError::MultiplicityError

Replace the scattered "internal error: ..." messages raised on
negative/non-positive/non-monotonic multiplicities and inconsistent
accumulators with a single structured EvalError variant,
MultiplicityError { kind, code_place, detail }.

The new Display does not use the "internal error:" prefix; it names the
likely root cause (invalid data in a Materialize source, caused either
by a bug in the external system feeding the source or by a bug in
Materialize) and mentions incorrect usage of the repeat_row table
function as a secondary cause.

Migrated sites:

* src/compute/src/render/reduce.rs (DistinctBy, ReduceInaccumulable,
  ReduceMinsMaxes, MinsMaxesHierarchical, hierarchical mins-maxes final,
  ReduceMonotonic, AccumulableErrorCheck net-zero + unsigned-sum).
* src/compute/src/render/top_k.rs (MonotonicTopK, TopK validator,
  MonotonicTop1).
* src/compute/src/compute_state.rs (persist peek, index peek error
  trace) and compute_state/peek_result_iterator.rs (index peek ok
  trace), using err.to_string() to feed PeekResponse::Error.
* src/adapter/src/coord/peek.rs and src/adapter/src/peek_client.rs
  constant-result fast paths: emit EvalError::MultiplicityError instead
  of a hand-rolled InvalidParameterValue / AdapterError::Unstructured.
* src/storage-operators/src/s3_oneshot_sink.rs: build and stringify
  a MultiplicityError, mirroring the [customer-data] logging contract
  inline (storage-operators cannot depend on compute).
* src/expr/src/relation/func.rs (scalar-subquery cardinality check):
  the old EvalError::NegativeRowsFromSubquery variant is removed.

Internal error reporting is unified: every MultiplicityError site logs
through log_multiplicity_error (free function in render/errors.rs, and
a thin forwarding method on ErrorLogger). All sites share the static
title "invalid record multiplicity", so they merge into a single Sentry
group; kind/code_place/detail go into a [customer-data]-tagged WARN
breadcrumb.

Additionally fix the reduce.rs monoid-variant mismatch site: the lhs/rhs
Row values no longer leak into the panic/Sentry title; a static title
helps with Sentry's issue merging.

Proto backcompat is not needed for EvalError::NegativeRowsFromSubquery
because it was introduced this week and hasn't shipped in any release;
its proto tag (83) is reused by the new ProtoMultiplicityError.

Co-authored-by: Junie <junie@jetbrains.com>

Author: ggevay

src/adapter/src/coord/peek.rs

@@ -676,9 +676,11 @@ impl crate::coord::Coordinator {
             let mut results = Vec::new();
             for (row, count) in rows {
                 if count.is_negative() {
-                    Err(EvalError::InvalidParameterValue(
-                        format!("Negative multiplicity in constant result: {}", count).into(),
-                    ))?
+                    Err(EvalError::MultiplicityError(mz_expr::MultiplicityError {
+                        kind: mz_expr::MultiplicityErrorKind::Negative,
+                        code_place: "constant result".into(),
+                        detail: format!("count={}", count).into(),
+                    }))?
                 };
                 if count.is_positive() {
                     let count = usize::cast_from(

src/adapter/src/peek_client.rs

@@ -265,9 +265,12 @@ impl PeekClient {
                 let count = match u64::try_from(count.into_inner()) {
                     Ok(u) => usize::cast_from(u),
                     Err(_) => {
-                        return Err(AdapterError::Unstructured(anyhow::anyhow!(
-                            "Negative multiplicity in constant result: {}",
-                            count
+                        return Err(AdapterError::Eval(mz_expr::EvalError::MultiplicityError(
+                            mz_expr::MultiplicityError {
+                                kind: mz_expr::MultiplicityErrorKind::Negative,
+                                code_place: "constant result".into(),
+                                detail: format!("count={}", count).into(),
+                            },
                         )));
                     }
                 };

src/compute/src/compute_state.rs

@@ -1345,15 +1345,17 @@ impl PersistPeek {
                 }
 
                 let count: usize = d.try_into().map_err(|_| {
-                    error!(
-                        shard = %metadata.data_shard, diff = d, ?row,
-                        "persist peek encountered negative multiplicities",
-                    );
-                    format!(
-                        "Invalid data in source, \
-                         saw retractions ({}) for row that does not exist: {:?}",
-                        -d, row,
-                    )
+                    let err = mz_expr::MultiplicityError {
+                        kind: mz_expr::MultiplicityErrorKind::Negative,
+                        code_place: "persist peek".into(),
+                        detail: format!(
+                            "shard={}, diff={}, row={:?}",
+                            metadata.data_shard, d, row,
+                        )
+                        .into(),
+                    };
+                    crate::render::errors::log_multiplicity_error(&err, "persist peek");
+                    err.to_string()
                 })?;
                 let Some(count) = NonZeroUsize::new(count) else {
                     continue;
@@ -1491,15 +1493,19 @@ impl IndexPeek {
             });
             if copies.is_negative() {
                 let error = cursor.key(&storage);
-                error!(
-                    target = %self.peek.target.id(), diff = %copies, %error,
-                    "index peek encountered negative multiplicities in error trace",
-                );
-                return PeekStatus::Ready(PeekResponse::Error(format!(
-                    "Invalid data in source errors, \
-                    saw retractions ({}) for row that does not exist: {}",
-                    -copies, error,
-                )));
+                let err = mz_expr::MultiplicityError {
+                    kind: mz_expr::MultiplicityErrorKind::Negative,
+                    code_place: "index peek error trace".into(),
+                    detail: format!(
+                        "target={}, diff={}, error={}",
+                        self.peek.target.id(),
+                        copies,
+                        error,
+                    )
+                    .into(),
+                };
+                crate::render::errors::log_multiplicity_error(&err, "index peek error trace");
+                return PeekStatus::Ready(PeekResponse::Error(err.to_string()));
             }
             if copies.is_positive() {
                 return PeekStatus::Ready(PeekResponse::Error(cursor.key(&storage).to_string()));

src/compute/src/compute_state/peek_result_iterator.rs

@@ -247,15 +247,17 @@ where
             });
             let copies: i64 = if copies.is_negative() {
                 let row = &*borrow;
-                tracing::error!(
-                    target = %self.target_id, diff = %copies, ?row,
-                    "index peek encountered negative multiplicities in ok trace",
-                );
-                return Err(format!(
-                    "Invalid data in source, \
-                             saw retractions ({}) for row that does not exist: {:?}",
-                    -copies, row,
-                ));
+                let err = mz_expr::MultiplicityError {
+                    kind: mz_expr::MultiplicityErrorKind::Negative,
+                    code_place: "index peek".into(),
+                    detail: format!(
+                        "target={}, diff={}, row={:?}",
+                        self.target_id, copies, row,
+                    )
+                    .into(),
+                };
+                crate::render::errors::log_multiplicity_error(&err, "index peek");
+                return Err(err.to_string());
             } else {
                 copies.into_inner()
             };

src/compute/src/render.rs

@@ -166,7 +166,7 @@ use crate::typedefs::{ErrBatcher, ErrBuilder, ErrSpine, KeyBatcher, MzTimestamp}
 
 pub mod context;
 pub(crate) mod continual_task;
-mod errors;
+pub(crate) mod errors;
 mod flat_map;
 mod join;
 mod reduce;

src/compute/src/render/errors.rs

@@ -9,8 +9,33 @@
 
 //! Helpers for handling errors encountered by operators.
 
+use mz_expr::MultiplicityError;
 use mz_repr::Row;
 
+/// Shared static title for every [`MultiplicityError`], so all occurrences
+/// merge into a single Sentry group regardless of kind or site.
+pub(crate) const MULTIPLICITY_ERROR_TITLE: &str = "invalid record multiplicity";
+
+/// Log a [`MultiplicityError`] using the standard "static title + [customer-data]
+/// tagged details" contract, without needing an [`ErrorLogger`] instance.
+///
+/// Use this from call sites that don't have an [`ErrorLogger`] in scope (peek
+/// paths, S3 oneshot sink, constant-result fast path in the coordinator).
+/// `context` is a short, static-ish string identifying where the logger is
+/// being called from (e.g. `"index peek"`, `"persist peek"`,
+/// `"S3 oneshot sink"`).
+pub(crate) fn log_multiplicity_error(err: &MultiplicityError, context: &str) {
+    tracing::warn!(
+        context = context,
+        "[customer-data] {} (kind={:?}, code_place={}, detail={})",
+        MULTIPLICITY_ERROR_TITLE,
+        err.kind,
+        err.code_place,
+        err.detail,
+    );
+    tracing::error!(MULTIPLICITY_ERROR_TITLE);
+}
+
 /// Used to make possibly-validating code generic: think of this as a kind of `MaybeResult`,
 /// specialized for use in compute.  Validation code will only run when the error constructor is
 /// Some.
@@ -87,6 +112,7 @@ impl ErrorLogger {
     /// the breadcrumbs to their associated error in Sentry.
     ///
     // TODO(database-issues#5362): Rethink or justify our error logging strategy.
+    #[allow(dead_code)] // kept for future non-multiplicity uses
     pub fn log(&self, message: &'static str, details: &str) {
         tracing::warn!(
             dataflow = self.dataflow_name,
@@ -105,4 +131,22 @@ impl ErrorLogger {
         );
         mz_ore::soft_panic_or_log!("{}", message);
     }
+
+    /// Log a [`MultiplicityError`] using the shared `MULTIPLICITY_ERROR_TITLE`
+    /// so that all multiplicity errors merge into a single Sentry group.
+    ///
+    /// The dynamic `kind`, `code_place`, and `detail` fields are emitted at
+    /// WARN level with the `[customer-data]` tag, alongside the dataflow
+    /// name of this `ErrorLogger`.
+    pub fn log_multiplicity_error(&self, err: &MultiplicityError) {
+        tracing::warn!(
+            dataflow = self.dataflow_name,
+            "[customer-data] {} (kind={:?}, code_place={}, detail={})",
+            MULTIPLICITY_ERROR_TITLE,
+            err.kind,
+            err.code_place,
+            err.detail,
+        );
+        tracing::error!(MULTIPLICITY_ERROR_TITLE);
+    }
 }