Skip to content

[DBSP][SQL] Debugging tools to detect invalid change streams at runtime#6523

Merged
mihaibudiu merged 3 commits into
feldera:mainfrom
mihaibudiu:issue6156
Jul 3, 2026
Merged

[DBSP][SQL] Debugging tools to detect invalid change streams at runtime#6523
mihaibudiu merged 3 commits into
feldera:mainfrom
mihaibudiu:issue6156

Conversation

@mihaibudiu

Copy link
Copy Markdown
Contributor

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

  • Unit tests added/updated
  • Documentation updated

@mihaibudiu

Copy link
Copy Markdown
Contributor Author

@ryzhyk can you please review the DBSP part? It's a very simple operator
@anandbraman please review the SQL test example to see how this is intended to be used.

@mihaibudiu
mihaibudiu requested review from anandbraman and ryzhyk June 22, 2026 22:56
Comment thread docs.feldera.com/docs/sql/grammar.md Outdated
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

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is a panic the best option? Would it be a nicer experience for the user to raise a warning instead?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 mythical-fred left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 # Panics section says "Panics when any weight in the integral is ≤ 0." But the code asserts weight >= 0 — i.e. panics on < 0, accepts weight == 0. In Z-set integration after compaction weight == 0 rows 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-step step_val() weight check for weight == 0 (it can't happen) or assert weight > 0 and document that.
  • integral_weight_validator builds 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_INPUTS in 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_integral has a great comment block. The companion integral_weight_validator_panics_on_negative_weight test 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.
  • InsertWeightValidation only wraps DBSPSourceMultisetOperator, which matches "tables without a primary key." Good. Worth a one-line comment in InsertWeightValidation.java stating the invariant (PK-bearing tables go through DBSPSourceSetOperator or DBSPSourceMapOperator, which already reject illegal updates upstream) so a future reader doesn't wonder why only multiset sources get the check.
  • DBSPWeightValidatorOperator.fromJson rebuilds with CalciteEmptyRel.INSTANCE — matches the pattern in other operators, no concern. The addAnnotations round-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-flag REMOVE FROM T VALUES (1) against CREATE 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 that ENFORCE_POSITIVE_INPUTS = TRUE doesn't insert the validator and the existing PK-based rejection path still runs (i.e. confirming the multiset-only narrowing in InsertWeightValidation is intentional).
  • SET ENFORCE_POSITIVE_INPUTS = TRUE. What happens if a user writes SET ENFORCE_POSITIVE_INPUTS = OFF after 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 other FELDERA_* 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

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There seems to be a bunch of redundancy in this trait bound.

+ HasZero
+ SizeOf
+ NumEntries
+ IndexedZSetReader

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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();

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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() {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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!(

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This checks that the current weight is >=0. You should instead check that the new weight (weight + _w) is >=0.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Actually this should be correct, since it's the weight in the integral.
If this was wrong the tests would fail.

Comment thread docs.feldera.com/docs/sql/grammar.md Outdated
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.

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>
@mihaibudiu

Copy link
Copy Markdown
Contributor Author

@ryzhyk I have pushed a new commit which hopefully addresses all your comments.

@mythical-fred mythical-fred left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Re-review on 919f9fe. All my prior points addressed cleanly:

  • Doc/code mismatch fixed — rustdoc and # Panics both now say "< 0" and "negative", and the implementation drops the inner per-value loop in favor of a single weight < 0 check 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 < 0 before 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::build to Runtime::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_weight test 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_insert to mirror the negative-integral test's narrative, but truly optional.
  • The debug_assert!(trace_cursor.val_valid()) after seek_key_exact returning 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.

@mihaibudiu
mihaibudiu requested a review from ryzhyk June 25, 2026 21:50
@mihaibudiu
mihaibudiu added this pull request to the merge queue Jul 2, 2026
@github-merge-queue
github-merge-queue Bot removed this pull request from the merge queue due to failed status checks Jul 2, 2026
@mihaibudiu
mihaibudiu added this pull request to the merge queue Jul 3, 2026
Merged via the queue into feldera:main with commit 31374e0 Jul 3, 2026
1 check passed
@mihaibudiu
mihaibudiu deleted the issue6156 branch July 3, 2026 22:50
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[DBSP][SQL] Detect negative rows in MATERIALIZED tables

4 participants