From 86278d0871eb9f318de5481f9b51ae539c31c546 Mon Sep 17 00:00:00 2001
From: Mihai Budiu <mbudiu@feldera.com>
Date: Mon, 22 Jun 2026 15:45:17 -0700
Subject: [PATCH 1/3] [DBSP] Add DBSP operator integral_weight_validator which
 checks whether an internal collection ever has negative weights and panics

Signed-off-by: Mihai Budiu <mbudiu@feldera.com>
---
 crates/dbsp/src/operator.rs                   |   1 +
 .../src/operator/integral_weight_validator.rs | 152 ++++++++++++++++++
 2 files changed, 153 insertions(+)
 create mode 100644 crates/dbsp/src/operator/integral_weight_validator.rs

diff --git a/crates/dbsp/src/operator.rs b/crates/dbsp/src/operator.rs
index dc7f013f15c..19ff96d359a 100644
--- 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
new file mode 100644
index 00000000000..d97a2b5d619
--- /dev/null
+++ b/crates/dbsp/src/operator/integral_weight_validator.rs
@@ -0,0 +1,152 @@
+//! Integral operator that panics when any record's accumulated weight
+//! becomes negative.
+
+use crate::{
+    NumEntries, ZWeight,
+    algebra::{AddAssignByRef, AddByRef, HasZero},
+    circuit::{Circuit, Stream, checkpointer::Checkpoint},
+    dynamic::Erase,
+    trace::{BatchReader as TraceBatchReader, Cursor as TraceCursor},
+    typed_batch::{Batch, IndexedZSetReader},
+};
+use size_of::SizeOf;
+
+impl<C, B> Stream<C, B>
+where
+    C: Circuit,
+    B: Checkpoint
+        + AddByRef
+        + AddAssignByRef
+        + Clone
+        + Eq
+        + HasZero
+        + SizeOf
+        + NumEntries
+        + IndexedZSetReader
+        + Batch<Time = ()>
+        + 'static,
+    B::InnerBatch: Send,
+{
+    /// Integrates the stream, panicking if any record's weight in the integral
+    /// becomes negative.
+    ///
+    /// Maintains a running integral of the input stream.
+    /// After every clock tick, every key that appears in the
+    /// current delta is checked: if any `(key, value, weight)` triple in the
+    /// integral has a non-positive weight, the operator panics, displaying the
+    /// offending key, value, weight and a custom message.
+    ///
+    /// This operator is intended for debugging and testing: it verifies that a
+    /// stream behaves like a valid set or multiset where every element has a
+    /// 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 sharded = self.shard();
+        let integral = sharded.integrate();
+        let message = message.to_owned();
+        // Only examine keys that appear in the current delta — those are the
+        // only keys whose integral weight could have decreased this step.
+        sharded.apply2(&integral, move |delta, integrated| {
+            let inner = integrated.inner();
+            let mut cursor = TraceBatchReader::cursor(inner);
+            for (key, _val, _w) in delta.iter() {
+                if cursor.seek_key_exact(key.erase(), None) {
+                    while cursor.val_valid() {
+                        let weight: ZWeight = **cursor.weight();
+                        let val = cursor.val();
+                        assert!(
+                            weight >= 0,
+                            "{message}: negative weight found {weight} for key {key:?}, val {val:?}"
+                        );
+                        cursor.step_val();
+                    }
+                }
+            }
+        });
+        self.clone()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::{
+        Circuit, RootCircuit, 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 circuit = RootCircuit::build(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| {
+                // The output is the original delta, not the integral.
+                assert_eq!(delta, &zset! { expected_step => 1 });
+                expected_step += 1;
+            });
+            Ok(())
+        })
+        .unwrap()
+        .0;
+
+        for _ in 0..10 {
+            circuit.transaction().unwrap();
+        }
+    }
+
+    /// Inserting an element once and then deleting it twice causes the integral
+    /// weight to go negative, which must trigger a panic.
+    #[test]
+    #[should_panic(expected = "table 'test': negative weight found")]
+    fn integral_weight_validator_panics_on_negative_integral() {
+        let circuit = RootCircuit::build(move |circuit| {
+            let mut step = 0i64;
+            let source = circuit.add_source(Generator::new(move || {
+                // Step 0: insert key 42 with weight +1 → integral weight 1.
+                // Step 1: delete key 42 with weight -2 → integral 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()
+        .0;
+
+        circuit.transaction().unwrap(); // step 0: integral weight 1 (ok)
+        circuit.transaction().unwrap(); // step 1: integral weight -1 (panic)
+    }
+
+    /// A weight that goes negative
+    #[test]
+    #[should_panic(expected = "table 'test': negative weight found")]
+    fn integral_weight_validator_panics_on_negative_weight() {
+        let circuit = RootCircuit::build(move |circuit| {
+            // Emit weight -1 immediately; integral weight is -1 after step 0.
+            let source = circuit.add_source(Generator::new(|| zset! { 42i64 => -1 }));
+            source.integral_weight_validator("table 'test'");
+            Ok(())
+        })
+        .unwrap()
+        .0;
+
+        circuit.transaction().unwrap();
+    }
+}

From 4eb253b0ed39cafc6945a07a4cfd81654355b4fb Mon Sep 17 00:00:00 2001
From: Mihai Budiu <mbudiu@feldera.com>
Date: Mon, 22 Jun 2026 15:52:40 -0700
Subject: [PATCH 2/3] [SQL] Compiler support for integral_weight_validator,
 driven by the SQL variable ENFORCE_POSITIVE_INPUTS

Signed-off-by: Mihai Budiu <mbudiu@feldera.com>
---
 docs.feldera.com/docs/sql/grammar.md          |  9 +++
 .../operator/DBSPWeightValidatorOperator.java | 55 +++++++++++++++++++
 .../sqlCompiler/compiler/ProgramMetadata.java | 12 +++-
 .../compiler/backend/ToJsonOuterVisitor.java  | 10 ++++
 .../compiler/backend/rust/ToRustVisitor.java  | 18 ++++++
 .../visitors/outer/CircuitCloneVisitor.java   |  5 ++
 .../visitors/outer/CircuitOptimizer.java      |  4 +-
 .../visitors/outer/CircuitVisitor.java        |  8 +++
 .../outer/InsertWeightValidation.java         | 22 ++++++++
 .../compiler/sql/simple/Regression3Tests.java | 13 ++++-
 .../compiler/sql/tools/BaseSQLTests.java      | 11 ++++
 11 files changed, 164 insertions(+), 3 deletions(-)
 create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWeightValidatorOperator.java
 create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/InsertWeightValidation.java

diff --git a/docs.feldera.com/docs/sql/grammar.md b/docs.feldera.com/docs/sql/grammar.md
index f3f10e90ef8..fbe3993572a 100644
--- a/docs.feldera.com/docs/sql/grammar.md
+++ b/docs.feldera.com/docs/sql/grammar.md
@@ -790,6 +790,15 @@ 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.
+
 `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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWeightValidatorOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWeightValidatorOperator.java
new file mode 100644
index 00000000000..3c28439bbfb
--- /dev/null
+++ b/sql-to-dbsp-compiler/SQL-compiler/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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/ProgramMetadata.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/ProgramMetadata.java
index 7d4f55ae919..8d17e36f009 100644
--- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/ProgramMetadata.java
+++ b/sql-to-dbsp-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-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java
index aadcc399fef..c4a8ef31348 100644
--- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java
+++ b/sql-to-dbsp-compiler/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-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java
index 7423d93ff96..0539faa14a0 100644
--- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java
+++ b/sql-to-dbsp-compiler/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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitCloneVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitCloneVisitor.java
index ba617ffc260..9b484594977 100644
--- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitCloneVisitor.java
+++ b/sql-to-dbsp-compiler/SQL-compiler/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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java
index a2699458f99..ae31c39176d 100644
--- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java
+++ b/sql-to-dbsp-compiler/SQL-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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitVisitor.java
index ee3a8d9aca1..7d7067208c3 100644
--- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitVisitor.java
+++ b/sql-to-dbsp-compiler/SQL-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/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/InsertWeightValidation.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/InsertWeightValidation.java
new file mode 100644
index 00000000000..cf68711c272
--- /dev/null
+++ b/sql-to-dbsp-compiler/SQL-compiler/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);
+    }
+}
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression3Tests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression3Tests.java
index 1c09deec61e..53c5bd5554c 100644
--- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression3Tests.java
+++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression3Tests.java
@@ -7,7 +7,6 @@
 import org.dbsp.sqlCompiler.compiler.sql.tools.SqlIoTest;
 import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor;
 import org.junit.Assert;
-import org.junit.Ignore;
 import org.junit.Test;
 
 public class Regression3Tests extends SqlIoTest {
@@ -717,4 +716,16 @@ r ROW (b VARCHAR)
 
                 CREATE VIEW Z AS SELECT b, r.b, t.b FROM T;""");
     }
+
+    @Test
+    public void testEnforcePositiveInputs() {
+        String sql = """
+                SET ENFORCE_POSITIVE_INPUTS = TRUE;
+                CREATE TABLE T(x INT NOT NULL);
+                CREATE VIEW V AS SELECT x FROM T;""";
+        DBSPCompiler compiler = this.testCompiler();
+        compiler.submitStatementsForCompilation(sql);
+        this.runtimeFail(compiler, "REMOVE FROM T VALUES (1);",
+                "Table t: negative weight found");
+    }
 }
diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/BaseSQLTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/BaseSQLTests.java
index 80f07a29a7a..df24ed1d111 100644
--- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/BaseSQLTests.java
+++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/BaseSQLTests.java
@@ -447,6 +447,17 @@ protected void runtimeFail(String query, String message, InputOutputChangeStream
         new CompilerCircuitStream(compiler, data, this, message);
     }
 
+    /** Run a test that fails at runtime with a panic matching {@code message}.
+     *
+     * @param compiler compiler
+     * @param script   DML statements (INSERT/REMOVE) that trigger the panic
+     * @param message  expected substring of the panic message */
+    protected void runtimeFail(DBSPCompiler compiler, String script, String message) {
+        CompilerCircuitStream ccs = new CompilerCircuitStream(compiler, this, message);
+        Change input = ccs.toChange(script);
+        ccs.addPair(input, new Change());
+    }
+
     /** Run a test that fails at runtime without needing any inputs */
     protected void runtimeConstantFail(String query, String message) {
         this.runtimeFail(query, message, this.streamWithEmptyChanges());

From 919f9fe2abfb5d36776d6fa216e6276a844ae49b Mon Sep 17 00:00:00 2001
From: Mihai Budiu <mbudiu@feldera.com>
Date: Wed, 24 Jun 2026 13:55:27 -0700
Subject: [PATCH 3/3] Address review comments

Signed-off-by: Mihai Budiu <mbudiu@feldera.com>
---
 .../src/operator/integral_weight_validator.rs | 127 +++++++++---------
 docs.feldera.com/docs/sql/grammar.md          |   7 +-
 2 files changed, 66 insertions(+), 68 deletions(-)

diff --git a/crates/dbsp/src/operator/integral_weight_validator.rs b/crates/dbsp/src/operator/integral_weight_validator.rs
index d97a2b5d619..8000105d1a9 100644
--- a/crates/dbsp/src/operator/integral_weight_validator.rs
+++ b/crates/dbsp/src/operator/integral_weight_validator.rs
@@ -2,87 +2,76 @@
 //! becomes negative.
 
 use crate::{
-    NumEntries, ZWeight,
-    algebra::{AddAssignByRef, AddByRef, HasZero},
+    ZWeight,
     circuit::{Circuit, Stream, checkpointer::Checkpoint},
-    dynamic::Erase,
     trace::{BatchReader as TraceBatchReader, Cursor as TraceCursor},
-    typed_batch::{Batch, IndexedZSetReader},
+    typed_batch::{BatchReader as TypedBatchReader, Spine, ZSet},
 };
-use size_of::SizeOf;
 
 impl<C, B> Stream<C, B>
 where
     C: Circuit,
-    B: Checkpoint
-        + AddByRef
-        + AddAssignByRef
-        + Clone
-        + Eq
-        + HasZero
-        + SizeOf
-        + NumEntries
-        + IndexedZSetReader
-        + Batch<Time = ()>
-        + 'static,
+    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 integral of the input stream.
+    /// Maintains a running trace of the input stream.
     /// After every clock tick, every key that appears in the
-    /// current delta is checked: if any `(key, value, weight)` triple in the
-    /// integral has a non-positive weight, the operator panics, displaying the
-    /// offending key, value, weight and a custom message.
+    /// 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 and testing: it verifies that a
-    /// stream behaves like a valid set or multiset where every element has a
-    /// positive multiplicity.
+    /// 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.
+    /// Panics when any weight in the integral is < 0.
     #[track_caller]
     pub fn integral_weight_validator(&self, message: &str) -> Self {
-        let sharded = self.shard();
-        let integral = sharded.integrate();
         let message = message.to_owned();
-        // Only examine keys that appear in the current delta — those are the
-        // only keys whose integral weight could have decreased this step.
-        sharded.apply2(&integral, move |delta, integrated| {
-            let inner = integrated.inner();
-            let mut cursor = TraceBatchReader::cursor(inner);
-            for (key, _val, _w) in delta.iter() {
-                if cursor.seek_key_exact(key.erase(), None) {
-                    while cursor.val_valid() {
-                        let weight: ZWeight = **cursor.weight();
-                        let val = cursor.val();
-                        assert!(
-                            weight >= 0,
-                            "{message}: negative weight found {weight} for key {key:?}, val {val:?}"
-                        );
-                        cursor.step_val();
+        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()
+            });
+            self.clone()
+        })
     }
 }
 
 #[cfg(test)]
 mod tests {
     use crate::{
-        Circuit, RootCircuit, ZWeight, operator::Generator, typed_batch::OrdZSet, utils::Tup2, zset,
+        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 circuit = RootCircuit::build(move |circuit| {
+        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 };
@@ -92,30 +81,28 @@ mod tests {
             let out = source.integral_weight_validator("test_table");
             let mut expected_step = 0u64;
             out.inspect(move |delta| {
-                // The output is the original delta, not the integral.
                 assert_eq!(delta, &zset! { expected_step => 1 });
                 expected_step += 1;
             });
             Ok(())
         })
-        .unwrap()
-        .0;
+        .unwrap();
 
         for _ in 0..10 {
             circuit.transaction().unwrap();
         }
+        circuit.kill().unwrap();
     }
 
-    /// Inserting an element once and then deleting it twice causes the integral
+    /// Inserting an element once and then deleting it twice causes the trace
     /// weight to go negative, which must trigger a panic.
     #[test]
-    #[should_panic(expected = "table 'test': negative weight found")]
     fn integral_weight_validator_panics_on_negative_integral() {
-        let circuit = RootCircuit::build(move |circuit| {
+        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 → integral weight 1.
-                // Step 1: delete key 42 with weight -2 → integral weight -1 (panic).
+                // 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 {
@@ -127,26 +114,36 @@ mod tests {
             source.integral_weight_validator("table 'test'");
             Ok(())
         })
-        .unwrap()
-        .0;
+        .unwrap();
 
-        circuit.transaction().unwrap(); // step 0: integral weight 1 (ok)
-        circuit.transaction().unwrap(); // step 1: integral weight -1 (panic)
+        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
+    /// A weight that goes negative (e.g., deleting something never inserted)
+    /// must also panic.
     #[test]
-    #[should_panic(expected = "table 'test': negative weight found")]
     fn integral_weight_validator_panics_on_negative_weight() {
-        let circuit = RootCircuit::build(move |circuit| {
-            // Emit weight -1 immediately; integral weight is -1 after step 0.
+        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()
-        .0;
+        .unwrap();
 
-        circuit.transaction().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
index fbe3993572a..9e17b5b75bb 100644
--- a/docs.feldera.com/docs/sql/grammar.md
+++ b/docs.feldera.com/docs/sql/grammar.md
@@ -795,9 +795,10 @@ 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.
+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