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