[DBSP][SQL] Debugging tools to detect invalid change streams at runtime#6523
Conversation
|
@ryzhyk can you please review the DBSP part? It's a very simple operator |
| each step, the check verifies that no record in the accumulated | ||
| integral has a negative weight (i.e., the table has not received more | ||
| deletions for a key than insertions). A violation causes the pipeline | ||
| to panic. This option is intended for |
There was a problem hiding this comment.
Is a panic the best option? Would it be a nicer experience for the user to raise a warning instead?
There was a problem hiding this comment.
Is a panic the best option? Would it be a nicer experience for the user to raise a warning instead?
The truth is that there's nothing we can do at this point that makes sense. Computing over collections with negative elements is not something that databases do. Ideally we would stop the pipeline, but there is no such API.
mythical-fred
left a comment
There was a problem hiding this comment.
Useful debugging aid. The shape is right: the new integral_weight_validator operator on the DBSP side, the SQL variable ENFORCE_POSITIVE_INPUTS driving an InsertWeightValidation rewriter that injects a DBSPWeightValidatorOperator after every DBSPSourceMultisetOperator, ToRustVisitor / ToJsonOuterVisitor / CircuitCloneVisitor / CircuitVisitor all wired with the right visit method. The Conditional dispatch in CircuitOptimizer.createOptimizer() keys off metadata::enforcePositiveInputs so zero-cost when the flag is off.
A few real points:
- Doc/code mismatch on weight semantics (
crates/dbsp/src/operator/integral_weight_validator.rs). The rustdoc says "if any(key, value, weight)triple in the integral has a non-positive weight, the operator panics" and the# Panicssection says "Panics when any weight in the integral is ≤ 0." But the code assertsweight >= 0— i.e. panics on< 0, acceptsweight == 0. In Z-set integration after compactionweight == 0rows shouldn't surface from the cursor, so the operational behavior is fine, but the doc is wrong. Fix the doc to say "negative" and "< 0", and either drop the per-stepstep_val()weight check forweight == 0(it can't happen) or assertweight > 0and document that. integral_weight_validatorbuilds a full integrate.sharded.integrate()materializes the running multiset of the entire table in memory. For an input table that grows to millions of rows this is the cost the user pays for the debugging guarantee. The rustdoc and the grammar.md entry both say "intended for debugging and testing" which is good, but neither says "this operator builds a full integral and is unsuitable for large tables." Worth a one-line warning in both places so users aren't surprised when their pipeline OOMs after flipping the flag in production. (Power of Ten #2 — unbounded resource use — is a useful framing for the doc warning.)- anand's panic-vs-warning question (line 798 of grammar.md). I think Mihai's call here is right: a panic is the only correct behavior when the invariant is broken, because once a key's integral weight goes negative the multiset is inconsistent and any downstream materialized view that depends on it is now producing wrong results. A "warning" that logs and keeps going would leak garbage data into outputs. The honest user-experience answer is: don't enable
ENFORCE_POSITIVE_INPUTSin production; do enable it in CI for the exact test that's misbehaving; the panic is a debugging signal, not a runtime fault. That story should be stated explicitly in grammar.md so the next user with the same question doesn't have to re-derive it. Suggested addition: "This option is intended for debugging only — enabling it in production will crash the pipeline when the invariant is broken, which is the correct behavior for catching upstream bugs but not what you want for live workloads." - Test
integral_weight_validator_panics_on_negative_integralhas a great comment block. The companionintegral_weight_validator_panics_on_negative_weighttest name is slightly misleading — it's not really "the weight went negative", it's "step 0 inserts a delete delta directly", which makes the integral negative on the first step. Worth a one-line comment on that test mirroring the other one. InsertWeightValidationonly wrapsDBSPSourceMultisetOperator, which matches "tables without a primary key." Good. Worth a one-line comment inInsertWeightValidation.javastating the invariant (PK-bearing tables go throughDBSPSourceSetOperatororDBSPSourceMapOperator, which already reject illegal updates upstream) so a future reader doesn't wonder why only multiset sources get the check.DBSPWeightValidatorOperator.fromJsonrebuilds withCalciteEmptyRel.INSTANCE— matches the pattern in other operators, no concern. TheaddAnnotationsround-trip looks correct.Regression3Tests.testEnforcePositiveInputs. Good: SQL-level smoke test that the flag drives the actual rewriter + runtime check. Worth also: (a) a test asserting the flag is off by default (so a no-flagREMOVE FROM T VALUES (1)againstCREATE TABLE T(x INT NOT NULL)does not panic and produces the "wrong" output silently — pinning the opt-in semantics), and (b) a test for a table with a primary key, asserting thatENFORCE_POSITIVE_INPUTS = TRUEdoesn't insert the validator and the existing PK-based rejection path still runs (i.e. confirming the multiset-only narrowing inInsertWeightValidationis intentional).SET ENFORCE_POSITIVE_INPUTS = TRUE. What happens if a user writesSET ENFORCE_POSITIVE_INPUTS = OFFafter the table is created and then flips to ON? That's a compile-time setting, not runtime, so the question is moot for one program — but worth a one-line note that this is a compile-time setting (consistent with the otherFELDERA_*variables documented above the new entry).
Overall direction is right. Address the doc/code mismatch (≤ 0 vs < 0) before merging; the rest are non-blocking.
| impl<C, B> Stream<C, B> | ||
| where | ||
| C: Circuit, | ||
| B: Checkpoint |
There was a problem hiding this comment.
There seems to be a bunch of redundancy in this trait bound.
| + HasZero | ||
| + SizeOf | ||
| + NumEntries | ||
| + IndexedZSetReader |
There was a problem hiding this comment.
If you change this to ZSetReader, then you don't need to iterate over values below (there's always only one () value for each key).
| #[track_caller] | ||
| pub fn integral_weight_validator(&self, message: &str) -> Self { | ||
| let sharded = self.shard(); | ||
| let integral = sharded.integrate(); |
There was a problem hiding this comment.
integrate will merge everything in one giant batch instead of a spine. Use integrate_trace instead.
| sharded.apply2(&integral, move |delta, integrated| { | ||
| let inner = integrated.inner(); | ||
| let mut cursor = TraceBatchReader::cursor(inner); | ||
| for (key, _val, _w) in delta.iter() { |
There was a problem hiding this comment.
you should use cursor here as well, iter() is only good for debugging
| while cursor.val_valid() { | ||
| let weight: ZWeight = **cursor.weight(); | ||
| let val = cursor.val(); | ||
| assert!( |
There was a problem hiding this comment.
This checks that the current weight is >=0. You should instead check that the new weight (weight + _w) is >=0.
There was a problem hiding this comment.
Actually this should be correct, since it's the weight in the integral.
If this was wrong the tests would fail.
| deletions for a key than insertions). A violation causes the pipeline | ||
| to panic. This option is intended for | ||
| debugging pipelines that produce unexpected results due to invalid | ||
| input data. |
There was a problem hiding this comment.
Adding this attribute will probably require backfiling the tables and any views that depend on them from scratch. I would warn the user about this.
…er an internal collection ever has negative weights and panics Signed-off-by: Mihai Budiu <mbudiu@feldera.com>
…QL variable ENFORCE_POSITIVE_INPUTS Signed-off-by: Mihai Budiu <mbudiu@feldera.com>
Signed-off-by: Mihai Budiu <mbudiu@feldera.com>
|
@ryzhyk I have pushed a new commit which hopefully addresses all your comments. |
mythical-fred
left a comment
There was a problem hiding this comment.
Re-review on 919f9fe. All my prior points addressed cleanly:
- Doc/code mismatch fixed — rustdoc and
# Panicsboth now say "< 0" and "negative", and the implementation drops the inner per-value loop in favor of a singleweight < 0check on the key in the integrated trace. Consistent now. - Cost warning in grammar.md added: "this setting is expensive, and will trigger a full backfill for the affected tables." Concise and accurate.
- Hot-path tightening — the new version filters on
delta_weight < 0before doing the trace lookup, so positive deltas pay only a delta-cursor step. Nice. - Wrapping in
circuit.region("integral_weight_validator", ...)gives the integral subgraph a stable label for tracing/profiling — small but good. - Switching from
RootCircuit::buildtoRuntime::init_circuit(1, ...)+circuit.transaction().expect_err(...)for the negative tests is the right pattern;#[should_panic]against multi-threaded runtimes is fragile.
Non-blocking nits, take or leave:
integral_weight_validator_panics_on_negative_weighttest name still describes the symptom slightly indirectly — the key only has one delta, weight -1, so trace weight after step 0 is -1. Maybe rename to..._panics_on_delete_without_insertto mirror the negative-integral test's narrative, but truly optional.- The
debug_assert!(trace_cursor.val_valid())afterseek_key_exactreturning true is correct (a key with no valid val shouldn't exist in a sharded integrated trace), but a one-line comment explaining the invariant would help a future reader who hasn't internalized the trace cursor contract.
APPROVE.
Fixes #6156
There are two separate commits, one for DBSP and one for the SQL compiler.
A table with primary keys will reject illegal updates, but it is possible to create a table without a primary key and delete a row before inserting it. This will produce an illegal collection, and the behavior of the SQL program is undefined. This PR introduces a SQL variable which will insert additional circuitry for all tables without primary keys to detect dynamically such violations. It is intended as a debugging aid.
Describe Manual Test Plan
Ran the new tests manually
Checklist