From 0f6054e413ff0546bd6fff2ab41b04968d466d66 Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Tue, 21 Jul 2026 20:25:33 -0700 Subject: [PATCH] [SQL] GC of tables with keys and LATENESS should never use retain_values, but can use retain_keys when keys have LATENESS Signed-off-by: Mihai Budiu --- crates/dbsp/src/operator/dynamic/input.rs | 14 +- crates/dbsp/src/operator/input.rs | 2 - .../DBSPIntegrateTraceRetainKeysOperator.java | 26 +- ...BSPIntegrateTraceRetainValuesOperator.java | 22 +- .../compiler/backend/ToJsonOuterVisitor.java | 4 +- .../visitors/outer/CircuitRewriter.java | 4 +- .../compiler/visitors/outer/StrayGC.java | 21 + .../outer/monotonicity/InsertLimiters.java | 61 +- .../visitors/outer/monotonicity/MergeGC.java | 2 +- .../compiler/sql/CatalogTests.java | 6 +- .../simple/IncrementalRegression2Tests.java | 4 +- .../sql/streaming/StreamingTests.java | 568 +++++++++++++++++- .../sql/tools/CompilerCircuitStream.java | 11 + 13 files changed, 689 insertions(+), 56 deletions(-) diff --git a/crates/dbsp/src/operator/dynamic/input.rs b/crates/dbsp/src/operator/dynamic/input.rs index e655f60e120..b63ef5db05a 100644 --- a/crates/dbsp/src/operator/dynamic/input.rs +++ b/crates/dbsp/src/operator/dynamic/input.rs @@ -781,8 +781,6 @@ impl RootCircuit { /// /// Retention conditions configured at logical time `t` /// are applied starting from logical time `t+1`. - /// - /// FIXME: see // TODO: Add a version that takes a custom hash function. #[track_caller] pub fn dyn_add_input_map( @@ -1970,10 +1968,8 @@ mod test { Ok(handle) } - // FIXME: the inputs to these tests are meant to exercise the logic that filters inputs based - // on lateness, but it does not currently work correctly (see https://github.com/feldera/feldera/issues/2669). - // We therefore don't use waterlines in tests and check for the standard upsert behavior - // without filtering. + // These tests check the standard upsert behavior without lateness-based filtering; + // filtering inputs based on lateness is implemented by `add_input_map_with_waterline`. #[test] fn map_test_st() { let (mut circuit, mut input_handle) = Runtime::init_circuit(1, move |circuit| { @@ -2033,10 +2029,8 @@ mod test { dbsp.kill().unwrap(); } - // FIXME: the inputs to these tests are meant to exercise the logic that filters inputs based - // on lateness, but it does not currently work correctly (see https://github.com/feldera/feldera/issues/2669). - // We therefore don't use waterlines in tests and check for the standard upsert behavior - // without filtering. + // These tests check the standard upsert behavior without lateness-based filtering; + // filtering inputs based on lateness is implemented by `add_input_map_with_waterline`. #[test] fn map_test_mt1() { map_test_mt(1, input_map_updates1, output_map_updates1); diff --git a/crates/dbsp/src/operator/input.rs b/crates/dbsp/src/operator/input.rs index 0e8ffe42b83..8c2b1ff8f99 100644 --- a/crates/dbsp/src/operator/input.rs +++ b/crates/dbsp/src/operator/input.rs @@ -606,8 +606,6 @@ impl RootCircuit { /// /// Retention conditions configured at logical time `t` /// are applied starting from logical time `t+1`. - /// - /// FIXME: see // TODO: Add a version that takes a custom hash function. #[track_caller] pub fn add_input_map( diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPIntegrateTraceRetainKeysOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPIntegrateTraceRetainKeysOperator.java index 5525dfd6c9c..7506c876e24 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPIntegrateTraceRetainKeysOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPIntegrateTraceRetainKeysOperator.java @@ -27,11 +27,16 @@ public final class DBSPIntegrateTraceRetainKeysOperator extends DBSPBinaryOperator implements IGCOperator { + /** Selects the Rust function that the operator invokes. + * GC applied to an input table requires 'false'; everything else requires 'true'. */ + public final boolean accumulate; + public DBSPIntegrateTraceRetainKeysOperator( CalciteRelNode node, DBSPExpression expression, - OutputPort data, OutputPort control) { - super(node, "accumulate_integrate_trace_retain_keys", expression, - data.outputType(), data.isMultiset(), data, control); + OutputPort data, OutputPort control, boolean accumulate) { + super(node, accumulate ? "accumulate_integrate_trace_retain_keys" : "integrate_trace_retain_keys", + expression, data.outputType(), data.isMultiset(), data, control); + this.accumulate = accumulate; } @Override @@ -43,6 +48,14 @@ public DBSPSimpleOperator asOperator() { @Nullable public static DBSPIntegrateTraceRetainKeysOperator create( CalciteRelNode node, OutputPort data, IMaybeMonotoneType dataProjection, OutputPort control) { + return create(node, data, dataProjection, control, true); + } + + /** Create an operator to retain keys and returns it. May return null if the keys contain no fields. */ + @Nullable + public static DBSPIntegrateTraceRetainKeysOperator create( + CalciteRelNode node, OutputPort data, IMaybeMonotoneType dataProjection, OutputPort control, + boolean accumulate) { DBSPType controlType = control.outputType(); Utilities.enforce(controlType.is(DBSPTypeTupleBase.class), () -> "Control type is not a tuple: " + controlType); DBSPTypeTupleBase controlTuple = controlType.to(DBSPTypeTupleBase.class); @@ -83,7 +96,7 @@ public static DBSPIntegrateTraceRetainKeysOperator create( compare = ExpressionCompiler.makeBinaryExpression( node, compare.getType(), DBSPOpcode.OR, compare0, compare); DBSPExpression closure = compare.closure(param, controlArg.asParameter()); - return new DBSPIntegrateTraceRetainKeysOperator(node, closure, data, control); + return new DBSPIntegrateTraceRetainKeysOperator(node, closure, data, control, accumulate); } @Override @@ -94,7 +107,7 @@ public DBSPSimpleOperator with( Utilities.enforce(newInputs.size() == 2, () -> "Expected 2 inputs, got " + newInputs.size()); return new DBSPIntegrateTraceRetainKeysOperator( this.getRelNode(), toClosure(function), - newInputs.get(0), newInputs.get(1)).copyAnnotations(this); + newInputs.get(0), newInputs.get(1), this.accumulate).copyAnnotations(this); } return this; } @@ -113,8 +126,9 @@ public void accept(CircuitVisitor visitor) { @SuppressWarnings("unused") public static DBSPIntegrateTraceRetainKeysOperator fromJson(JsonNode node, JsonDecoder decoder) { DBSPSimpleOperator.CommonInfo info = commonInfoFromJson(node, decoder); + boolean accumulate = Utilities.getBooleanProperty(node, "accumulate"); return new DBSPIntegrateTraceRetainKeysOperator(CalciteEmptyRel.INSTANCE, - info.getFunction(), info.getInput(0), info.getInput(1)) + info.getFunction(), info.getInput(0), info.getInput(1), accumulate) .addAnnotations(info.annotations(), DBSPIntegrateTraceRetainKeysOperator.class); } } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPIntegrateTraceRetainValuesOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPIntegrateTraceRetainValuesOperator.java index 9a8b082a5e7..59216d3a498 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPIntegrateTraceRetainValuesOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPIntegrateTraceRetainValuesOperator.java @@ -27,14 +27,11 @@ public final class DBSPIntegrateTraceRetainValuesOperator extends DBSPBinaryOperator implements IGCOperator { - public final boolean accumulate; - public DBSPIntegrateTraceRetainValuesOperator( CalciteRelNode node, DBSPExpression function, - OutputPort data, OutputPort control, boolean accumulate) { - super(node, accumulate ? "accumulate_integrate_trace_retain_values" : "integrate_trace_retain_values", + OutputPort data, OutputPort control) { + super(node, "accumulate_integrate_trace_retain_values", function, data.outputType(), data.isMultiset(), data, control); - this.accumulate = accumulate; } @Override @@ -43,7 +40,7 @@ public DBSPSimpleOperator asOperator() { } public static DBSPIntegrateTraceRetainValuesOperator create( - CalciteRelNode node, OutputPort data, IMaybeMonotoneType dataProjection, OutputPort control, boolean accumulate) { + CalciteRelNode node, OutputPort data, IMaybeMonotoneType dataProjection, OutputPort control) { DBSPType controlType = control.outputType(); Utilities.enforce(controlType.is(DBSPTypeTupleBase.class), () -> "Control type is not a tuple: " + controlType); @@ -66,14 +63,9 @@ public static DBSPIntegrateTraceRetainValuesOperator create( compare = ExpressionCompiler.makeBinaryExpression( node, compare.getType(), DBSPOpcode.OR, compare0, compare); DBSPExpression closure = compare.closure(param, controlArg.asParameter()); - return new DBSPIntegrateTraceRetainValuesOperator(node, closure, data, control, accumulate); + return new DBSPIntegrateTraceRetainValuesOperator(node, closure, data, control); } - public static DBSPIntegrateTraceRetainValuesOperator create( - CalciteRelNode node, OutputPort data, IMaybeMonotoneType dataProjection, OutputPort control) { - return create(node, data, dataProjection, control, true); -} - @Override public DBSPSimpleOperator with( @Nullable DBSPExpression function, DBSPType outputType, @@ -82,7 +74,7 @@ public DBSPSimpleOperator with( Utilities.enforce(newInputs.size() == 2, () -> "Expected 2 inputs, got " + newInputs.size()); return new DBSPIntegrateTraceRetainValuesOperator( this.getRelNode(), Objects.requireNonNull(function), - newInputs.get(0), newInputs.get(1), this.accumulate); + newInputs.get(0), newInputs.get(1)); } return this; } @@ -101,10 +93,8 @@ public void accept(CircuitVisitor visitor) { @SuppressWarnings("unused") public static DBSPIntegrateTraceRetainValuesOperator fromJson(JsonNode node, JsonDecoder decoder) { DBSPSimpleOperator.CommonInfo info = commonInfoFromJson(node, decoder); - boolean accumulate = Utilities.getBooleanProperty(node, "accumulate"); - return new DBSPIntegrateTraceRetainValuesOperator(CalciteEmptyRel.INSTANCE, - info.getFunction(), info.getInput(0), info.getInput(1), accumulate) + info.getFunction(), info.getInput(0), info.getInput(1)) .addAnnotations(info.annotations(), DBSPIntegrateTraceRetainValuesOperator.class); } } 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 c4a8ef31348..364de81bfeb 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 @@ -23,7 +23,7 @@ import org.dbsp.sqlCompiler.circuit.operator.DBSPUnaryOperator; 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.DBSPIntegrateTraceRetainKeysOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPWeightValidatorOperator; import org.dbsp.sqlCompiler.compiler.DBSPCompiler; import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.ProgramIdentifier; @@ -337,7 +337,7 @@ public VisitDecision preorder(DBSPWindowOperator operator) { } @Override - public VisitDecision preorder(DBSPIntegrateTraceRetainValuesOperator operator) { + public VisitDecision preorder(DBSPIntegrateTraceRetainKeysOperator operator) { if (this.preorder(operator.to(DBSPBinaryOperator.class)).stop()) return VisitDecision.STOP; this.property("accumulate"); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitRewriter.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitRewriter.java index e29c62464f3..c73986c6e6c 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitRewriter.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitRewriter.java @@ -722,7 +722,7 @@ public void postorder(DBSPIntegrateTraceRetainKeysOperator operator) { || Linq.different(sources, operator.inputs) || function != operator.getFunction()) { result = new DBSPIntegrateTraceRetainKeysOperator(operator.getRelNode(), function, - sources.get(0), sources.get(1)) + sources.get(0), sources.get(1), operator.accumulate) .copyAnnotations(operator); } this.map(operator, result); @@ -738,7 +738,7 @@ public void postorder(DBSPIntegrateTraceRetainValuesOperator operator) { || Linq.different(sources, operator.inputs) || function != operator.getFunction()) { result = new DBSPIntegrateTraceRetainValuesOperator(operator.getRelNode(), function, - sources.get(0), sources.get(1), operator.accumulate) + sources.get(0), sources.get(1)) .copyAnnotations(operator); } this.map(operator, result); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/StrayGC.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/StrayGC.java index d170edb8fe8..f42093ae855 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/StrayGC.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/StrayGC.java @@ -22,6 +22,22 @@ public StrayGC(DBSPCompiler compiler, CircuitGraphs g) { super(compiler, g); } + /** Check that the retain operator invokes the runtime function variant that + * matches its data source: input tables require the non-accumulate variant, + * everything else the accumulate_ one. The wrong variant makes the + * operator a silent no-op. */ + void checkAccumulate(DBSPBinaryOperator operator, boolean accumulate) { + boolean input = operator.left().operator.is(DBSPInputMapWithWaterlineOperator.class); + if (input && accumulate) + throw new InternalCompilerError( + "Operator " + operator + " garbage-collects an input table " + + "and must use the non-accumulate variant"); + if (!input && !accumulate) + throw new InternalCompilerError( + "Operator " + operator + " garbage-collects a trace inside the " + + "accumulate framework and must use the accumulate_ variant"); + } + void check(DBSPBinaryOperator operator) { // At least one sibling on the left input must contain an integral var left = operator.left(); @@ -45,16 +61,21 @@ void check(DBSPBinaryOperator operator) { @Override public void postorder(DBSPIntegrateTraceRetainValuesOperator operator) { + // This operator always uses the accumulate_ variant + this.checkAccumulate(operator, true); this.check(operator); } @Override public void postorder(DBSPIntegrateTraceRetainNValuesOperator operator) { + // This operator always uses the accumulate_ variant + this.checkAccumulate(operator, true); this.check(operator); } @Override public void postorder(DBSPIntegrateTraceRetainKeysOperator operator) { + this.checkAccumulate(operator, operator.accumulate); DBSPOperator left = operator.left().operator; if (left.is(DBSPAggregateLinearPostprocessRetainKeysOperator.class) || left.is(DBSPChainAggregateOperator.class) || diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/InsertLimiters.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/InsertLimiters.java index 47e46280947..5df2101530c 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/InsertLimiters.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/InsertLimiters.java @@ -1741,19 +1741,58 @@ DBSPOperator processLateness( if (operator != expansion) this.markBound(expansion.outputPort(), extend.outputPort()); - if (INSERT_RETAIN_VALUES && replaceIndexedInput && + if (INSERT_RETAIN_KEYS && replaceIndexedInput && multisetInput != null && + // Do not GC materialized tables. !multisetInput.getMetadata().materialized) { - // Do not GC materialized tables - IMaybeMonotoneType projection = me.getMonotoneType().to(MonotoneClosureType.class).getBodyType(); - // The new input operator produces an indexed Zset, need to adjust the projection - projection = new PartiallyMonotoneTuple( - List.of(NonMonotoneType.nonMonotone(indexedOutputType.keyType), projection), true, false); - DBSPSimpleOperator retain = DBSPIntegrateTraceRetainValuesOperator.create( - operator.getRelNode(), newSource.getOutput(0), - // For inputs "accumulate" is 'false'. - projection, extend.outputPort(), false); - this.addOperator(retain); + // The trace of an input with a primary key resolves upserts and deletions, + // so it can be garbage-collected only by key, and only using LATENESS + // columns that belong to the key: a key pruned this way can never be + // legally referenced again, because any subsequent update or deletion + // carries the same key and is rejected as late. + // A LATENESS column outside the key permits no garbage collection at all: + // - a live row (timestamp at or above the waterline) must be retained + // to produce the retraction when its key is updated or deleted; + // - a frozen row (timestamp below the waterline) can never be retracted, + // but its presence is what lets the input operator reject writes to + // its key as late; if the row is dropped, the next insert on that key + // is accepted as a brand-new key without a retraction, so the output + // would depend on when compaction runs. + List keyFieldsMonotone = new ArrayList<>(); + List keyWaterlineFields = new ArrayList<>(); + int latenessIndex = 0; + for (IColumnMetadata column: multisetInput.to(IHasColumnsMetadata.class).getColumnsMetadata()) { + boolean hasLateness = column.getLateness() != null; + if (column.isPrimaryKey()) { + if (hasLateness) { + keyFieldsMonotone.add(new MonotoneType(column.getType())); + keyWaterlineFields.add(latenessIndex); + } else { + keyFieldsMonotone.add(NonMonotoneType.nonMonotone(column.getType())); + } + } + if (hasLateness) + latenessIndex++; + } + if (!keyWaterlineFields.isEmpty()) { + PartiallyMonotoneTuple keyProjection = new PartiallyMonotoneTuple(keyFieldsMonotone, false, false); + IMaybeMonotoneType projection = new PartiallyMonotoneTuple( + List.of(keyProjection, NonMonotoneType.nonMonotone(indexedOutputType.elementType)), + true, false); + // Project the waterline to the components that correspond to key columns + DBSPVariablePath w = timestamp.getType().ref().var(); + DBSPExpression keyWaterline = new DBSPRawTupleExpression( + new DBSPTupleExpression( + Linq.map(keyWaterlineFields, f -> w.deepCopy().deref().field(f)), + false)); + OutputPort control = this.createApply(extend.outputPort(), null, keyWaterline.closure(w)); + DBSPSimpleOperator retain = DBSPIntegrateTraceRetainKeysOperator.create( + operator.getRelNode(), newSource.getOutput(0), + // For inputs "accumulate" is 'false'. + projection, control, false); + if (retain != null) + this.addOperator(retain); + } } return result; diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/MergeGC.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/MergeGC.java index 26fdba98b40..e65ab4849bc 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/MergeGC.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/MergeGC.java @@ -186,7 +186,7 @@ DBSPIntegrateTraceRetainKeysOperator merge(List no output ccs.step(""" INSERT INTO auction VALUES('2024-01-01 00:00:00', '2024-01-01 01:00:00', 0); @@ -840,6 +841,571 @@ auction INT FOREIGN KEY REFERENCES auction(id) ----------------------------------------------------------"""); } + /** Asserts that the circuit contains exactly {@code keys} + * integrate_trace_retain_keys operators, and no integrate_trace_retain_values + * operators, that garbage-collect an input table. */ + void checkInputGC(CompilerCircuit cc, int keys) { + cc.visit(new CircuitVisitor(cc.compiler) { + int retainKeys = 0; + int retainValues = 0; + + @Override + public void postorder(DBSPIntegrateTraceRetainKeysOperator operator) { + if (operator.left().operator.is(DBSPInputMapWithWaterlineOperator.class)) + this.retainKeys++; + } + + @Override + public void postorder(DBSPIntegrateTraceRetainValuesOperator operator) { + if (operator.left().operator.is(DBSPInputMapWithWaterlineOperator.class)) + this.retainValues++; + } + + @Override + public void endVisit() { + Assert.assertEquals(keys, this.retainKeys); + Assert.assertEquals(0, this.retainValues); + } + }); + } + + @Test + public void gcUpsertBoundary() { + // The LATENESS column is not part of the primary key, so the input + // trace is never garbage-collected: any key can still be updated. + // Rows whose timestamp is exactly on the waterline are not late and + // can be updated and deleted. + // The waterline is max over all data of (ts - 100); each step is + // filtered with the waterline computed from the previous steps. + String sql = """ + CREATE TABLE t ( + id INT NOT NULL PRIMARY KEY, + ts BIGINT NOT NULL LATENESS 100 + ); + CREATE VIEW v AS SELECT * FROM t;"""; + CompilerCircuitStream ccs = this.getCCS(sql).compactAfterEachStep(); + this.checkInputGC(ccs, 0); + // Before this step: waterline = minimum, table is empty + ccs.step(""" + INSERT INTO t VALUES(1, 0), (2, 100), (3, 200); + """, """ + id | ts | weight + -------------------- + 1 | 0 | 1 + 2 | 100 | 1 + 3 | 200 | 1"""); + // Before this step, table contents (rows above the ==== line are + // below the waterline): + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 200 - 100 + // 2 | 100 | on the waterline: still updatable + // 3 | 200 | + // Updating key 2 retracts the old row. + ccs.step(""" + INSERT INTO t VALUES(2, 300); + """, """ + id | ts | weight + -------------------- + 2 | 100 | -1 + 2 | 300 | 1"""); + // Before this step, table contents: + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 300 - 100 + // 3 | 200 | on the waterline: still deletable + // 2 | 300 | + ccs.step(""" + REMOVE FROM t VALUES(3, 200); + """, """ + id | ts | weight + -------------------- + 3 | 200 | -1"""); + // Before this step, table contents: + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 300 - 100 + // 2 | 300 | + ccs.step(""" + INSERT INTO t VALUES(4, 400); + """, """ + id | ts | weight + -------------------- + 4 | 400 | 1"""); + // Before this step, table contents: + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 400 - 100 + // 2 | 300 | on the waterline + // 4 | 400 | + // A new key with a timestamp below the waterline is late and ignored. + ccs.step(""" + INSERT INTO t VALUES(5, 100); + """, """ + id | ts | weight + --------------------"""); + } + + @Test + public void gcDeleteOldRow() { + // Deleting a key whose row is below the waterline is ignored: the + // deletion itself is late. Deleting a row at or above the waterline works. + String sql = """ + CREATE TABLE t ( + id INT NOT NULL PRIMARY KEY, + ts BIGINT NOT NULL LATENESS 100 + ); + CREATE VIEW v AS SELECT * FROM t;"""; + CompilerCircuitStream ccs = this.getCCS(sql).compactAfterEachStep(); + this.checkInputGC(ccs, 0); + // Before this step: waterline = minimum, table is empty + ccs.step(""" + INSERT INTO t VALUES(1, 0), (2, 200); + """, """ + id | ts | weight + -------------------- + 1 | 0 | 1 + 2 | 200 | 1"""); + // Before this step, table contents: + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 200 - 100 + // 2 | 200 | + ccs.step(""" + INSERT INTO t VALUES(3, 250); + """, """ + id | ts | weight + -------------------- + 3 | 250 | 1"""); + // Before this step, table contents: + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 250 - 100 + // 2 | 200 | + // 3 | 250 | + // Deleting frozen row (1, 0) is ignored. + ccs.step(""" + REMOVE FROM t VALUES(1, 0); + """, """ + id | ts | weight + --------------------"""); + // Before this step: waterline = 250 - 100, table contents unchanged. + // Row (2, 200) is above the waterline: deleting it works. + ccs.step(""" + REMOVE FROM t VALUES(2, 200); + """, """ + id | ts | weight + -------------------- + 2 | 200 | -1"""); + } + + @Test + public void gcUpsertOldRow() { + // Updating a key whose row is below the waterline must be ignored: + // the update would have to retract the old row, which is behind the + // lateness threshold. See the warning about primary keys and LATENESS + // in docs.feldera.com/docs/tutorials/time-series.md: "old" records in + // such a table can never be updated or deleted. + String sql = """ + CREATE TABLE t ( + id INT NOT NULL PRIMARY KEY, + ts BIGINT NOT NULL LATENESS 100 + ); + CREATE VIEW v AS SELECT * FROM t;"""; + CompilerCircuitStream ccs = this.getCCS(sql).compactAfterEachStep(); + this.checkInputGC(ccs, 0); + // Before this step: waterline = minimum, table is empty + ccs.step(""" + INSERT INTO t VALUES(1, 0), (2, 200); + """, """ + id | ts | weight + -------------------- + 1 | 0 | 1 + 2 | 200 | 1"""); + // Before this step, table contents: + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 200 - 100 + // 2 | 200 | + ccs.step(""" + INSERT INTO t VALUES(3, 250); + """, """ + id | ts | weight + -------------------- + 3 | 250 | 1"""); + // Before this step, table contents: + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 250 - 100 + // 2 | 200 | + // 3 | 250 | + // The update of key 1 is ignored, even though the new timestamp 300 + // is above the waterline: it would have to retract frozen row (1, 0). + ccs.step(""" + INSERT INTO t VALUES(1, 300); + """, """ + id | ts | weight + --------------------"""); + } + + @Test + public void gcUpsertOldRowLongLag() { + // Like gcUpsertOldRow, but with several steps between the moment the + // row falls below the waterline and the attempt to update it, giving + // compaction many opportunities to run. The outcome must not depend + // on compaction timing. + String sql = """ + CREATE TABLE t ( + id INT NOT NULL PRIMARY KEY, + ts BIGINT NOT NULL LATENESS 100 + ); + CREATE VIEW v AS SELECT * FROM t;"""; + CompilerCircuitStream ccs = this.getCCS(sql).compactAfterEachStep(); + this.checkInputGC(ccs, 0); + // Before this step: waterline = minimum, table is empty + ccs.step(""" + INSERT INTO t VALUES(1, 0), (2, 200); + """, """ + id | ts | weight + -------------------- + 1 | 0 | 1 + 2 | 200 | 1"""); + // Before this step, table contents: + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 200 - 100 + // 2 | 200 | + ccs.step(""" + INSERT INTO t VALUES(3, 250); + """, """ + id | ts | weight + -------------------- + 3 | 250 | 1"""); + // Before this step, table contents: + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 250 - 100 + // 2 | 200 | + // 3 | 250 | + ccs.step(""" + INSERT INTO t VALUES(4, 260); + """, """ + id | ts | weight + -------------------- + 4 | 260 | 1"""); + // Before this step, table contents: + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 260 - 100 + // 2 | 200 | + // 3 | 250 | + // 4 | 260 | + ccs.step(""" + INSERT INTO t VALUES(5, 270); + """, """ + id | ts | weight + -------------------- + 5 | 270 | 1"""); + // Before this step, table contents: + // id | ts | + // ----+-----+ + // 1 | 0 | frozen + // ================ waterline = 270 - 100 + // 2 | 200 | + // 3 | 250 | + // 4 | 260 | + // 5 | 270 | + // The update of key 1 is ignored, even though the new timestamp 300 + // is above the waterline: it would have to retract frozen row (1, 0). + ccs.step(""" + INSERT INTO t VALUES(1, 300); + """, """ + id | ts | weight + --------------------"""); + } + + @Test + public void gcTwoLatenessColumns() { + // Neither LATENESS column is part of the primary key, so the input + // trace is never garbage-collected. The waterline is a pair compared + // pointwise; a row is frozen when it is below the waterline in ANY + // component. A row fresh in both components stays updatable. + String sql = """ + CREATE TABLE t ( + id INT NOT NULL PRIMARY KEY, + ts1 BIGINT NOT NULL LATENESS 100, + ts2 BIGINT NOT NULL LATENESS 100 + ); + CREATE VIEW v AS SELECT * FROM t;"""; + CompilerCircuitStream ccs = this.getCCS(sql).compactAfterEachStep(); + this.checkInputGC(ccs, 0); + // Before this step: waterline = (minimum, minimum), table is empty + ccs.step(""" + INSERT INTO t VALUES(1, 0, 1000), (2, 1000, 0), (3, 1000, 1000); + """, """ + id | ts1 | ts2 | weight + ---------------------------- + 1 | 0 | 1000 | 1 + 2 | 1000 | 0 | 1 + 3 | 1000 | 1000 | 1"""); + // Before this step: waterline = (1000 - 100, 1000 - 100). + // The waterline pair is compared pointwise; table contents: + // id | ts1 | ts2 | + // ----+------+------+----------------------------- + // 1 | 0 | 1000 | ts1 below waterline: frozen + // 2 | 1000 | 0 | ts2 below waterline: frozen + // 3 | 1000 | 1000 | + // Row 3 is at or above the waterline in both components, + // so updating key 3 retracts the old row. + ccs.step(""" + INSERT INTO t VALUES(3, 1100, 1100); + """, """ + id | ts1 | ts2 | weight + ---------------------------- + 3 | 1000 | 1000 | -1 + 3 | 1100 | 1100 | 1"""); + // Before this step: waterline = (1100 - 100, 1100 - 100), table contents: + // id | ts1 | ts2 | + // ----+------+------+----------------------------- + // 1 | 0 | 1000 | ts1 below waterline: frozen + // 2 | 1000 | 0 | ts2 below waterline: frozen + // 3 | 1100 | 1100 | + // Row 1 is frozen through ts1 alone: the update is ignored, even + // though the old ts2 and both new timestamps are fresh. + ccs.step(""" + INSERT INTO t VALUES(1, 1150, 1150); + """, """ + id | ts1 | ts2 | weight + ----------------------------"""); + // Before this step: waterline = (1100 - 100, 1100 - 100), table contents unchanged. + // Row 2 is frozen through ts2 alone: the deletion is ignored. + ccs.step(""" + REMOVE FROM t VALUES(2, 1000, 0); + """, """ + id | ts1 | ts2 | weight + ----------------------------"""); + // Before this step: waterline = (1100 - 100, 1100 - 100), table contents unchanged + ccs.step(""" + INSERT INTO t VALUES(4, 1200, 1200); + """, """ + id | ts1 | ts2 | weight + ---------------------------- + 4 | 1200 | 1200 | 1"""); + } + + @Test + public void gcKeyLateness() { + // The LATENESS column is the primary key, so the compiler + // garbage-collects the input trace by key. + String sql = """ + CREATE TABLE t ( + ts BIGINT NOT NULL PRIMARY KEY LATENESS 100, + v INT + ); + CREATE VIEW vw AS SELECT * FROM t;"""; + CompilerCircuitStream ccs = this.getCCS(sql).compactAfterEachStep(); + this.checkInputGC(ccs, 1); + // Before this step: waterline = minimum, table is empty + ccs.step(""" + INSERT INTO t VALUES(0, 1), (100, 1), (200, 1); + """, """ + ts | v | weight + ------------------- + 0 | 1 | 1 + 100 | 1 | 1 + 200 | 1 | 1"""); + // Before this step, table contents (rows above the ==== line are + // below the waterline): + // ts | v | + // -----+---+ + // 0 | 1 | GC'd + // ================ waterline = 200 - 100 + // 100 | 1 | on the waterline: still updatable + // 200 | 1 | + // Updating key 100 retracts the old row. + ccs.step(""" + INSERT INTO t VALUES(100, 2); + """, """ + ts | v | weight + ------------------- + 100 | 1 | -1 + 100 | 2 | 1"""); + // Before this step, table contents: + // ts | v | + // -----+---+ + // 0 | 1 | GC'd + // ================ waterline = 200 - 100 + // 100 | 2 | + // 200 | 1 | + // Updating GC'd key 0 is rejected: the record itself is late. + ccs.step(""" + INSERT INTO t VALUES(0, 5); + """, """ + ts | v | weight + -------------------"""); + // Before this step, table contents: + // ts | v | + // -----+---+ + // 0 | 1 | GC'd + // ================ waterline = 200 - 100 + // 100 | 2 | + // 200 | 1 | + // Deleting GC'd key 0 is a no-op. + ccs.step(""" + REMOVE FROM t VALUES(0, 1); + """, """ + ts | v | weight + -------------------"""); + // Before this step, table contents: + // ts | v | + // -----+---+ + // 0 | 1 | GC'd + // ================ waterline = 200 - 100 + // 100 | 2 | + // 200 | 1 | + ccs.step(""" + INSERT INTO t VALUES(300, 1); + """, """ + ts | v | weight + ------------------- + 300 | 1 | 1"""); + // Before this step, table contents: + // ts | v | + // -----+---+ + // 0 | 1 | GC'd + // 100 | 2 | GC'd + // ================ waterline = 300 - 100 + // 200 | 1 | on the waterline: still deletable + // 300 | 1 | + ccs.step(""" + REMOVE FROM t VALUES(200, 1); + """, """ + ts | v | weight + ------------------- + 200 | 1 | -1"""); + // Before this step, table contents: + // ts | v | + // -----+---+ + // 0 | 1 | GC'd + // 100 | 2 | GC'd + // ================ waterline = 300 - 100 + // 300 | 1 | + ccs.step(""" + INSERT INTO t VALUES(300, 9); + """, """ + ts | v | weight + ------------------- + 300 | 1 | -1 + 300 | 9 | 1"""); + } + + @Test + public void gcCompositeKeyLateness() { + // Composite primary key (id, ts) where only ts has LATENESS, plus a + // LATENESS column outside the key. The input trace is + // garbage-collected by key, using only the ts component of the + // waterline. + String sql = """ + CREATE TABLE t ( + id INT NOT NULL, + ts BIGINT NOT NULL LATENESS 100, + extra BIGINT NOT NULL LATENESS 100, + v INT, + PRIMARY KEY (id, ts) + ); + CREATE VIEW vw AS SELECT * FROM t;"""; + CompilerCircuitStream ccs = this.getCCS(sql).compactAfterEachStep(); + this.checkInputGC(ccs, 1); + // Before this step: waterline = (minimum, minimum), table is empty + ccs.step(""" + INSERT INTO t VALUES(1, 0, 0, 1), (2, 200, 200, 1); + """, """ + id | ts | extra | v | weight + -------------------------------- + 1 | 0 | 0 | 1 | 1 + 2 | 200 | 200 | 1 | 1"""); + // Before this step, table ==== line partitions rows by + // the ts component of the key, the only one used for GC): + // id | ts | extra | + // ----+-----+-------+ + // 1 | 0 | 0 | GC'd + // ======================== waterline = (200 - 100, 200 - 100) + // 2 | 200 | 200 | + // Updating live key (2, 200) retracts the old row. + ccs.step(""" + INSERT INTO t VALUES(2, 200, 200, 9); + """, """ + id | ts | extra | v | weight + -------------------------------- + 2 | 200 | 200 | 1 | -1 + 2 | 200 | 200 | 9 | 1"""); + // Before this step: + // id | ts | extra | + // ----+-----+-------+ + // 1 | 0 | 0 | GC'd + // ======================== waterline = (200 - 100, 200 - 100) + // 2 | 200 | 200 | + // Updating GC'd key (1, 0) is rejected: its ts component is late, + // even though the extra column is fresh. + ccs.step(""" + INSERT INTO t VALUES(1, 0, 300, 5); + """, """ + id | ts | extra | v | weight + --------------------------------"""); + // Before this step, table contents: + // id | ts | extra | + // ----+-----+-------+ + // 1 | 0 | 0 | GC'd + // ======================== waterline = (200 - 100, 200 - 100) + // 2 | 200 | 200 | + // Key (1, 300) is new; it does not collide with GC'd key (1, 0). + ccs.step(""" + INSERT INTO t VALUES(1, 300, 300, 1); + """, """ + id | ts | extra | v | weight + -------------------------------- + 1 | 300 | 300 | 1 | 1"""); + // Before this step, table contents: + // id | ts | extra | + // ----+-----+-------+ + // 1 | 0 | 0 | GC'd + // ======================== waterline = (300 - 100, 300 - 100) + // 2 | 200 | 200 | on the waterline: deletable + // 1 | 300 | 300 | + ccs.step(""" + REMOVE FROM t VALUES(2, 200, 200, 9); + """, """ + id | ts | extra | v | weight + -------------------------------- + 2 | 200 | 200 | 9 | -1"""); + } + + @Test + public void gcMaterializedNoRetain() { + // Materialized tables are never garbage-collected, so the compiler + // must not insert any retain operator for them, even when a primary + // key column has LATENESS. + String sql = """ + CREATE TABLE t ( + ts BIGINT NOT NULL PRIMARY KEY LATENESS 100, + v INT + ) WITH ('materialized' = 'true'); + CREATE VIEW vw AS SELECT * FROM t;"""; + CompilerCircuit cc = this.getCC(sql); + this.checkInputGC(cc, 0); + } + @Test public void testEmitFail() { this.statementsFailingInCompilation(""" diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/CompilerCircuitStream.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/CompilerCircuitStream.java index 5d751538307..6adf7d9400f 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/CompilerCircuitStream.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/CompilerCircuitStream.java @@ -21,6 +21,7 @@ * Submits the code for compilation by the Rust compiler. */ public class CompilerCircuitStream extends CompilerCircuit { final InputOutputChangeStream stream; + boolean compactAfterEachStep; public CompilerCircuitStream(DBSPCompiler compiler, BaseSQLTests test) { this(compiler, new InputOutputChangeStream(), test); @@ -39,6 +40,12 @@ public CompilerCircuitStream(DBSPCompiler compiler, InputOutputChangeStream stre super(compiler); this.stream = streams; test.addRustTestCase(this); + this.compactAfterEachStep = false; + } + + public CompilerCircuitStream compactAfterEachStep() { + this.compactAfterEachStep = true; + return this; } public CompilerCircuitStream( @@ -74,6 +81,8 @@ public void step(String script, String expected) { DBSPType outputType = this.circuit.getSingleOutputType(); Change output = TableParser.parseChangeTable(expected, outputType); this.stream.addPair(input, output); + if (this.compactAfterEachStep) + this.blockForCompaction(); } /** @@ -103,6 +112,8 @@ public void stepWeightOne(String script, String expected) { public void step(Change input, Change output) { this.stream.addPair(input, output); + if (this.compactAfterEachStep) + this.blockForCompaction(); } /** Execute some insert/delete statements using sqlite and return the produced result.