Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 22 additions & 3 deletions misc/python/materialize/parallel_workload/action.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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:
Expand Down Expand Up @@ -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})"
)
Expand Down
1 change: 1 addition & 0 deletions misc/python/materialize/sqlancer.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
]


Expand Down
Loading