[SQL] Implement SESSION windows #6782
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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. | ||
| * | ||
| * <p>{@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. | ||
| * | ||
| * <pre> | ||
| * LogicalTableFunctionScan(SESSION(DESCRIPTOR($ts), DESCRIPTOR($k), gap)) | ||
| * Input($0..$n-1) | ||
| * </pre> | ||
| * becomes ("sessionized" appears twice but is built once): | ||
| * <pre> | ||
| * 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 | ||
| * </pre> | ||
| * where "sessionized" numbers each row's session within its key: | ||
| * <pre> | ||
| * 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) | ||
| * </pre> | ||
| * 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. | ||
| * | ||
| * <p>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<DefaultOptRuleConfig<SessionRewriteRule>> | ||
| implements TransformationRule { | ||
| public SessionRewriteRule() { | ||
| super(CONFIG); | ||
| } | ||
|
|
||
| /** Column indexes of a DESCRIPTOR call, or null if 'node' is not one. */ | ||
| @Nullable | ||
| static List<Integer> descriptorColumns(RexNode node) { | ||
| if (!(node instanceof RexCall call) || call.getKind() != SqlKind.DESCRIPTOR) | ||
| return null; | ||
| List<Integer> 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<RexNode> operands = invocation.getOperands(); | ||
| if (operands.size() != 2 && operands.size() != 3) | ||
| return; | ||
| final List<Integer> 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<Integer> 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<Integer> 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<RexNode> 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<RexNode> 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<RexNode> replaceLast(List<RexNode> fields, int keep, RexNode last) { | ||
| List<RexNode> result = new ArrayList<>(fields.subList(0, keep)); | ||
| result.add(last); | ||
| return result; | ||
| } | ||
|
|
||
| public static final DefaultOptRuleConfig<SessionRewriteRule> CONFIG = | ||
| DefaultOptRuleConfig.<SessionRewriteRule>create() | ||
| .withOperandSupplier( | ||
| b -> b.operand(LogicalTableFunctionScan.class).anyInputs()); | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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: | ||
| * <pre> | ||
| * === at X === | ||
| * 748 JoinIndex <- [747:0, 731:0] | ||
| * 749 IntegrateTraceRetainNValues <- [748:0, 745:0] GC | ||
| * </pre> | ||
| * Insert it between passes to inspect the graph without generating images: | ||
| * <pre> | ||
| * this.add(new DumpTopology(compiler, "at X")); | ||
| * </pre> | ||
| */ | ||
| public class DumpTopology extends CircuitVisitor { | ||
| private final String label; | ||
| /** Restrict dump to operators selected by this predicate */ | ||
| private final Predicate<DBSPOperator> filter; | ||
|
|
||
| public DumpTopology(DBSPCompiler compiler, String label) { | ||
| this(compiler, label, op -> true); | ||
| } | ||
|
|
||
| public DumpTopology(DBSPCompiler compiler, String label, Predicate<DBSPOperator> 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 + ")"; | ||
| } | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
what is the inactivity gap
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
the distance between two consecutive events.
A session starts after a gap and is extended as long as there are events no farther than the specified distance from each other. Think web browser sessions.