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