From 69dba7fd9bfd8685590b2ecc5423eaee78b71280 Mon Sep 17 00:00:00 2001 From: Dennis Felsing Date: Mon, 3 Aug 2026 06:32:35 +0000 Subject: [PATCH] compute: reject a negative record count in accumulable reduce `AccumulableErrorCheck` accumulates a record count per key, and `finalize_accum` returns that count directly for `count(*)`. The check only ever compared the count against zero: one arm caught a net-zero count paired with a non-zero accumulation, and another caught negative accumulation for unsigned sums. A NEGATIVE count passed both, so it reached `finalize_accum` unexamined and surfaced as a user-visible negative number with no error and no log line. A materialized view over such an aggregate then persisted the nonsense value with a positive multiplicity, making it durable in a derived shard and indistinguishable from a real result. A well formed collection has non-negative multiplicities, so a negative count means the input is already corrupt. Report it as an error instead. The check is per key rather than per aggregate, because the record count is a property of the input records, not of any one aggregate. This is defense in depth, not a root cause fix: it only fires once something upstream has already driven a collection negative. Its value is turning silently wrong answers into a diagnosable error, which is exactly what has been missing when invalid retractions show up in production. Note the check is inherently per key. A global aggregate accumulates every record under one empty key, so an unmatched retraction there is absorbed into a positive total and stays invisible. Only grouped aggregates are covered. Extends test/testdrive/avro-cdcv2.td with a regression case that injects an unmatched retraction through `ENVELOPE MATERIALIZE`, the only source envelope that passes upstream diffs through verbatim, and asserts both that a grouped count errors and that the error propagates into a materialized view over it. A positive control covers real retractions, so the new arm cannot fire on well formed input. Co-Authored-By: Claude Opus 5 (1M context) --- .../negative_accumulation_errors.py | 1 + src/compute/src/render/reduce.rs | 23 ++++++ test/testdrive/avro-cdcv2.td | 74 +++++++++++++++++++ test/testdrive/negative-multiplicities.td | 12 ++- 4 files changed, 106 insertions(+), 4 deletions(-) diff --git a/misc/python/materialize/parallel_workload/negative_accumulation_errors.py b/misc/python/materialize/parallel_workload/negative_accumulation_errors.py index 715ce5ba2704e..c4e95db3744c7 100644 --- a/misc/python/materialize/parallel_workload/negative_accumulation_errors.py +++ b/misc/python/materialize/parallel_workload/negative_accumulation_errors.py @@ -35,6 +35,7 @@ "Non-positive accumulation", "Invalid negative unsigned aggregation in ReduceAccumulable", "saw negative accumulation", + "saw negative record count", # Peek handling "Invalid data in source, saw retractions", "index peek encountered negative multiplicities in ok trace", diff --git a/src/compute/src/render/reduce.rs b/src/compute/src/render/reduce.rs index 3d8425ad36ec3..624b3a4b275fe 100644 --- a/src/compute/src/render/reduce.rs +++ b/src/compute/src/render/reduce.rs @@ -1523,6 +1523,29 @@ impl<'scope, T: RenderTimestamp> Context<'scope, T> { "AccumulableErrorCheck", move |key, input, output| { let (ref accums, total) = input[0].1; + + // A well formed collection has non-negative multiplicities, so a negative + // record count means the input is already corrupt. This is checked per key + // rather than per aggregate, because it is a property of the input records. + // + // NOTE: none of the per-aggregate checks below cover this. The net-zero check + // tests `total == 0` specifically, and the unsigned check inspects + // accumulators rather than the count. Left unchecked, `finalize_accum` hands + // `total` straight to `AggregateFunc::Count`, so `count(*)` returns a negative + // number as an ordinary result with no error and no log. + if total.is_negative() { + error_logger.log( + "Negative record count in ReduceAccumulable", + &format!("total={total}"), + ); + let key = key.to_row(); + let message = format!( + "Invalid data in source, saw negative record count for key {key} \ + in accumulable aggregate" + ); + output.push((EvalError::Internal(message.into()).into(), Diff::ONE)); + } + for (aggr, accum) in err_full_aggrs.iter().zip_eq(accums) { // We first test here if inputs without net-positive records are present, // producing an error to the logs and to the query output if that is the case. diff --git a/test/testdrive/avro-cdcv2.td b/test/testdrive/avro-cdcv2.td index 2b848a0487b96..df5e401a35d99 100644 --- a/test/testdrive/avro-cdcv2.td +++ b/test/testdrive/avro-cdcv2.td @@ -159,6 +159,80 @@ id price -------- 6 10 +# Regression: the accumulable reduce must reject a negative record count rather +# than returning it as an aggregate result. +# +# `AccumulableErrorCheck` accumulates a record count per key and `finalize_accum` +# returns it directly for `count(*)`. The check used to compare that count +# against zero only, so a NEGATIVE count passed every arm and surfaced as a +# user-visible negative number with no error and no log, and an MV over such an +# aggregate persisted the nonsense value with a positive multiplicity. +# +# ENVELOPE MATERIALIZE is the injection primitive here because it is the only +# source envelope that passes upstream diffs through verbatim. The check itself +# is source-agnostic. +# +# NOTE: the check fires on a key whose own record count is negative, so what it +# catches depends on the grouping. A global aggregate puts every record under one +# empty key, so an unmatched retraction is masked whenever the rest of the +# collection outweighs it, and is caught only once the whole collection goes +# net-negative. Grouping isolates the corrupt record in its own key, which is +# what makes it visible here regardless of the other keys. +# +# A separate topic and source keep the injected corruption away from +# data_schema_inline_tbl, which the assertions below still read. + +$ kafka-create-topic topic=negative + +> BEGIN +> CREATE SOURCE negative_multiplicity + IN CLUSTER ${arg.single-replica-cluster} + FROM KAFKA CONNECTION kafka_conn (TOPIC 'testdrive-negative-${testdrive.seed}') + +> CREATE TABLE negative_multiplicity_tbl + FROM SOURCE negative_multiplicity (REFERENCE "testdrive-negative-${testdrive.seed}") + FORMAT AVRO USING SCHEMA '${schema}' + ENVELOPE MATERIALIZE +> COMMIT + +# Positive control: real retractions must NOT trip the check. id=1 is retracted +# and reinserted at time 2. + +$ kafka-ingest format=avro topic=negative schema=${schema} +{"array":[{"data":{"id":1,"price":{"int":10}},"time":1,"diff":1}]} +{"array":[{"data":{"id":2,"price":{"int":20}},"time":1,"diff":1}]} +{"array":[{"data":{"id":1,"price":{"int":10}},"time":2,"diff":-1}]} +{"array":[{"data":{"id":1,"price":{"int":11}},"time":2,"diff":1}]} +{"com.materialize.cdc.progress":{"lower":[0],"upper":[3],"counts":[{"time":1,"count":2},{"time":2,"count":2}]}} + +> SELECT id, count(*) FROM negative_multiplicity_tbl GROUP BY id +id count +-------- +1 1 +2 1 + +# Inject an unmatched retraction, giving key id=99 a net record count of -1. + +$ kafka-ingest format=avro topic=negative schema=${schema} +{"array":[{"data":{"id":99,"price":{"int":999}},"time":3,"diff":-1}]} +{"com.materialize.cdc.progress":{"lower":[3],"upper":[10],"counts":[{"time":3,"count":1}]}} + +! SELECT id, count(*) FROM negative_multiplicity_tbl GROUP BY id +contains:saw negative record count + +# The error must propagate into a materialized view over that aggregate rather +# than the negative value becoming durable in the derived shard. + +> CREATE MATERIALIZED VIEW negative_multiplicity_counts + IN CLUSTER ${arg.single-replica-cluster} AS + SELECT id, count(*) AS n FROM negative_multiplicity_tbl GROUP BY id + +! SELECT * FROM negative_multiplicity_counts +contains:saw negative record count + +> DROP MATERIALIZED VIEW negative_multiplicity_counts +> DROP SOURCE negative_multiplicity CASCADE + # Test that tails report progress messages even without new data # The ouput of SUBSCRIBE is dependent on the replica size diff --git a/test/testdrive/negative-multiplicities.td b/test/testdrive/negative-multiplicities.td index 7987b380898e2..49fdfe4320480 100644 --- a/test/testdrive/negative-multiplicities.td +++ b/test/testdrive/negative-multiplicities.td @@ -19,16 +19,20 @@ ALTER SYSTEM SET enable_repeat_row = true > INSERT INTO base VALUES (1, 1, -1), (1, 1, -1) -> SELECT count(*) FROM data --1 +# The whole collection is now net-negative, so the empty key of the global +# aggregate carries a negative record count. An accumulable aggregate rejects +# that rather than returning it, which for count(*) would be a bare negative +# number with no error. +! SELECT count(*) FROM data +contains:saw negative record count ! SELECT * FROM data contains:Invalid data in source, saw retractions (1) for row that does not exist: [Int64 > INSERT INTO base VALUES (1, 1, -1) -> SELECT count(*) FROM data --2 +! SELECT count(*) FROM data +contains:saw negative record count ! SELECT * FROM data contains:Invalid data in source, saw retractions (2) for row that does not exist: [Int64