feldera · mihaibudiu · Jul 3, 2026 · Jun 22, 2026 · Jun 22, 2026 · Jun 24, 2026
diff --git a/crates/dbsp/src/operator.rs b/crates/dbsp/src/operator.rs
@@ -22,6 +22,7 @@ mod csv;
 mod delta0;
 mod differentiate;
 mod generator;
+mod integral_weight_validator;
 mod integrate;
 mod neg;
 mod output;

diff --git a/crates/dbsp/src/operator/integral_weight_validator.rs b/crates/dbsp/src/operator/integral_weight_validator.rs
@@ -0,0 +1,149 @@
+//! Integral operator that panics when any record's accumulated weight
+//! becomes negative.
+
+use crate::{
+    ZWeight,
+    circuit::{Circuit, Stream, checkpointer::Checkpoint},
+    trace::{BatchReader as TraceBatchReader, Cursor as TraceCursor},
+    typed_batch::{BatchReader as TypedBatchReader, Spine, ZSet},
+};
+
+impl<C, B> Stream<C, B>
+where
+    C: Circuit,
+    B: Checkpoint + Clone + ZSet + 'static,
+    B::InnerBatch: Send,
+{
+    /// Integrates the stream, panicking if any record's weight in the integral
+    /// becomes negative.
+    ///
+    /// Maintains a running trace of the input stream.
+    /// After every clock tick, every key that appears in the
+    /// current delta is checked: if the key's weight in the trace is
+    /// negative, the operator panics, displaying the offending key, weight,
+    /// and a custom message.
+    ///
+    /// This operator is intended for debugging: it verifies that a stream
+    /// behaves like a valid set or multiset where every element has positive
+    /// multiplicity.
+    ///
+    /// The operator passes the input delta through unchanged as its output.
+    ///
+    /// # Panics
+    ///
+    /// Panics when any weight in the integral is < 0.
+    #[track_caller]
+    pub fn integral_weight_validator(&self, message: &str) -> Self {
+        let message = message.to_owned();
+        self.circuit().region("integral_weight_validator", || {
+            let sharded = self.shard();
+            let trace = sharded.integrate_trace();
+            sharded.apply2(&trace, move |delta: &B, integrated: &Spine<B>| {
+                let mut delta_cursor = TraceBatchReader::cursor(delta.inner());
+                let mut trace_cursor = TraceBatchReader::cursor(integrated.inner());
+                while delta_cursor.key_valid() {
+                    // Only examine keys that appear in the current delta with a negative weight
+                    let delta_weight: ZWeight = **delta_cursor.weight();
+                    if delta_weight < 0 {
+                        let key = delta_cursor.key();
+                        if trace_cursor.seek_key_exact(key, None) {
+                            debug_assert!(trace_cursor.val_valid());
+                            let weight: ZWeight = **trace_cursor.weight();
+                            if weight < 0 {
+                                panic!("{message}: negative weight found {weight} for key {key:?}");
+                            }
+                        }
+                    }
+                    delta_cursor.step_key();
+                }
+            });
+            self.clone()
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::{
+        Circuit, Runtime, ZWeight, operator::Generator, typed_batch::OrdZSet, utils::Tup2, zset,
+    };
+
+    /// A stream of purely positive deltas should pass the check at every step.
+    #[test]
+    fn integral_weight_validator_positive_weights() {
+        let (mut circuit, ()) = Runtime::init_circuit(1, move |circuit| {
+            let mut step = 0u64;
+            let source = circuit.add_source(Generator::new(move || {
+                let batch = zset! { step => 1 };
+                step += 1;
+                batch
+            }));
+            let out = source.integral_weight_validator("test_table");
+            let mut expected_step = 0u64;
+            out.inspect(move |delta| {
+                assert_eq!(delta, &zset! { expected_step => 1 });
+                expected_step += 1;
+            });
+            Ok(())
+        })
+        .unwrap();
+
+        for _ in 0..10 {
+            circuit.transaction().unwrap();
+        }
+        circuit.kill().unwrap();
+    }
+
+    /// Inserting an element once and then deleting it twice causes the trace
+    /// weight to go negative, which must trigger a panic.
+    #[test]
+    fn integral_weight_validator_panics_on_negative_integral() {
+        let (mut circuit, ()) = Runtime::init_circuit(1, move |circuit| {
+            let mut step = 0i64;
+            let source = circuit.add_source(Generator::new(move || {
+                // Step 0: insert key 42 with weight +1 → weight 1.
+                // Step 1: delete key 42 with weight -2 → weight -1 (panic).
+                let batch: OrdZSet<i64> = if step == 0 {
+                    OrdZSet::from_tuples((), vec![Tup2(Tup2(42i64, ()), 1 as ZWeight)])
+                } else {
+                    OrdZSet::from_tuples((), vec![Tup2(Tup2(42i64, ()), -2 as ZWeight)])
+                };
+                step += 1;
+                batch
+            }));
+            source.integral_weight_validator("table 'test'");
+            Ok(())
+        })
+        .unwrap();
+
+        circuit.transaction().unwrap(); // step 0: weight 1 (ok)
+        let err = circuit.transaction().expect_err("expected a panic");
+        assert!(
+            err.to_string()
+                .contains("table 'test': negative weight found"),
+            "unexpected error: {err}"
+        );
+        circuit.kill().unwrap();
+    }
+
+    /// A weight that goes negative (e.g., deleting something never inserted)
+    /// must also panic.
+    #[test]
+    fn integral_weight_validator_panics_on_negative_weight() {
+        let (mut circuit, ()) = Runtime::init_circuit(1, move |circuit| {
+            // Emit weight -1 immediately
+            let source = circuit.add_source(Generator::new(|| zset! { 42i64 => -1 }));
+            source.integral_weight_validator("table 'test'");
+            Ok(())
+        })
+        .unwrap();
+
+        let err = circuit.transaction().expect_err("expected a panic");
+        assert!(
+            err.to_string()
+                .contains("table 'test': negative weight found"),
+            "unexpected error: {err}"
+        );
+        circuit.kill().unwrap();
+    }
+}
diff --git a/docs.feldera.com/docs/sql/grammar.md b/docs.feldera.com/docs/sql/grammar.md
@@ -790,6 +790,16 @@ FELDERA_USE_MULTI_JOINS = OFF`.
 
 Currently the following options are available:
 
+`ENFORCE_POSITIVE_INPUTS` if set to `TRUE`, the compiler inserts a
+runtime check after every input table that has no primary key.  At
+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 debugging pipelines that
+produce unexpected results due to invalid input data.  Note: this
+setting is expensive, and will trigger a full backfill for the
+affected tables.
+
 `FELDERA_AVOID_STAR_JOINS` if set to true the compiler will not
 generate code using the new feature of "star joins", which are a
 simple form of multi-joins.  We expect that option will be removed in

diff --git a/...iler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWeightValidatorOperator.java b/...iler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWeightValidatorOperator.java
@@ -0,0 +1,55 @@
+package org.dbsp.sqlCompiler.circuit.operator;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import org.dbsp.sqlCompiler.circuit.OutputPort;
+import org.dbsp.sqlCompiler.compiler.backend.JsonDecoder;
+import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteEmptyRel;
+import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteRelNode;
+import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision;
+import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor;
+import org.dbsp.sqlCompiler.ir.expression.DBSPExpression;
+import org.dbsp.sqlCompiler.ir.type.DBSPType;
+import org.dbsp.util.Utilities;
+
+import javax.annotation.Nullable;
+import java.util.List;
+
+/** Operator that panics at runtime if the running integral of its input stream
+ * contains any record with a non-positive weight. */
+public final class DBSPWeightValidatorOperator extends DBSPUnaryOperator {
+    /** Panic message. */
+    public final String message;
+
+    public DBSPWeightValidatorOperator(CalciteRelNode node, OutputPort source, String message) {
+        super(node, "integral_weight_validator", null, source.outputType(), source.isMultiset(), source);
+        this.message = message;
+    }
+
+    @Override
+    public void accept(CircuitVisitor visitor) {
+        visitor.push(this);
+        VisitDecision decision = visitor.preorder(this);
+        if (!decision.stop())
+            visitor.postorder(this);
+        visitor.pop(this);
+    }
+
+    @Override
+    public DBSPSimpleOperator with(
+            @Nullable DBSPExpression function, DBSPType outputType,
+            List<OutputPort> newInputs, boolean force) {
+        if (this.mustReplace(force, function, newInputs, outputType)) {
+            return new DBSPWeightValidatorOperator(
+                    this.getRelNode(), newInputs.get(0), this.message).copyAnnotations(this);
+        }
+        return this;
+    }
+
+    @SuppressWarnings("unused")
+    public static DBSPWeightValidatorOperator fromJson(JsonNode node, JsonDecoder decoder) {
+        CommonInfo info = commonInfoFromJson(node, decoder);
+        String message = Utilities.getStringProperty(node, "message");
+        return new DBSPWeightValidatorOperator(CalciteEmptyRel.INSTANCE, info.getInput(0), message)
+                .addAnnotations(info.annotations(), DBSPWeightValidatorOperator.class);
+    }
+}
diff --git a/...sp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/ProgramMetadata.java b/...sp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/ProgramMetadata.java
@@ -45,7 +45,8 @@ private static String canonicalVariableName(String variable) {
 
     static final Set<String> reserved = Set.of(
             DBSPCompiler.WARNINGS_ARE_ERRORS.toLowerCase(Locale.ENGLISH),
-            ProgramMetadata.AVOID_STAR_JOINS.toLowerCase(Locale.ENGLISH)
+            ProgramMetadata.AVOID_STAR_JOINS.toLowerCase(Locale.ENGLISH),
+            ProgramMetadata.ENFORCE_POSITIVE_INPUTS.toLowerCase(Locale.ENGLISH)
     );
 
     private boolean known(String variable) {
@@ -110,11 +111,20 @@ public boolean isExplicitlyOff(String variable) {
     }
 
     public static final String AVOID_STAR_JOINS = "FELDERA_AVOID_STAR_JOINS";
+    /** When set to {@code true}, inserts a weight-validation check after every
+     * input table that has no primary key. */
+    public static final String ENFORCE_POSITIVE_INPUTS = "ENFORCE_POSITIVE_INPUTS";
 
     public boolean noStarJoins() {
         return this.isExplicitlyOn(AVOID_STAR_JOINS);
     }
 
+    /** Returns {@code true} if weight validation should be inserted after
+     * inputs without a primary key. */
+    public boolean enforcePositiveInputs() {
+        return this.isExplicitlyOn(ENFORCE_POSITIVE_INPUTS);
+    }
+
     /** True if a feature is not set or set to 'false' */
     public boolean isOffOrMissing(String variable) {
         if (!this.hasValue(variable))

diff --git a/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java b/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java
@@ -24,6 +24,7 @@
 import org.dbsp.sqlCompiler.circuit.operator.DBSPViewBaseOperator;
 import org.dbsp.sqlCompiler.circuit.operator.DBSPWindowOperator;
 import org.dbsp.sqlCompiler.circuit.operator.DBSPIntegrateTraceRetainValuesOperator;
+import org.dbsp.sqlCompiler.circuit.operator.DBSPWeightValidatorOperator;
 import org.dbsp.sqlCompiler.compiler.DBSPCompiler;
 import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.ProgramIdentifier;
 import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision;
@@ -241,6 +242,15 @@ public VisitDecision preorder(DBSPSourceMultisetOperator operator) {
         return this.preorder(operator.to(DBSPSourceTableOperator.class));
     }
 
+    @Override
+    public VisitDecision preorder(DBSPWeightValidatorOperator operator) {
+        if (this.preorder(operator.to(DBSPSimpleOperator.class)).stop())
+            return VisitDecision.STOP;
+        this.property("message");
+        this.stream.append(operator.message);
+        return VisitDecision.CONTINUE;
+    }
+
     @Override
     public VisitDecision preorder(DBSPSourceMapOperator operator) {
         if (this.preorder(operator.to(DBSPSourceTableOperator.class)).stop())

diff --git a/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java b/.../SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java
@@ -968,6 +968,24 @@ VisitDecision retainOperator(DBSPBinaryOperator operator) {
         return VisitDecision.STOP;
     }
 
+    @Override
+    public VisitDecision preorder(DBSPWeightValidatorOperator operator) {
+        this.computeHash(operator);
+        DBSPType streamType = this.streamType(operator);
+        this.writeComments(operator)
+                .append("let ")
+                .append(operator.getNodeName(this.preferHash))
+                .append(": ");
+        streamType.accept(this.innerVisitor);
+        this.builder.append(" = ")
+                .append(this.getInputName(operator, 0))
+                .append(".integral_weight_validator(")
+                .append(Utilities.doubleQuote(operator.message, false))
+                .append(");");
+        this.tagStream(operator);
+        return VisitDecision.STOP;
+    }
+
     @Override
     public VisitDecision preorder(DBSPInternOperator operator) {
         this.writeComments(operator)

diff --git a/...piler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitCloneVisitor.java b/...piler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitCloneVisitor.java
@@ -313,6 +313,11 @@ public void postorder(DBSPIntegrateOperator operator) {
         this.replace(operator);
     }
 
+    @Override
+    public void postorder(DBSPWeightValidatorOperator operator) {
+        this.replace(operator);
+    }
+
     @Override
     public void postorder(DBSPUpsertFeedbackOperator operator) {
         this.replace(operator);

diff --git a/...compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java b/...compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java
@@ -55,7 +55,7 @@ public CircuitOptimizer(DBSPCompiler compiler) {
         super("Optimizer", compiler);
         this.createOptimizer();
     }
-    
+
     static class StopOnError extends CircuitVisitor {
         public StopOnError(DBSPCompiler compiler) {
             super(compiler);
@@ -125,6 +125,8 @@ void createOptimizer() {
         this.add(new OptimizeWithGraph(compiler, g -> new CloneOperatorsWithFanout(compiler, g)));
         this.add(new LinearPostprocessRetainKeys(compiler));
         this.add(new ExpandIndexedInputs(compiler));
+        this.add(new Conditional(compiler, new InsertWeightValidation(compiler),
+                this.compiler.metadata::enforcePositiveInputs));
         this.add(new OptimizeWithGraph(compiler, g -> new FilterJoinVisitor(compiler, g)));
         this.add(new DeadCode(compiler, true));
         this.add(new Simplify(compiler).circuitRewriter(true));

diff --git a/...L-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitVisitor.java b/...L-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitVisitor.java
@@ -302,6 +302,10 @@ public VisitDecision preorder(DBSPIntegrateOperator node) {
         return this.preorder((DBSPUnaryOperator) node);
     }
 
+    public VisitDecision preorder(DBSPWeightValidatorOperator node) {
+        return this.preorder((DBSPUnaryOperator) node);
+    }
+
     public VisitDecision preorder(DBSPUpsertFeedbackOperator node) {
         return this.preorder((DBSPUnaryOperator) node);
     }
@@ -677,6 +681,10 @@ public void postorder(DBSPIntegrateOperator node) {
         this.postorder((DBSPUnaryOperator) node);
     }
 
+    public void postorder(DBSPWeightValidatorOperator node) {
+        this.postorder((DBSPUnaryOperator) node);
+    }
+
     public void postorder(DBSPUpsertFeedbackOperator node) {
         this.postorder((DBSPUnaryOperator) node);
     }

diff --git a/...er/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/InsertWeightValidation.java b/...er/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/InsertWeightValidation.java
@@ -0,0 +1,22 @@
+package org.dbsp.sqlCompiler.compiler.visitors.outer;
+
+import org.dbsp.sqlCompiler.circuit.operator.DBSPSourceMultisetOperator;
+import org.dbsp.sqlCompiler.circuit.operator.DBSPWeightValidatorOperator;
+import org.dbsp.sqlCompiler.compiler.DBSPCompiler;
+
+/** Inserts a {@link DBSPWeightValidatorOperator} after every
+ * {@link DBSPSourceMultisetOperator} (input tables without a primary key). */
+public class InsertWeightValidation extends CircuitCloneVisitor {
+    public InsertWeightValidation(DBSPCompiler compiler) {
+        super(compiler, false);
+    }
+
+    @Override
+    public void postorder(DBSPSourceMultisetOperator node) {
+        super.replace(node);
+        DBSPWeightValidatorOperator validator = new DBSPWeightValidatorOperator(
+                node.getRelNode(), this.mapped(node.outputPort()), "Table " + node.tableName.name());
+        this.remap.put(node.outputPort(), validator.outputPort());
+        this.addOperator(validator);
+    }
+}