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..8000105d1a9 --- /dev/null +++ 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 Stream +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| { + 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 = 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 index f3f10e90ef8..9e17b5b75bb 100644 --- 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/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 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 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());