diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py
index c585e9bccb09a..0ba029322c50f 100644
--- a/misc/python/materialize/parallel_workload/action.py
+++ b/misc/python/materialize/parallel_workload/action.py
@@ -424,14 +424,30 @@ def errors_to_ignore(self, exe: Executor) -> list[str]:
         Oid,
     ) + tuple(RANGE_TYPES)
 
-    def aggregate_fns(self, column: Column) -> list[str]:
+    def aggregate_fns(self, column: Column, window: bool = False) -> list[str]:
         """Aggregate function templates valid for the column's type.
 
         Used both in window position (OVER ..) and in GROUP BY position. The
         collection aggregates (array_agg/list_agg/jsonb_agg/string_agg)
         exercise the "collection" reduce rendering, distinct from the
         accumulable sum/count path. Type exclusions are empirically derived,
-        e.g. array_agg rejects char and cannot nest map/list/array."""
+        e.g. array_agg rejects char and cannot nest map/list/array.
+
+        TODO: Reenable when CPU-200 is fixed.
+
+        `window` drops the collection aggregates. In GROUP BY position they
+        emit one collected value per group, which is linear in the input. In
+        window position every row of a partition receives an aggregate over
+        the whole partition, so the reduce's output arrangement holds N rows
+        of O(N) bytes each and a single partition of N rows costs O(N^2) on
+        the replica. Nothing bounds that. `LIMIT` lands in the peek's
+        `Finish`, above the dataflow that already built the whole collection,
+        and `max_result_size` only measures the final peek result. Views
+        whose partition-key column is a literal put every row in one
+        partition, and CDC source tables carry up to
+        `MySqlSource.prepopulate_rows` rows rather than `MAX_ROWS`, so N
+        reaches the tens of thousands. See the `window-collection-aggregate`
+        scenario in test/bounded-memory."""
         dt = column.data_type
         fns = ["COUNT({})"]
         if dt in NUMBER_TYPES:
@@ -450,6 +466,9 @@ def aggregate_fns(self, column: Column) -> list[str]:
             fns.extend(["BOOL_AND({})", "BOOL_OR({})"])
         if dt not in self._MINMAX_EXCLUDED:
             fns.extend(["MAX({})", "MIN({})"])
+        if window:
+            # TODO: Reenable when CPU-200 is fixed.
+            return fns
         # Collection aggregates.
         fns.append("jsonb_agg({})")
         if dt != Char:
@@ -653,7 +672,7 @@ def where_clause() -> str:
                 column1 = self.rng.choice(all_columns)
                 column2 = self.rng.choice(all_columns)
                 column3 = self.rng.choice(all_columns)
-                window_fn = self.rng.choice(self.aggregate_fns(column1))
+                window_fn = self.rng.choice(self.aggregate_fns(column1, window=True))
                 select_list.append(
                     f"{window_fn.format(column1)} OVER (PARTITION BY {column2} ORDER BY {column3})"
                 )
diff --git a/misc/python/materialize/sqlancer.py b/misc/python/materialize/sqlancer.py
index fe2f28aaba0a0..bf0aab7cb1c1e 100644
--- a/misc/python/materialize/sqlancer.py
+++ b/misc/python/materialize/sqlancer.py
@@ -114,6 +114,7 @@
     r"must use value within",
     r"zero raised to a negative power is undefined",
     r"range type over",
+    r"at the beginning of a statement",
 ]