From 0a8c752d72e722a16ce8eb8aa0f9d7b0d5d2b54d Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Fri, 31 Jul 2026 17:11:17 -0700 Subject: [PATCH] [SQL] Implement SESSION windows Signed-off-by: Mihai Budiu --- docs.feldera.com/docs/sql/function-index.md | 1 + docs.feldera.com/docs/sql/table.md | 67 ++++++ .../docs/sql/unsupported-operations.md | 5 - .../calciteCompiler/CalciteFunctions.java | 2 + .../optimizer/CalciteOptimizer.java | 4 + .../optimizer/SessionRewriteRule.java | 197 +++++++++++++++ .../compiler/visitors/outer/DumpTopology.java | 74 ++++++ .../outer/RemoveIdentityOperators.java | 180 ++++++++++++-- .../compiler/visitors/outer/StrayGC.java | 26 ++ .../compiler/sql/quidem/SessionTests.java | 227 ++++++++++++++++++ .../sql/streaming/StreamingTests.java | 218 +++++++++++++++++ 11 files changed, 973 insertions(+), 28 deletions(-) create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/SessionRewriteRule.java create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DumpTopology.java create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/SessionTests.java diff --git a/docs.feldera.com/docs/sql/function-index.md b/docs.feldera.com/docs/sql/function-index.md index f8955169ef1..eaad31628bb 100644 --- a/docs.feldera.com/docs/sql/function-index.md +++ b/docs.feldera.com/docs/sql/function-index.md @@ -229,6 +229,7 @@ * `SECH`: [float](float.md#sech) * `SECOND`: [datetime](datetime.md#date_second), [datetime](datetime.md#time_second), [datetime](datetime.md#timestamp_second) * `SEQUENCE`: [integer](integer.md#sequence) +* `SESSION`: [table](table.md#session) * `SIGN`: [decimal](decimal.md#sign) * `SIN`: [float](float.md#sin) * `SINH`: [float](float.md#sinh) diff --git a/docs.feldera.com/docs/sql/table.md b/docs.feldera.com/docs/sql/table.md index bd884c7d158..b1dbee83f27 100644 --- a/docs.feldera.com/docs/sql/table.md +++ b/docs.feldera.com/docs/sql/table.md @@ -118,3 +118,70 @@ function: - `window_end`, of the same type as the column `orders.rowtime` A `NULL` timestamp produces no rows in the result. + +### `SESSION` + +`SESSION` groups rows into sessions based on a timestamp column. Two +rows belong to the same session when their timestamps are less than +`size` (the inactivity gap) apart. Unlike `TUMBLE` and `HOP` windows, +session windows are not fixed in absolute time: each session starts at +the timestamp of its first row and ends `size` after the timestamp of +its last row. The optional `key` descriptor partitions the rows; +sessions are formed separately within each key. + +Here is an example showing session windows defined by intervals longer +than 10 minutes (we only show the timestamps of the rows involved, +sorted increasingly). + +``` +10:00 -- session starts +10:04 | +10:13 | +10:20 -- session ends +10:32 -- session starts +10:36 | +10:40 -- session ends +10:51 -- session starts and ends +``` + +#### Syntax: + +``` +SESSION(data, DESCRIPTOR(timecol) [, DESCRIPTOR(key) ], size) +``` + +The type of the `timecol` has to be `TIMESTAMP`. + +Here is an example: + +```sql +SELECT * FROM TABLE( + SESSION( + TABLE orders, + DESCRIPTOR(rowtime), + DESCRIPTOR(product), + INTERVAL '20' MINUTE)); + +-- or with the named params +-- note: the DATA param must be the first +SELECT * FROM TABLE( + SESSION( + DATA => TABLE orders, + TIMECOL => DESCRIPTOR(rowtime), + KEY => DESCRIPTOR(product), + SIZE => INTERVAL '20' MINUTE)); +``` + +groups the rows of `orders` into sessions per `product`; a session +ends when a product receives no orders for 20 minutes. + +The result is a table that has all the columns of the `orders` table, +and in addition the following columns, defined by the `SESSION` +function: +- `window_start`, of the same type as the column `orders.rowtime`; + the timestamp of the session's first row +- `window_end`, of the same type as the column `orders.rowtime`; + the timestamp of the session's last row plus `size` + +A `NULL` timestamp produces no rows in the result. A `NULL` key +groups rows like any other key value. diff --git a/docs.feldera.com/docs/sql/unsupported-operations.md b/docs.feldera.com/docs/sql/unsupported-operations.md index f13d2d276a9..e20c3a0fbfe 100644 --- a/docs.feldera.com/docs/sql/unsupported-operations.md +++ b/docs.feldera.com/docs/sql/unsupported-operations.md @@ -120,11 +120,6 @@ example usage. Dynamic `PIVOT` is not yet supported. ## `MULTISET` Data Type The `MULTISET` data type is not currently supported. -## Session windows - -Session windows (grouping events into sessions based on a gap in -activity) are not yet supported. - ## `TIME` with timezone The type `TIME WITH TIME ZONE` is not supported. diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/CalciteFunctions.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/CalciteFunctions.java index c230020e72b..26265464d63 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/CalciteFunctions.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/CalciteFunctions.java @@ -498,6 +498,8 @@ record Func(SqlOperator function, String functionName, SqlLibrary library, "runtime_aggtest/illarg_tests/test_grammar_tbl_fn.py", false), new Func(SqlStdOperatorTable.HOP, "HOP", SqlLibrary.STANDARD, "table#hop", "runtime_aggtest/illarg_tests/test_grammar_tbl_fn.py", false), + new Func(SqlStdOperatorTable.SESSION, "SESSION", SqlLibrary.STANDARD, "table#session", + FunctionDocumentation.NO_FILE, false), // SqlLibraryOperators operators // DATEADD is not implemented, but give a better error message diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java index 74a896b7458..6bee6e2140a 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/CalciteOptimizer.java @@ -227,6 +227,10 @@ void createOptimizer() { CoreRules.AGGREGATE_MERGE, CoreRules.INTERSECT_MERGE); + // Must run before "Expand windows", which converts the RexOver + // projections this rewrite creates. + this.addStep(new SimpleOptimizerStep("Rewrite SESSION", 0, + new SessionRewriteRule())); this.addStep(new SimpleOptimizerStep("Constant fold", 2, CoreRules.COERCE_INPUTS, SingleValuesOptimizationRules.JOIN_LEFT_INSTANCE, diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/SessionRewriteRule.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/SessionRewriteRule.java new file mode 100644 index 00000000000..2e2d74fe432 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/optimizer/SessionRewriteRule.java @@ -0,0 +1,197 @@ +package org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.optimizer; + +import org.apache.calcite.plan.RelOptRuleCall; +import org.apache.calcite.plan.RelRule; +import org.apache.calcite.rel.RelNode; +import org.apache.calcite.rel.core.JoinRelType; +import org.apache.calcite.rel.logical.LogicalTableFunctionScan; +import org.apache.calcite.rel.rules.TransformationRule; +import org.apache.calcite.rex.RexCall; +import org.apache.calcite.rex.RexInputRef; +import org.apache.calcite.rex.RexNode; +import org.apache.calcite.rex.RexUtil; +import org.apache.calcite.rex.RexWindowBounds; +import org.apache.calcite.sql.SqlKind; +import org.apache.calcite.sql.SqlWindowTableFunction; +import org.apache.calcite.sql.fun.SqlStdOperatorTable; +import org.apache.calcite.sql.type.SqlTypeUtil; +import org.apache.calcite.tools.RelBuilder; +import org.apache.calcite.util.ImmutableBitSet; + +import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.List; + +/** + * Rewrites the SESSION window table function into standard relational + * operators. + * + *

{@code SESSION(TABLE t, DESCRIPTOR(ts), DESCRIPTOR(k), gap)} returns + * the rows of t with two extra columns. Rows with the same key k whose + * timestamps are less than gap apart belong to the same session. For every + * row, window_start is the session's first timestamp and window_end is its + * last timestamp plus gap. The key descriptor is optional; without it all + * rows share one session timeline. Rows with a NULL timestamp are dropped. + * + * 

+ * LogicalTableFunctionScan(SESSION(DESCRIPTOR($ts), DESCRIPTOR($k), gap))
+ *   Input($0..$n-1)
+ *
+ * becomes ("sessionized" appears twice but is built once): + * 
+ * LogicalProject($0..$n-1, window_start=[$min], window_end=[$max + gap])
+ *   LogicalJoin(INNER, $k IS NOT DISTINCT FROM $k', $sid IS NOT DISTINCT FROM $sid')
+ *     sessionized
+ *     LogicalAggregate(group=[{$k, $sid}], min=[MIN($ts)], max=[MAX($ts)])
+ *       sessionized
+ *
+ * where "sessionized" numbers each row's session within its key: + * 
+ * LogicalProject($0..$n-1, sid=[SUM($brk) OVER w])
+ *   LogicalProject($0..$n-1, brk=[CASE($prev IS NULL OR $ts >= $prev + gap, 1, 0)])
+ *     LogicalProject($0..$n-1, prev=[LAG($ts) OVER w])
+ *       LogicalFilter($ts IS NOT NULL)      (only for a nullable column)
+ *         Input($0..$n-1)
+ *
+ * with w = PARTITION BY $k ORDER BY $ts RANGE UNBOUNDED PRECEDING. + * A row starts a new session (brk = 1) when it is the first of its key or + * follows its predecessor by gap or more, so the running sum of brk + * identifies the row's session. + * + *

With ties in $ts the RANGE frame sums brk over all peers, which + * assigns tied rows the same session; this matches the SESSION semantics, + * since rows with equal timestamps always share a session. + */ +public class SessionRewriteRule + extends RelRule> + implements TransformationRule { + public SessionRewriteRule() { + super(CONFIG); + } + + /** Column indexes of a DESCRIPTOR call, or null if 'node' is not one. */ + @Nullable + static List descriptorColumns(RexNode node) { + if (!(node instanceof RexCall call) || call.getKind() != SqlKind.DESCRIPTOR) + return null; + List result = new ArrayList<>(); + for (RexNode operand : call.getOperands()) { + if (!(operand instanceof RexInputRef ref)) + return null; + result.add(ref.getIndex()); + } + return result; + } + + @Override + public void onMatch(RelOptRuleCall call) { + final LogicalTableFunctionScan scan = call.rel(0); + if (!(scan.getCall() instanceof RexCall invocation)) + return; + if (!(invocation.getOperator() instanceof SqlWindowTableFunction) + || !invocation.getOperator().getName().equals("SESSION")) + return; + if (scan.getInputs().size() != 1) + return; + + // Operands after the TABLE argument: DESCRIPTOR(ts) [, DESCRIPTOR(k)], gap. + // A call with named arguments may hold DEFAULT for the omitted key. + final List operands = invocation.getOperands(); + if (operands.size() != 2 && operands.size() != 3) + return; + final List tsColumns = descriptorColumns(operands.get(0)); + if (tsColumns == null || tsColumns.size() != 1) + return; + final int tsIndex = tsColumns.get(0); + ImmutableBitSet keys = ImmutableBitSet.of(); + if (operands.size() == 3 && operands.get(1).getKind() != SqlKind.DEFAULT) { + final List keyColumns = descriptorColumns(operands.get(1)); + if (keyColumns == null) + return; + keys = ImmutableBitSet.of(keyColumns); + } + final RexNode gap = operands.get(operands.size() - 1); + if (!SqlTypeUtil.isInterval(gap.getType()) || RexUtil.containsInputRef(gap)) + return; + + final RelBuilder b = call.builder(); + final RelNode input = scan.getInput(0); + b.push(input); + final int n = input.getRowType().getFieldCount(); + if (input.getRowType().getFieldList().get(tsIndex).getType().isNullable()) + b.filter(b.isNotNull(b.field(tsIndex))); + + // prev = LAG(ts) OVER w, appended as field $n + final List keyList = keys.toList(); + b.projectPlus( + b.aggregateCall(SqlStdOperatorTable.LAG, b.field(tsIndex)) + .over() + .partitionBy(b.fields(keyList)) + .orderBy(b.field(tsIndex)) + .rangeTo(RexWindowBounds.CURRENT_ROW) + .toRex()); + + // brk = CASE(prev IS NULL OR ts >= prev + gap, 1, 0), replaces prev as field $n + final RexNode prev = b.field(n); + b.project(replaceLast(b.fields(), n, + b.call(SqlStdOperatorTable.CASE, + b.or(b.isNull(prev), + b.call(SqlStdOperatorTable.GREATER_THAN_OR_EQUAL, + b.field(tsIndex), + b.call(SqlStdOperatorTable.DATETIME_PLUS, prev, gap))), + b.literal(1), b.literal(0)))); + + // sid = SUM(brk) OVER w, replaces brk as field $n + b.project(replaceLast(b.fields(), n, + b.aggregateCall(SqlStdOperatorTable.SUM, b.field(n)) + .over() + .partitionBy(b.fields(keyList)) + .orderBy(b.field(tsIndex)) + .rangeTo(RexWindowBounds.CURRENT_ROW) + .toRex())); + final RelNode sessionized = b.build(); + + // bounds = [k..., sid, MIN(ts), MAX(ts)] grouped by (k, sid) + final ImmutableBitSet group = keys.union(ImmutableBitSet.of(n)); + b.push(sessionized) + .aggregate(b.groupKey(group), + b.min(b.field(tsIndex)), b.max(b.field(tsIndex))); + final RelNode bounds = b.build(); + + // Attach each session's bounds to each of its rows. The join keys + // use IS NOT DISTINCT FROM because the key columns may hold NULL. + b.push(sessionized).push(bounds); + final int groupCount = group.cardinality(); + final List conditions = new ArrayList<>(); + for (int i = 0; i < keyList.size(); i++) + conditions.add(b.call(SqlStdOperatorTable.IS_NOT_DISTINCT_FROM, + b.field(2, 0, keyList.get(i)), b.field(2, 1, i))); + conditions.add(b.call(SqlStdOperatorTable.IS_NOT_DISTINCT_FROM, + b.field(2, 0, n), b.field(2, 1, groupCount - 1))); + b.join(JoinRelType.INNER, b.and(conditions)); + + // Final project: sessionized $0..$n, bounds keys and sid, min, max + final int boundsBase = (n + 1) + groupCount; + final List results = new ArrayList<>(b.fields().subList(0, n)); + results.add(b.field(boundsBase)); + results.add(b.call(SqlStdOperatorTable.DATETIME_PLUS, b.field(boundsBase + 1), gap)); + b.project(results, scan.getRowType().getFieldNames()) + .convert(scan.getRowType(), false); + + call.transformTo(b.build()); + // prune if this ever runs under Volcano + call.getPlanner().prune(scan); + } + + /** The first 'keep' fields, followed by 'last'. */ + static List replaceLast(List fields, int keep, RexNode last) { + List result = new ArrayList<>(fields.subList(0, keep)); + result.add(last); + return result; + } + + public static final DefaultOptRuleConfig CONFIG = + DefaultOptRuleConfig.create() + .withOperandSupplier( + b -> b.operand(LogicalTableFunctionScan.class).anyInputs()); +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DumpTopology.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DumpTopology.java new file mode 100644 index 00000000000..d8a0d01119d --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/DumpTopology.java @@ -0,0 +1,74 @@ +package org.dbsp.sqlCompiler.compiler.visitors.outer; + +import org.dbsp.sqlCompiler.circuit.OutputPort; +import org.dbsp.sqlCompiler.circuit.operator.DBSPOperator; +import org.dbsp.sqlCompiler.circuit.operator.IGCOperator; +import org.dbsp.sqlCompiler.compiler.DBSPCompiler; +import org.dbsp.sqlCompiler.ir.IDBSPOuterNode; + +import java.util.function.Predicate; + +/** Debugging helper. + * Dumps the circuit topology as text, one line per operator, in visit order: + * 

+ * === at X ===
+ *   748 JoinIndex <- [747:0, 731:0]
+ *   749 IntegrateTraceRetainNValues <- [748:0, 745:0]  GC
+ *
+ * Insert it between passes to inspect the graph without generating images: + * 
+ * this.add(new DumpTopology(compiler, "at X"));
+ *
+ */ +public class DumpTopology extends CircuitVisitor { + private final String label; + /** Restrict dump to operators selected by this predicate */ + private final Predicate filter; + + public DumpTopology(DBSPCompiler compiler, String label) { + this(compiler, label, op -> true); + } + + public DumpTopology(DBSPCompiler compiler, String label, Predicate filter) { + super(compiler); + this.label = label; + this.filter = filter; + } + + /** Dumps only the garbage-collection operators and their sources. */ + public static DumpTopology gcOnly(DBSPCompiler compiler, String label) { + return new DumpTopology(compiler, label, op -> op.is(IGCOperator.class)); + } + + @Override + public Token startVisit(IDBSPOuterNode node) { + System.out.println("=== " + this.label + " ==="); + return super.startVisit(node); + } + + @Override + public void postorder(DBSPOperator operator) { + if (!this.filter.test(operator)) + return; + StringBuilder line = new StringBuilder(); + line.append(" ").append(operator.id).append(" ") + .append(operator.getClass().getSimpleName().replace("DBSP", "").replace("Operator", "")) + .append(" <- ["); + boolean first = true; + for (OutputPort input : operator.inputs) { + if (!first) + line.append(", "); + first = false; + line.append(input.node().id).append(":").append(input.port()); + } + line.append("]"); + if (operator.is(IGCOperator.class)) + line.append(" GC"); + System.out.println(line); + } + + @Override + public String toString() { + return "DumpTopology(" + this.label + ")"; + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/RemoveIdentityOperators.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/RemoveIdentityOperators.java index 3e51c298c80..84dd56978ce 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/RemoveIdentityOperators.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/RemoveIdentityOperators.java @@ -1,10 +1,19 @@ package org.dbsp.sqlCompiler.compiler.visitors.outer; import org.dbsp.sqlCompiler.circuit.OutputPort; +import org.dbsp.sqlCompiler.circuit.operator.DBSPBinaryOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPIntegrateTraceRetainKeysOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPIntegrateTraceRetainNValuesOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPIntegrateTraceRetainValuesOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPMapIndexOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPMapOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPNoopOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPSimpleOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPUnaryOperator; import org.dbsp.sqlCompiler.compiler.DBSPCompiler; import org.dbsp.sqlCompiler.compiler.visitors.inner.EquivalenceContext; +import org.dbsp.sqlCompiler.ir.IDBSPOuterNode; import org.dbsp.sqlCompiler.ir.expression.DBSPClosureExpression; import org.dbsp.sqlCompiler.ir.expression.DBSPRawTupleExpression; import org.dbsp.sqlCompiler.ir.expression.DBSPTupleExpression; @@ -14,9 +23,44 @@ import org.dbsp.sqlCompiler.ir.type.derived.DBSPTypeRef; import org.dbsp.sqlCompiler.ir.type.derived.DBSPTypeTupleBase; -public class RemoveIdentityOperators extends CircuitCloneVisitor { +import javax.annotation.Nullable; +import java.util.ArrayList; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Removes Map and MapIndex operators that compute the identity function. + * + *

Such an operator can only be removed if its garbage-collection operators + * can move to its input. {@link Decide} scans the graph and picks, for each + * input and dimension, the one operator that may move; {@link RemoveOrReplace} then + * rewrites the circuit. Operators that cannot be removed become + * {@link DBSPNoopOperator}s. */ +public class RemoveIdentityOperators extends Passes { public RemoveIdentityOperators(DBSPCompiler compiler) { - super(compiler, false); + super("RemoveIdentityOperators", compiler); + Graph graph = new Graph(compiler); + this.add(graph); + Decide decide = new Decide(compiler, graph.getGraphs()); + this.add(decide); + this.add(new RemoveOrReplace(compiler, decide.remove)); + } + + enum GCKind { + KEYS, + VALUES + } + + @Nullable + static GCKind gcKind(DBSPOperator operator) { + if (operator.is(DBSPIntegrateTraceRetainKeysOperator.class)) + return GCKind.KEYS; + if (operator.is(DBSPIntegrateTraceRetainValuesOperator.class) || + operator.is(DBSPIntegrateTraceRetainNValuesOperator.class)) + return GCKind.VALUES; + return null; } /** Check whether a closure is an "identity" function for a Map or MapIndex operator. @@ -62,37 +106,127 @@ public static boolean isIdentityFunction(DBSPClosureExpression expression) { } } - @Override - public void postorder(DBSPMapOperator operator) { - if (operator.function == null || !operator.function.is(DBSPClosureExpression.class)) { - super.postorder(operator); - return; + /** True for a Map or MapIndex operator that computes the identity. */ + static boolean isIdentityOperator(DBSPOperator operator) { + if (!operator.is(DBSPMapOperator.class) && !operator.is(DBSPMapIndexOperator.class)) + return false; + DBSPSimpleOperator simple = operator.to(DBSPSimpleOperator.class); + if (simple.function == null || !simple.function.is(DBSPClosureExpression.class)) + return false; + return isIdentityFunction(simple.getClosureFunction()); + } + + /** Decides which identity operators can be removed. + * + *

Removing an operator moves GC operators attached to its output to its input; + * a node cannot have incompatible GC operators. */ + static class Decide extends CircuitWithGraphsVisitor { + /** Operators to remove, mapped to true if it may be removed and + * false if it becomes a noop */ + public final Map remove = new HashMap<>(); + final List candidates = new ArrayList<>(); + + /** @param toMove GC kinds that removal would move to the input. + * @param existing GC kinds that exist on the input. */ + record Candidate(DBSPUnaryOperator operator, OutputPort input, + Set toMove, Set existing) {} + + Decide(DBSPCompiler compiler, CircuitGraphs graphs) { + super(compiler, graphs); } - DBSPClosureExpression function = operator.getClosureFunction(); - if (isIdentityFunction(function)) { - OutputPort input = this.mapped(operator.input()); - this.map(operator.outputPort(), input, false); - return; + @Override + public Token startVisit(IDBSPOuterNode node) { + this.remove.clear(); + this.candidates.clear(); + return super.startVisit(node); + } + + /** What kinds of GC operators are attached to an existing port? */ + Set getGCKinds(OutputPort port) { + Set result = EnumSet.noneOf(GCKind.class); + for (var successor : this.getGraph().getSuccessors(port.node())) { + if (successor.port() != 0) + continue; + GCKind retention = gcKind(successor.node()); + if (retention != null + && successor.node().to(DBSPBinaryOperator.class).left().equals(port)) + result.add(retention); + } + return result; } - super.postorder(operator); + @Override + public void postorder(DBSPMapOperator operator) { + this.consider(operator); + } + + @Override + public void postorder(DBSPMapIndexOperator operator) { + this.consider(operator); + } + + void consider(DBSPUnaryOperator operator) { + if (!isIdentityOperator(operator)) + return; + OutputPort input = operator.input(); + this.candidates.add(new Candidate(operator, input, + this.getGCKinds(operator.outputPort()), + this.getGCKinds(input))); + } + + @Override + public void endVisit() { + Map> claimed = new HashMap<>(); + for (Candidate candidate : this.candidates) { + if (candidate.toMove().isEmpty()) { + this.remove.put(candidate.operator(), true); + continue; + } + Set taken = + claimed.computeIfAbsent(candidate.input(), p -> EnumSet.copyOf(candidate.existing())); + boolean canRemove = taken.stream().noneMatch(candidate.toMove()::contains); + if (canRemove) + taken.addAll(candidate.toMove()); + this.remove.put(candidate.operator(), canRemove); + } + super.endVisit(); + } } - @Override - public void postorder(DBSPMapIndexOperator operator) { - if (operator.function == null || !operator.function.is(DBSPClosureExpression.class)) { - super.postorder(operator); - return; + /** Applies the decisions taken by {@link Decide}. */ + static class RemoveOrReplace extends CircuitCloneVisitor { + final Map actions; + + RemoveOrReplace(DBSPCompiler compiler, Map actions) { + super(compiler, false); + this.actions = actions; } - DBSPClosureExpression function = operator.getClosureFunction(); - if (isIdentityFunction(function)) { + boolean replaceIdentity(DBSPUnaryOperator operator) { + Boolean remove = this.actions.get(operator); + if (remove == null) + // Not an identity operator + return false; OutputPort input = this.mapped(operator.input()); - this.map(operator.outputPort(), input, false); - return; + if (remove) { + this.map(operator.outputPort(), input, false); + } else { + this.map(operator, new DBSPNoopOperator(operator.getRelNode(), input)); + } + return true; } - super.postorder(operator); + @Override + public void postorder(DBSPMapOperator operator) { + if (!this.replaceIdentity(operator)) + super.postorder(operator); + } + + @Override + public void postorder(DBSPMapIndexOperator operator) { + if (!this.replaceIdentity(operator)) + super.postorder(operator); + } } } 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 f42093ae855..47d5e6a070f 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 @@ -1,5 +1,6 @@ package org.dbsp.sqlCompiler.compiler.visitors.outer; +import org.dbsp.sqlCompiler.circuit.OutputPort; import org.dbsp.sqlCompiler.circuit.operator.DBSPAggregateLinearPostprocessRetainKeysOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPAggregateOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPBinaryOperator; @@ -16,12 +17,34 @@ import org.dbsp.sqlCompiler.compiler.errors.InternalCompilerError; import org.dbsp.util.graph.Port; +import java.util.HashMap; +import java.util.Map; + /** Check if all GC operators have an obvious operator they apply to */ public class StrayGC extends CircuitWithGraphsVisitor { + /** Maps each port that has a value-retention operator to that operator. */ + final Map valueRetainers = new HashMap<>(); + /** Maps each port that has a key-retention operator to that operator. */ + final Map keyRetainers = new HashMap<>(); + public StrayGC(DBSPCompiler compiler, CircuitGraphs g) { super(compiler, g); } + /** Check that 'operator' is the only retention of its kind attached to + * its source. + * The same operator can be visited twice, hence the identity test. */ + void checkSingleRetainer(DBSPBinaryOperator operator, + Map retainers, String what) { + DBSPOperator previous = retainers.put(operator.left(), operator); + if (previous != null && previous != operator) { + throw new InternalCompilerError( + "Operators " + previous + " and " + operator + + " both garbage-collect the " + what + " of " + operator.left().operator + + "; the runtime honors only one " + what + " retention condition per trace"); + } + } + /** 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 @@ -63,6 +86,7 @@ void check(DBSPBinaryOperator operator) { public void postorder(DBSPIntegrateTraceRetainValuesOperator operator) { // This operator always uses the accumulate_ variant this.checkAccumulate(operator, true); + this.checkSingleRetainer(operator, this.valueRetainers, "values"); this.check(operator); } @@ -70,12 +94,14 @@ public void postorder(DBSPIntegrateTraceRetainValuesOperator operator) { public void postorder(DBSPIntegrateTraceRetainNValuesOperator operator) { // This operator always uses the accumulate_ variant this.checkAccumulate(operator, true); + this.checkSingleRetainer(operator, this.valueRetainers, "values"); this.check(operator); } @Override public void postorder(DBSPIntegrateTraceRetainKeysOperator operator) { this.checkAccumulate(operator, operator.accumulate); + this.checkSingleRetainer(operator, this.keyRetainers, "keys"); DBSPOperator left = operator.left().operator; if (left.is(DBSPAggregateLinearPostprocessRetainKeysOperator.class) || left.is(DBSPChainAggregateOperator.class) || diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/SessionTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/SessionTests.java new file mode 100644 index 00000000000..adee3fd640d --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/quidem/SessionTests.java @@ -0,0 +1,227 @@ +package org.dbsp.sqlCompiler.compiler.sql.quidem; + +import org.dbsp.sqlCompiler.compiler.DBSPCompiler; +import org.dbsp.sqlCompiler.compiler.sql.tools.SqlIoTest; +import org.junit.Test; + +// Based on stream.iq from Calcite +public class SessionTests extends SqlIoTest { + @Override + public void prepareInputs(DBSPCompiler compiler) { + String sql = """ + CREATE TABLE orders( + rowtime TIMESTAMP NOT NULL, + id INTEGER, + product VARCHAR, + units INTEGER + ); + + INSERT INTO orders VALUES + ('2015-02-15 10:15:00', 1, 'paint', 10), + ('2015-02-15 10:24:15', 2, 'paper', 5), + ('2015-02-15 10:24:45', 3, 'brush', 12), + ('2015-02-15 10:58:00', 4, 'paint', 3), + ('2015-02-15 11:10:00', 5, 'paint', 3); + + CREATE TABLE events( + ts TIMESTAMP, + uid VARCHAR + ); + + INSERT INTO events VALUES + (NULL, 'a'), + ('2020-01-01 10:00:00', 'a'), + ('2020-01-01 10:05:00', 'a'), + ('2020-01-01 10:30:00', 'a'), + ('2020-01-01 10:00:00', 'b'), + ('2020-01-01 10:15:00', 'b'), + ('2020-01-01 10:00:00', NULL), + ('2020-01-01 10:14:59', NULL), + ('2020-01-01 10:35:00', NULL); + + -- Columns before, between and after the two columns that + -- SESSION uses, to check that the rewrite keeps them in place + CREATE TABLE surrounded( + before INTEGER, + ts TIMESTAMP NOT NULL, + between VARCHAR, + k VARCHAR, + after INTEGER + ); + + INSERT INTO surrounded VALUES + (1, '2020-01-01 10:00:00', 'x', 'k1', 100), + (2, '2020-01-01 10:05:00', 'y', 'k1', 200), + (3, '2020-01-01 10:30:00', 'z', 'k1', 300), + (4, '2020-01-01 10:00:00', 'w', 'k2', 400);"""; + compiler.submitStatementsForCompilation(sql); + } + + // Expected output taken from Calcite's stream.iq and validated with an + // external Python implementation of the SESSION semantics + @Test + public void testSession() { + this.qst(""" + SELECT * FROM TABLE(SESSION(TABLE ORDERS, DESCRIPTOR(ROWTIME), DESCRIPTOR(PRODUCT), INTERVAL '20' MINUTE)); + +---------------------+----+---------+-------+-------------------------+-------------------------+ + | ROWTIME | ID | PRODUCT | UNITS | window_start | window_end | + +---------------------+----+---------+-------+-------------------------+-------------------------+ + | 2015-02-15 10:15:00 | 1 | paint | 10 | 2015-02-15 10:15:00.000 | 2015-02-15 10:35:00.000 | + | 2015-02-15 10:24:15 | 2 | paper | 5 | 2015-02-15 10:24:15.000 | 2015-02-15 10:44:15.000 | + | 2015-02-15 10:24:45 | 3 | brush | 12 | 2015-02-15 10:24:45.000 | 2015-02-15 10:44:45.000 | + | 2015-02-15 10:58:00 | 4 | paint | 3 | 2015-02-15 10:58:00.000 | 2015-02-15 11:30:00.000 | + | 2015-02-15 11:10:00 | 5 | paint | 3 | 2015-02-15 10:58:00.000 | 2015-02-15 11:30:00.000 | + +---------------------+----+---------+-------+-------------------------+-------------------------+ + (5 rows) + + SELECT * FROM TABLE( + SESSION( + DATA => TABLE ORDERS, + TIMECOL => DESCRIPTOR(ROWTIME), + KEY => DESCRIPTOR(PRODUCT), + SIZE => INTERVAL '20' MINUTE)); + +---------------------+----+---------+-------+-------------------------+-------------------------+ + | ROWTIME | ID | PRODUCT | UNITS | window_start | window_end | + +---------------------+----+---------+-------+-------------------------+-------------------------+ + | 2015-02-15 10:15:00 | 1 | paint | 10 | 2015-02-15 10:15:00.000 | 2015-02-15 10:35:00.000 | + | 2015-02-15 10:24:15 | 2 | paper | 5 | 2015-02-15 10:24:15.000 | 2015-02-15 10:44:15.000 | + | 2015-02-15 10:24:45 | 3 | brush | 12 | 2015-02-15 10:24:45.000 | 2015-02-15 10:44:45.000 | + | 2015-02-15 10:58:00 | 4 | paint | 3 | 2015-02-15 10:58:00.000 | 2015-02-15 11:30:00.000 | + | 2015-02-15 11:10:00 | 5 | paint | 3 | 2015-02-15 10:58:00.000 | 2015-02-15 11:30:00.000 | + +---------------------+----+---------+-------+-------------------------+-------------------------+ + (5 rows) + + SELECT * FROM TABLE(SESSION((SELECT * FROM ORDERS), DESCRIPTOR(ROWTIME), DESCRIPTOR(PRODUCT), INTERVAL '20' MINUTE)); + +---------------------+----+---------+-------+-------------------------+-------------------------+ + | ROWTIME | ID | PRODUCT | UNITS | window_start | window_end | + +---------------------+----+---------+-------+-------------------------+-------------------------+ + | 2015-02-15 10:15:00 | 1 | paint | 10 | 2015-02-15 10:15:00.000 | 2015-02-15 10:35:00.000 | + | 2015-02-15 10:24:15 | 2 | paper | 5 | 2015-02-15 10:24:15.000 | 2015-02-15 10:44:15.000 | + | 2015-02-15 10:24:45 | 3 | brush | 12 | 2015-02-15 10:24:45.000 | 2015-02-15 10:44:45.000 | + | 2015-02-15 10:58:00 | 4 | paint | 3 | 2015-02-15 10:58:00.000 | 2015-02-15 11:30:00.000 | + | 2015-02-15 11:10:00 | 5 | paint | 3 | 2015-02-15 10:58:00.000 | 2015-02-15 11:30:00.000 | + +---------------------+----+---------+-------+-------------------------+-------------------------+ + (5 rows)"""); + } + + // Without a key descriptor all rows share one session timeline. + // Expected output validated with an external Python implementation of the + // SESSION semantics; Calcite cannot execute the keyless form + @Test + public void testSessionNoKey() { + this.qst(""" + SELECT * FROM TABLE(SESSION(TABLE ORDERS, DESCRIPTOR(ROWTIME), INTERVAL '20' MINUTE)); + +---------------------+----+---------+-------+-------------------------+-------------------------+ + | ROWTIME | ID | PRODUCT | UNITS | window_start | window_end | + +---------------------+----+---------+-------+-------------------------+-------------------------+ + | 2015-02-15 10:15:00 | 1 | paint | 10 | 2015-02-15 10:15:00.000 | 2015-02-15 10:44:45.000 | + | 2015-02-15 10:24:15 | 2 | paper | 5 | 2015-02-15 10:15:00.000 | 2015-02-15 10:44:45.000 | + | 2015-02-15 10:24:45 | 3 | brush | 12 | 2015-02-15 10:15:00.000 | 2015-02-15 10:44:45.000 | + | 2015-02-15 10:58:00 | 4 | paint | 3 | 2015-02-15 10:58:00.000 | 2015-02-15 11:30:00.000 | + | 2015-02-15 11:10:00 | 5 | paint | 3 | 2015-02-15 10:58:00.000 | 2015-02-15 11:30:00.000 | + +---------------------+----+---------+-------+-------------------------+-------------------------+ + (5 rows) + + SELECT * FROM TABLE( + SESSION( + DATA => TABLE ORDERS, + TIMECOL => DESCRIPTOR(ROWTIME), + SIZE => INTERVAL '20' MINUTE)); + +---------------------+----+---------+-------+-------------------------+-------------------------+ + | ROWTIME | ID | PRODUCT | UNITS | window_start | window_end | + +---------------------+----+---------+-------+-------------------------+-------------------------+ + | 2015-02-15 10:15:00 | 1 | paint | 10 | 2015-02-15 10:15:00.000 | 2015-02-15 10:44:45.000 | + | 2015-02-15 10:24:15 | 2 | paper | 5 | 2015-02-15 10:15:00.000 | 2015-02-15 10:44:45.000 | + | 2015-02-15 10:24:45 | 3 | brush | 12 | 2015-02-15 10:15:00.000 | 2015-02-15 10:44:45.000 | + | 2015-02-15 10:58:00 | 4 | paint | 3 | 2015-02-15 10:58:00.000 | 2015-02-15 11:30:00.000 | + | 2015-02-15 11:10:00 | 5 | paint | 3 | 2015-02-15 10:58:00.000 | 2015-02-15 11:30:00.000 | + +---------------------+----+---------+-------+-------------------------+-------------------------+ + (5 rows)"""); + } + + // Rows with a NULL timestamp are dropped; NULL keys sessionize like any + // other key value; rows exactly gap apart belong to different sessions. + // Expected output validated with an external Python implementation of the + // SESSION semantics + @Test + public void testSessionNulls() { + this.qst(""" + SELECT * FROM TABLE(SESSION(TABLE EVENTS, DESCRIPTOR(TS), DESCRIPTOR(UID), INTERVAL '15' MINUTE)); + +---------------------+------+-------------------------+-------------------------+ + | TS | UID | window_start | window_end | + +---------------------+------+-------------------------+-------------------------+ + | 2020-01-01 10:00:00 | a | 2020-01-01 10:00:00.000 | 2020-01-01 10:20:00.000 | + | 2020-01-01 10:05:00 | a | 2020-01-01 10:00:00.000 | 2020-01-01 10:20:00.000 | + | 2020-01-01 10:30:00 | a | 2020-01-01 10:30:00.000 | 2020-01-01 10:45:00.000 | + | 2020-01-01 10:00:00 | b | 2020-01-01 10:00:00.000 | 2020-01-01 10:15:00.000 | + | 2020-01-01 10:15:00 | b | 2020-01-01 10:15:00.000 | 2020-01-01 10:30:00.000 | + | 2020-01-01 10:00:00 |NULL | 2020-01-01 10:00:00.000 | 2020-01-01 10:29:59.000 | + | 2020-01-01 10:14:59 |NULL | 2020-01-01 10:00:00.000 | 2020-01-01 10:29:59.000 | + | 2020-01-01 10:35:00 |NULL | 2020-01-01 10:35:00.000 | 2020-01-01 10:50:00.000 | + +---------------------+------+-------------------------+-------------------------+ + (8 rows)"""); + } + + // The timestamp and the key are neither the first nor the last column, so + // the rewrite has to carry the columns around them through unchanged. + // Expected output validated with an external Python implementation of the + // SESSION semantics + @Test + public void testSessionSurroundedColumns() { + this.qst(""" + SELECT * FROM TABLE(SESSION(TABLE SURROUNDED, DESCRIPTOR(TS), DESCRIPTOR(K), INTERVAL '15' MINUTE)); + +--------+---------------------+---------+----+-------+-------------------------+-------------------------+ + | BEFORE | TS | BETWEEN | K | AFTER | window_start | window_end | + +--------+---------------------+---------+----+-------+-------------------------+-------------------------+ + | 1 | 2020-01-01 10:00:00 | x | k1 | 100 | 2020-01-01 10:00:00.000 | 2020-01-01 10:20:00.000 | + | 2 | 2020-01-01 10:05:00 | y | k1 | 200 | 2020-01-01 10:00:00.000 | 2020-01-01 10:20:00.000 | + | 3 | 2020-01-01 10:30:00 | z | k1 | 300 | 2020-01-01 10:30:00.000 | 2020-01-01 10:45:00.000 | + | 4 | 2020-01-01 10:00:00 | w | k2 | 400 | 2020-01-01 10:00:00.000 | 2020-01-01 10:15:00.000 | + +--------+---------------------+---------+----+-------+-------------------------+-------------------------+ + (4 rows)"""); + } + + @Test + public void testSessionNegative() { + // The gap must be an interval + this.statementsFailingInCompilation(""" + CREATE VIEW V AS SELECT * FROM TABLE( + SESSION(TABLE ORDERS, DESCRIPTOR(ROWTIME), DESCRIPTOR(PRODUCT), 10))""", + "Cannot apply 'SESSION' to arguments"); + // The time column must have a timestamp type + this.statementsFailingInCompilation(""" + CREATE VIEW V AS SELECT * FROM TABLE( + SESSION(TABLE ORDERS, DESCRIPTOR(PRODUCT), DESCRIPTOR(ID), INTERVAL '20' MINUTE))""", + "Cannot apply 'SESSION' to arguments"); + // The descriptors must name existing columns + this.statementsFailingInCompilation(""" + CREATE VIEW V AS SELECT * FROM TABLE( + SESSION(TABLE ORDERS, DESCRIPTOR(NO_SUCH_COLUMN), DESCRIPTOR(PRODUCT), INTERVAL '20' MINUTE))""", + "Unknown identifier"); + // Missing gap argument + this.statementsFailingInCompilation(""" + CREATE VIEW V AS SELECT * FROM TABLE( + SESSION(TABLE ORDERS, DESCRIPTOR(ROWTIME)))""", + "Invalid number of arguments"); + } + + // Aggregation on top of SESSION, the typical use of the table function. + // Expected output validated with an external Python implementation of the + // SESSION semantics + @Test + public void testSessionAggregate() { + this.qst(""" + SELECT PRODUCT, COUNT(*) AS event_count, window_start, window_end + FROM TABLE(SESSION(TABLE ORDERS, DESCRIPTOR(ROWTIME), DESCRIPTOR(PRODUCT), INTERVAL '20' MINUTE)) + GROUP BY PRODUCT, window_start, window_end; + +---------+-------------+-------------------------+-------------------------+ + | PRODUCT | event_count | window_start | window_end | + +---------+-------------+-------------------------+-------------------------+ + | brush | 1 | 2015-02-15 10:24:45.000 | 2015-02-15 10:44:45.000 | + | paint | 1 | 2015-02-15 10:15:00.000 | 2015-02-15 10:35:00.000 | + | paint | 2 | 2015-02-15 10:58:00.000 | 2015-02-15 11:30:00.000 | + | paper | 1 | 2015-02-15 10:24:15.000 | 2015-02-15 10:44:15.000 | + +---------+-------------+-------------------------+-------------------------+ + (4 rows)"""); + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/streaming/StreamingTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/streaming/StreamingTests.java index 799f94321d5..09b4635025d 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/streaming/StreamingTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/streaming/StreamingTests.java @@ -870,6 +870,224 @@ public void endVisit() { }); } + @Test + public void sessionGc() { + // LATENESS on the SESSION timestamp column with SESSION windows. + // Both RetainNValues operators attach to the same JoinIndex (the + // LAG); the steps below run with compaction to check that this works. + String sql = """ + CREATE TABLE events( + ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOURS, + uid VARCHAR + ); + CREATE VIEW sessions AS + SELECT uid, COUNT(*) AS cnt, window_start, window_end + FROM TABLE(SESSION(TABLE events, DESCRIPTOR(ts), DESCRIPTOR(uid), INTERVAL 10 MINUTES)) + GROUP BY uid, window_start, window_end;"""; + CompilerCircuitStream ccs = this.getCCS(sql).compactAfterEachStep().withStringTrim(); + // TODO: window_start cannot have a waterline (a session can be unbounded) + // but window_end does, but the algorithm we use does not find it. + ccs.visit(new CircuitVisitor(ccs.compiler) { + int retainKeys = 0; + int retainNValues = 0; + int rollingWithWaterline = 0; + + @Override + public void postorder(DBSPIntegrateTraceRetainKeysOperator operator) { + this.retainKeys++; + } + + @Override + public void postorder(DBSPIntegrateTraceRetainNValuesOperator operator) { + this.retainNValues++; + } + + @Override + public void postorder(DBSPPartitionedRollingAggregateWithWaterlineOperator operator) { + this.rollingWithWaterline++; + } + + @Override + public void endVisit() { + Assert.assertEquals(1, this.rollingWithWaterline); + Assert.assertEquals(2, this.retainKeys); + Assert.assertEquals(2, this.retainNValues); + } + }); + // The waterline is max over all data of (ts - 1 hour); each step is + // filtered with the waterline computed from the previous steps. + // Before this step: waterline = minimum, table is empty. + // Two sessions start: 'a' has two events 5 minutes apart, 'b' one event + ccs.step(""" + INSERT INTO events VALUES('2020-01-01 10:00:00', 'a'), + ('2020-01-01 10:05:00', 'a'), + ('2020-01-01 10:00:00', 'b');""", """ + uid | cnt | window_start | window_end | weight + ---------------------------------------------------------------- + a | 2 | 2020-01-01 10:00:00 | 2020-01-01 10:15:00 | 1 + b | 1 | 2020-01-01 10:00:00 | 2020-01-01 10:10:00 | 1"""); + // Before this step, table contents (rows above the ==== line are + // below the waterline): + // ts | uid | + // -------+-----+ + // ============== waterline = 10:05 - 1:00 = 09:05 + // 10:00 | a | + // 10:00 | b | + // 10:05 | a | + // 7 minutes after the last 'a' event: extends the 'a' session + ccs.step(""" + INSERT INTO events VALUES('2020-01-01 10:12:00', 'a');""", """ + uid | cnt | window_start | window_end | weight + ---------------------------------------------------------------- + a | 2 | 2020-01-01 10:00:00 | 2020-01-01 10:15:00 | -1 + a | 3 | 2020-01-01 10:00:00 | 2020-01-01 10:22:00 | 1"""); + // Before this step: + // ts | uid | + // -------+-----+ + // ============== waterline = 10:12 - 1:00 = 09:12 + // 10:00 | a | + // 10:00 | b | + // 10:05 | a | + // 10:12 | a | + // Far from the previous event: a new session + ccs.step(""" + INSERT INTO events VALUES('2020-01-01 13:00:00', 'a');""", """ + uid | cnt | window_start | window_end | weight + ----------------------------------------------------------------- + a | 1 | 2020-01-01 13:00:00 | 2020-01-01 13:10:00 | 1"""); + // Before this step: + // ts | uid | + // -------+-----+ + // 10:00 | a | frozen + // 10:00 | b | frozen + // 10:05 | a | frozen + // 10:12 | a | frozen + // ============== waterline = 13:00 - 1:00 = 12:00 + // 13:00 | a | + // The frozen sessions can no longer change; compaction may collect + // their state. This row is below the waterline: late, dropped + ccs.step(""" + INSERT INTO events VALUES('2020-01-01 11:30:00', 'a');""", """ + uid | cnt | window_start | window_end | weight + ------------------------------------------------"""); + // Before this step: same contents and waterline as the previous step. + // Exactly on the waterline: not late; more than one gap away from + // both neighbor sessions, so it forms its own + ccs.step(""" + INSERT INTO events VALUES('2020-01-01 12:00:00', 'a');""", """ + uid | cnt | window_start | window_end | weight + ---------------------------------------------------------------- + a | 1 | 2020-01-01 12:00:00 | 2020-01-01 12:10:00 | 1"""); + } + + @Test + public void sessionDelete() { + // Sessions react to deletions: removing a row can split a session in + // two, or move its start. No LATENESS. + String sql = """ + CREATE TABLE events( + ts TIMESTAMP NOT NULL, + uid VARCHAR + ); + CREATE VIEW sessions AS + SELECT uid, COUNT(*) AS cnt, window_start, window_end + FROM TABLE(SESSION(TABLE events, DESCRIPTOR(ts), DESCRIPTOR(uid), INTERVAL 10 MINUTES)) + GROUP BY uid, window_start, window_end;"""; + CompilerCircuitStream ccs = this.getCCS(sql).withStringTrim(); + // 10:05 and 10:20 are 15 minutes apart, so there are two sessions + ccs.step(""" + INSERT INTO events VALUES('2020-01-01 10:00:00', 'a'), + ('2020-01-01 10:05:00', 'a'), + ('2020-01-01 10:20:00', 'a');""", """ + uid | cnt | window_start | window_end | weight + ---------------------------------------------------------------- + a | 2 | 2020-01-01 10:00:00 | 2020-01-01 10:15:00 | 1 + a | 1 | 2020-01-01 10:20:00 | 2020-01-01 10:30:00 | 1"""); + // 10:12 is within the gap of both neighbors, so it bridges the two + // sessions into one + ccs.step(""" + INSERT INTO events VALUES('2020-01-01 10:12:00', 'a');""", """ + uid | cnt | window_start | window_end | weight + ---------------------------------------------------------------- + a | 2 | 2020-01-01 10:00:00 | 2020-01-01 10:15:00 | -1 + a | 1 | 2020-01-01 10:20:00 | 2020-01-01 10:30:00 | -1 + a | 4 | 2020-01-01 10:00:00 | 2020-01-01 10:30:00 | 1"""); + // Removing the bridge splits the session again + ccs.step(""" + REMOVE FROM events VALUES('2020-01-01 10:12:00', 'a');""", """ + uid | cnt | window_start | window_end | weight + ---------------------------------------------------------------- + a | 4 | 2020-01-01 10:00:00 | 2020-01-01 10:30:00 | -1 + a | 2 | 2020-01-01 10:00:00 | 2020-01-01 10:15:00 | 1 + a | 1 | 2020-01-01 10:20:00 | 2020-01-01 10:30:00 | 1"""); + // Removing the first row of a session moves its window_start + ccs.step(""" + REMOVE FROM events VALUES('2020-01-01 10:00:00', 'a');""", """ + uid | cnt | window_start | window_end | weight + ---------------------------------------------------------------- + a | 2 | 2020-01-01 10:00:00 | 2020-01-01 10:15:00 | -1 + a | 1 | 2020-01-01 10:05:00 | 2020-01-01 10:15:00 | 1"""); + // Removing the last remaining row of a session deletes it + ccs.step(""" + REMOVE FROM events VALUES('2020-01-01 10:20:00', 'a');""", """ + uid | cnt | window_start | window_end | weight + ---------------------------------------------------------------- + a | 1 | 2020-01-01 10:20:00 | 2020-01-01 10:30:00 | -1"""); + } + + @Test + public void sessionGcLateMerge() { + // A session that straddles the waterline can still merge with an + // on-time row; the merged session's window_start comes from a value + // below the waterline, so value retention must keep it. + String sql = """ + CREATE TABLE events( + ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOURS, + uid VARCHAR + ); + CREATE VIEW sessions AS + SELECT uid, COUNT(*) AS cnt, window_start, window_end + FROM TABLE(SESSION(TABLE events, DESCRIPTOR(ts), DESCRIPTOR(uid), INTERVAL 10 MINUTES)) + GROUP BY uid, window_start, window_end;"""; + CompilerCircuitStream ccs = this.getCCS(sql).compactAfterEachStep().withStringTrim(); + // Before this step: waterline = minimum, table is empty + ccs.step(""" + INSERT INTO events VALUES('2020-01-01 11:50:00', 'a'), + ('2020-01-01 11:55:00', 'a');""", """ + uid | cnt | window_start | window_end | weight + ---------------------------------------------------------------- + a | 2 | 2020-01-01 11:50:00 | 2020-01-01 12:05:00 | 1"""); + // Before this step: + // ts | uid | + // -------+-----+ + // ============== waterline = 11:55 - 1:00 = 10:55 + // 11:50 | a | + // 11:55 | a | + // An unrelated key advances the waterline to 12:00; the 'a' session + // now straddles it: its rows are below, but a row on the waterline + // can still merge with it + ccs.step(""" + INSERT INTO events VALUES('2020-01-01 13:00:00', 'z');""", """ + uid | cnt | window_start | window_end | weight + ---------------------------------------------------------------- + z | 1 | 2020-01-01 13:00:00 | 2020-01-01 13:10:00 | 1"""); + // Before this step: + // ts | uid | + // -------+-----+ + // 11:50 | a | below the waterline, but the session is still live + // 11:55 | a | (a row at 12:00..12:05 can merge with it) + // ============== waterline = 13:00 - 1:00 = 12:00 + // 13:00 | z | + // On-time row 9 minutes after 11:55: merges; window_start must still + // be 11:50, which only survives GC if value retention kept it + ccs.step(""" + INSERT INTO events VALUES('2020-01-01 12:04:00', 'a');""", """ + uid | cnt | window_start | window_end | weight + ---------------------------------------------------------------- + a | 2 | 2020-01-01 11:50:00 | 2020-01-01 12:05:00 | -1 + a | 3 | 2020-01-01 11:50:00 | 2020-01-01 12:14:00 | 1"""); + } + @Test public void gcUpsertBoundary() { // The LATENESS column is not part of the primary key, so the input