Fix SingleTimeMonotonic to take into account FlatMap's TableFunc's preserves_monotonicity

Author: ggevay

src/compute-types/src/plan/interpret/physically_monotonic.rs

@@ -139,12 +139,10 @@ impl Interpreter for SingleTimeMonotonic<'_> {
         _input_key: &Option<Vec<MirScalarExpr>>,
         input: Self::Domain,
         _exprs: &Vec<MirScalarExpr>,
-        _func: &TableFunc,
+        func: &TableFunc,
         _mfp: &MapFilterProject,
     ) -> Self::Domain {
-        // In a single-time context, we just propagate the monotonicity
-        // status of the input
-        input
+        PhysicallyMonotonic(input.0 && func.preserves_monotonicity())
     }
 
     fn join(

test/sqllogictest/transform/relax_must_consolidate.slt

@@ -1217,8 +1217,14 @@ COMPLETE 0
 statement ok
 CREATE SOURCE mono_src FROM WEBHOOK BODY FORMAT TEXT;
 
-# Regression test for https://github.com/MaterializeInc/database-issues/issues/11310
+# Regression test for https://github.com/MaterializeInc/database-issues/issues/11310, i.e., that we set the
+# `must_consolidate` flag when `repeat_row` is involved.
 # Relies on RelaxMustConsolidate's SingleTimeMonotonic correctly handling FlatMap's TableFunc being !preserves_monotonicity.
+# Note that the plans are still somewhat wrong, in that monotonic operators won't produce correct results when presented
+# with negative accumulations even with the `must_consolidate` flag set. However, the expectation is that they'll have
+# an easier time detecting negative accumulations and cleanly erroring out when the `must_consolidate` flag is set.
+# See also https://github.com/MaterializeInc/materialize/pull/36184 for more discussion, where we considered not
+# choosing monotonic operators when `repeat_row` is involved, but eventually decided to go with the current behavior.
 query T multiline
 EXPLAIN PHYSICAL PLAN AS VERBOSE TEXT FOR
 SELECT MAX(body) FROM mono_src, repeat_row(length(body));
@@ -1229,6 +1235,7 @@ Explained Query:
       Reduce::Hierarchical
         aggr_funcs=[max]
         monotonic
+        must_consolidate
         key_plan
           project=()
         val_plan=id