From b1fa8a336c91e42b7388c6bc5ed8cf02990ed563 Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Wed, 22 Jul 2026 15:53:54 -0700 Subject: [PATCH 1/4] [SQL] Test for finite-state program with temporal filters Signed-off-by: Mihai Budiu --- .../sql/streaming/StreamingTests.java | 115 +++++++++++++++--- .../sql/tools/CompilerCircuitStream.java | 13 +- .../compiler/sql/tools/TableParser.java | 6 +- 3 files changed, 113 insertions(+), 21 deletions(-) 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 e8cf50511cf..8844fc88db5 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 @@ -51,12 +51,12 @@ CREATE TABLE t1( x INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE TABLE t2( y INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE VIEW v WITH ('emit_final' = 'ts') AS SELECT t1.ts @@ -259,7 +259,7 @@ CREATE TABLE t ( ) WITH ( 'append_only' = 'true' ); - + create view v1 AS SELECT TIMESTAMP_TRUNC(ts, DAY) as d, @@ -327,12 +327,12 @@ CREATE TABLE t1( x INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE TABLE t2( y INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE VIEW v WITH ('emit_final' = 'ts') AS SELECT @@ -397,7 +397,7 @@ CREATE TABLE T ( 'materialized' = 'true', 'append_only' = 'true' ); - + CREATE VIEW V WITH ('emit_final' = 'ts') AS SELECT * FROM T @@ -674,13 +674,13 @@ create table TRANSACTION ( id bigint NOT NULL, unix_time BIGINT LATENESS 100 ); - + create table FEEDBACK ( id bigint, status int, unix_time bigint NOT NULL LATENESS 100 ); - + CREATE VIEW TRANSACT AS SELECT feedback.*, transaction.* FROM @@ -2044,9 +2044,9 @@ CREATE TABLE data ( t0 TIMESTAMP NOT NULL LATENESS INTERVAL '2' HOURS, location INT NOT NULL ); - + CREATE LOCAL VIEW IT AS SELECT (t0 - TIMESTAMP '2020-01-01 00:00:00') HOURS AS t, location FROM data; - + CREATE VIEW V AS SELECT *, @@ -3056,7 +3056,7 @@ create table T ( y TIMESTAMP, site_id varchar ); - + create view V as select site_id from T where ( x >= NOW() + INTERVAL 30 DAYS @@ -3341,7 +3341,7 @@ CREATE TABLE T( lp VARCHAR, lsd TIMESTAMP ); - + create view V as SELECT s @@ -3389,7 +3389,7 @@ CREATE TABLE T( lp VARCHAR, lsd TIMESTAMP ); - + create view V as SELECT s @@ -3443,7 +3443,7 @@ CREATE TABLE T( lp VARCHAR, lsd TIMESTAMP ); - + create view V as SELECT s @@ -3501,7 +3501,7 @@ create table T ( properties variant, site_id varchar ); - + create view V as (select site_id from T where CAST(properties['x'] AS TIMESTAMP) >= NOW() + INTERVAL 30 DAYS) @@ -3519,16 +3519,97 @@ CREATE TABLE t1( x INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE TABLE t2( y INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE VIEW v WITH ('emit_final' = 'ts') AS SELECT t1.ts FROM t1 LEFT JOIN t2 on t1.ts = t2.ts;"""; this.getCCS(sql); } + + @Test + public void changeLog() { + // TLOG is a log of insertions and deletions applied to a table with + // primary key t_key; view T reconstructs the current table contents + // from the log entries of the last 25 hours: the latest entry per key + // wins, and a key whose latest entry is a deletion is absent. The op + // filter must sit outside the TOP-1, otherwise a deletion would + // resurrect the previous insertion. The temporal filter makes rows + // age out of T once their latest entry falls behind the window. + String sql = """ + CREATE TABLE TLOG ( + t_key INT NOT NULL, + payload VARCHAR, + op VARCHAR NOT NULL, + ts TIMESTAMP NOT NULL + ) WITH ('append_only' = 'true'); + + CREATE LOCAL VIEW RECENT AS + SELECT * FROM TLOG + WHERE ts >= NOW() - INTERVAL 25 HOURS AND ts <= NOW(); + + CREATE VIEW T AS + SELECT t_key, payload + FROM ( + SELECT t_key, payload, op, + ROW_NUMBER() OVER (PARTITION BY t_key ORDER BY ts DESC) AS rn + FROM RECENT + ) latest + WHERE rn = 1 AND op = 'insert';"""; + var ccs = this.getCCS(sql).withStringTrim(); + ccs.step(""" + INSERT INTO NOW VALUES('2020-01-01 01:00:00'); + INSERT INTO TLOG VALUES(1, 'aaa', 'insert', '2020-01-01 00:00:00'); + INSERT INTO TLOG VALUES(2, 'bbb', 'insert', '2020-01-01 00:10:00');""", """ + t_key | payload | weight + -------------------------- + 1 | aaa | 1 + 2 | bbb | 1"""); + // An update is a newer insertion for an existing key + ccs.step(""" + INSERT INTO NOW VALUES('2020-01-01 01:10:00'); + INSERT INTO TLOG VALUES(1, 'ccc', 'insert', '2020-01-01 00:20:00');""", """ + t_key | payload | weight + -------------------------- + 1 | aaa | -1 + 1 | ccc | 1"""); + // The latest entry for key 2 is a deletion, so the key disappears + ccs.step(""" + INSERT INTO NOW VALUES('2020-01-01 01:20:00'); + INSERT INTO TLOG VALUES(2, NULL, 'delete', '2020-01-01 00:30:00');""", """ + t_key | payload | weight + -------------------------- + 2 | bbb | -1"""); + // Out-of-order entries older than the latest entry for their key + // leave the view unchanged + ccs.step(""" + INSERT INTO NOW VALUES('2020-01-01 01:30:00'); + INSERT INTO TLOG VALUES(1, 'xxx', 'insert', '2020-01-01 00:15:00'); + INSERT INTO TLOG VALUES(1, NULL, 'delete', '2020-01-01 00:18:00');""", """ + t_key | payload | weight + --------------------------"""); + // A key deleted earlier reappears with a newer insertion + ccs.step(""" + INSERT INTO NOW VALUES('2020-01-01 01:40:00'); + INSERT INTO TLOG VALUES(2, 'ddd', 'insert', '2020-01-01 00:40:00');""", """ + t_key | payload | weight + -------------------------- + 2 | ddd | 1"""); + // 25 hours later all key-1 entries have left the window, so key 1 + // disappears; key 2 keeps its 00:40 insertion + ccs.step("INSERT INTO NOW VALUES('2020-01-02 01:30:00');", """ + t_key | payload | weight + -------------------------- + 1 | ccc | -1"""); + // ... and once the 00:40 insertion ages out too, T becomes empty + ccs.step("INSERT INTO NOW VALUES('2020-01-02 01:45:00');", """ + t_key | payload | weight + -------------------------- + 2 | ddd | -1"""); + } } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/CompilerCircuitStream.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/CompilerCircuitStream.java index 6adf7d9400f..e73f606256f 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/CompilerCircuitStream.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/CompilerCircuitStream.java @@ -22,6 +22,9 @@ public class CompilerCircuitStream extends CompilerCircuit { final InputOutputChangeStream stream; boolean compactAfterEachStep; + /** When true, string cells in expected tables are trimmed before validation, + * so cells can be padded for alignment; see {@link TableParser#parseValue}. */ + boolean trimStrings = false; public CompilerCircuitStream(DBSPCompiler compiler, BaseSQLTests test) { this(compiler, new InputOutputChangeStream(), test); @@ -55,6 +58,14 @@ public CompilerCircuitStream( test.addFailingRustTestCase(failureMessage, this); } + /** Trim string cells in expected tables before validating. + * Without this a string cell must end flush against the next column + * separator, since trailing spaces are part of the value. */ + public CompilerCircuitStream withStringTrim() { + this.trimStrings = true; + return this; + } + /** Compiles a SQL script composed of INSERT statements. * into a Change. */ public Change toChange(String script) { @@ -79,7 +90,7 @@ public Change toChange(String script) { public void step(String script, String expected) { Change input = this.toChange(script); DBSPType outputType = this.circuit.getSingleOutputType(); - Change output = TableParser.parseChangeTable(expected, outputType); + Change output = TableParser.parseChangeTable(expected, outputType, this.trimStrings); this.stream.addPair(input, output); if (this.compactAfterEachStep) this.blockForCompaction(); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/TableParser.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/TableParser.java index 16887a06da7..77e79b0adba 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/TableParser.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/tools/TableParser.java @@ -490,12 +490,12 @@ public static DBSPTupleExpression parseRow(String line, DBSPTypeTupleBase rowTyp /** Parse a change table. A change table is like a table, but has an extra * integer column that contains the weights. */ - public static Change parseChangeTable(String table, DBSPType outputType) { + public static Change parseChangeTable(String table, DBSPType outputType, boolean trimTrailingSpaces) { List extraFields = Linq.list(outputType.to(DBSPTypeZSet.class).elementType.to(DBSPTypeTuple.class).tupFields); extraFields.add(new DBSPTypeInteger(CalciteObject.EMPTY, 64, true, false)); DBSPType extraOutputType = new DBSPTypeTuple(extraFields); - Change change = parseTable(table, new DBSPTypeZSet(extraOutputType), -1, false); + Change change = parseTable(table, new DBSPTypeZSet(extraOutputType), -1, trimTrailingSpaces); TableData[] extracted = Linq.map(change.sets, SqlIoTest::extractWeight, TableData.class); return new Change(extracted); } @@ -598,6 +598,6 @@ public static Change fromResultSet(ResultSet data, DBSPType outputType) throws S builder.append("\t1"); } - return parseChangeTable(builder.toString(), outputType); + return parseChangeTable(builder.toString(), outputType, false); } } From b457268ba272704ba6eef85851cbcfae9d7df290 Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Wed, 22 Jul 2026 16:24:18 -0700 Subject: [PATCH 2/4] [SQL] Reformat test output Signed-off-by: Mihai Budiu --- .../sqlCompiler/compiler/sql/WindowTests.java | 94 +++++++------- .../simple/IncrementalRegression2Tests.java | 12 +- .../compiler/sql/simple/InternTests.java | 34 ++--- .../compiler/sql/simple/Regression1Tests.java | 44 +++---- .../compiler/sql/simple/Regression2Tests.java | 63 +++++----- .../compiler/sql/simple/VariantTests.java | 6 +- .../sql/streaming/StreamingTests.java | 119 +++++++++--------- .../sql/suites/nexmark/NexmarkTest.java | 16 +-- 8 files changed, 199 insertions(+), 189 deletions(-) diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/WindowTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/WindowTests.java index cd6d2019e3a..b82df5b27a4 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/WindowTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/WindowTests.java @@ -330,30 +330,30 @@ public void testRankGroup() { var ccs = this.getCCS(""" CREATE TABLE t1 (grp VARCHAR, x INT); CREATE VIEW V AS SELECT grp, x, RANK() OVER (PARTITION BY grp ORDER BY x) AS r - FROM t1;"""); + FROM t1;""").withStringTrim(); ccs.stepWeightOne("INSERT INTO t1 VALUES ('A', 1), ('A', 1), ('A', 2), ('B', 5), ('B', 5), ('B', 5)", """ grp | x | rank ---------------- - A| 1 | 1 - A| 1 | 1 - A| 2 | 3 - B| 5 | 1 - B| 5 | 1 - B| 5 | 1"""); + A | 1 | 1 + A | 1 | 1 + A | 2 | 3 + B | 5 | 1 + B | 5 | 1 + B | 5 | 1"""); ccs = this.getCCS(""" CREATE TABLE t1 (grp VARCHAR, x INT); CREATE VIEW V AS SELECT grp, x, DENSE_RANK() OVER (PARTITION BY grp ORDER BY x) AS r - FROM t1;"""); + FROM t1;""").withStringTrim(); ccs.stepWeightOne("INSERT INTO t1 VALUES ('A', 1), ('A', 1), ('A', 2), ('B', 5), ('B', 5), ('B', 5)", """ grp | x | rank ---------------- - A| 1 | 1 - A| 1 | 1 - A| 2 | 2 - B| 5 | 1 - B| 5 | 1 - B| 5 | 1"""); + A | 1 | 1 + A | 1 | 1 + A | 2 | 2 + B | 5 | 1 + B | 5 | 1 + B | 5 | 1"""); } @Test @@ -473,7 +473,7 @@ CREATE TABLE t_multi ( RANK() OVER (PARTITION BY grp ORDER BY score DESC) AS r_rank, DENSE_RANK() OVER (PARTITION BY grp ORDER BY score DESC) AS r_dense FROM t_multi - ORDER BY grp, score DESC, ts;"""); + ORDER BY grp, score DESC, ts;""").withStringTrim(); ccs.stepWeightOne(""" INSERT INTO t_multi VALUES ('A', 10, 1), @@ -489,16 +489,16 @@ CREATE TABLE t_multi ( ('B', 9, 5);""", """ grp | score | ts | rank | dense --------------------------------- - A| 30 | 5 | 1 | 1 - A| 20 | 3 | 2 | 2 - A| 20 | 4 | 2 | 2 - A| 10 | 1 | 4 | 3 - A| 10 | 2 | 4 | 3 - B| 9 | 4 | 1 | 1 - B| 9 | 5 | 1 | 1 - B| 7 | 3 | 3 | 2 - B| 5 | 1 | 4 | 3 - B| 5 | 2 | 4 | 3"""); + A | 30 | 5 | 1 | 1 + A | 20 | 3 | 2 | 2 + A | 20 | 4 | 2 | 2 + A | 10 | 1 | 4 | 3 + A | 10 | 2 | 4 | 3 + B | 9 | 4 | 1 | 1 + B | 9 | 5 | 1 | 1 + B | 7 | 3 | 3 | 2 + B | 5 | 1 | 4 | 3 + B | 5 | 2 | 4 | 3"""); } @Test @@ -522,7 +522,7 @@ WITH ranked AS ( SELECT * FROM ranked WHERE rnk <= 3 - ORDER BY rnk, score DESC, ts;"""); + ORDER BY rnk, score DESC, ts;""").withStringTrim(); ccs.stepWeightOne(""" INSERT INTO t_multi VALUES ('A', 10, 1), @@ -538,9 +538,9 @@ WITH ranked AS ( ('B', 9, 5);""", """ grp | score | ts | rank -------------------------- - A| 30 | 5 | 1 - A| 20 | 3 | 2 - A| 20 | 4 | 2"""); + A | 30 | 5 | 1 + A | 20 | 3 | 2 + A | 20 | 4 | 2"""); } @Test @@ -562,7 +562,7 @@ CREATE VIEW V AS WITH ranked AS ( RANK() OVER (ORDER BY ts DESC) AS rnk_recent FROM t ) - SELECT * FROM ranked;"""); + SELECT * FROM ranked;""").withStringTrim(); // no filtering first ccs.stepWeightOne(""" INSERT INTO t VALUES @@ -576,13 +576,13 @@ CREATE VIEW V AS WITH ranked AS ( """, """ id | score | ts | rnk_score | rnk_recent ------------------------------------------- - g| 100 | 60 | 1 | 1 - a| 100 | 10 | 1 | 7 - c| 95 | 30 | 3 | 5 - b| 95 | 20 | 3 | 6 - d| 80 | 40 | 5 | 4 - e| 70 | 50 | 6 | 3 - f| 60 | 60 | 7 | 1"""); + g | 100 | 60 | 1 | 1 + a | 100 | 10 | 1 | 7 + c | 95 | 30 | 3 | 5 + b | 95 | 20 | 3 | 6 + d | 80 | 40 | 5 | 4 + e | 70 | 50 | 6 | 3 + f | 60 | 60 | 7 | 1"""); ccs = this.getCCS(""" CREATE TABLE T ( @@ -601,7 +601,7 @@ CREATE VIEW V AS WITH ranked AS ( FROM t ) SELECT * FROM ranked - WHERE (rnk_score <= 3) AND (rnk_recent <= 2);"""); + WHERE (rnk_score <= 3) AND (rnk_recent <= 2);""").withStringTrim(); ccs.stepWeightOne(""" INSERT INTO t VALUES ('a', 100, 10), @@ -614,7 +614,7 @@ CREATE VIEW V AS WITH ranked AS ( """, """ id | score | ts | rnk_score | rnk_recent ------------------------------------------- - g| 100 | 60 | 1 | 1"""); + g | 100 | 60 | 1 | 1"""); } @Test @@ -636,7 +636,7 @@ CREATE VIEW V AS WITH ranked AS ( RANK() OVER (PARTITION BY score ORDER BY ts DESC, id) AS rnk_recent FROM t ) - SELECT * FROM ranked;"""); + SELECT * FROM ranked;""").withStringTrim(); // no filtering first ccs.stepWeightOne(""" INSERT INTO t VALUES @@ -650,13 +650,13 @@ CREATE VIEW V AS WITH ranked AS ( """, """ id | score | ts | rnk_score | rnk_recent ------------------------------------------- - g| 100 | 60 | 1 | 1 - a| 100 | 10 | 1 | 2 - c| 95 | 30 | 1 | 1 - b| 95 | 20 | 1 | 2 - d| 80 | 40 | 1 | 1 - e| 70 | 50 | 1 | 1 - f| 60 | 60 | 1 | 1"""); + g | 100 | 60 | 1 | 1 + a | 100 | 10 | 1 | 2 + c | 95 | 30 | 1 | 1 + b | 95 | 20 | 1 | 2 + d | 80 | 40 | 1 | 1 + e | 70 | 50 | 1 | 1 + f | 60 | 60 | 1 | 1"""); } @Test diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/IncrementalRegression2Tests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/IncrementalRegression2Tests.java index a51f13443ad..dcff374070c 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/IncrementalRegression2Tests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/IncrementalRegression2Tests.java @@ -287,18 +287,18 @@ CREATE TABLE customers ( c1.name FROM customers AS c1 JOIN customers AS c2 - ON c1.customer_id = c2.customer_id;"""); + ON c1.customer_id = c2.customer_id;""").withStringTrim(); // Validated on Postgres ccs.step(""" INSERT INTO customers (customer_id, name, first) VALUES (1, 'Johnson', 'Alice'), (2, 'Smith', 'Bob'), (3, 'White', 'Carol');""", """ - first | name1 | name | weight - --------------------------------------- - Alice| Johnson| Johnson| 1 - Bob| Smith| Smith| 1 - Carol| White| White| 1"""); + first | name1 | name | weight + ------------------------------------ + Alice | Johnson | Johnson | 1 + Bob | Smith | Smith | 1 + Carol | White | White | 1"""); ccs.visit(new CircuitVisitor(ccs.compiler) { int mapIndexCount = 0; diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/InternTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/InternTests.java index cef32b51b48..7de1fd312d9 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/InternTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/InternTests.java @@ -43,20 +43,21 @@ public void testSort() { public void testInterning() { var ccs = this.getCCS(""" CREATE TABLE T(x INT, s VARCHAR INTERNED, u VARCHAR); - CREATE VIEW V AS SELECT MAX(u), SUM(x), s FROM T GROUP BY s;"""); + CREATE VIEW V AS SELECT MAX(u), SUM(x), s FROM T GROUP BY s;""") + .withStringTrim(); ccs.stepWeightOne("INSERT INTO T VALUES(0, 'a', 'b');", """ max | sum | s --------------- - b| 0 | a"""); + b | 0 | a"""); ccs.stepWeightOne("INSERT INTO T VALUES(1, 'd', 'c');", """ max | sum | s --------------- - c| 1 | d"""); + c | 1 | d"""); ccs.step("INSERT INTO T VALUES(2, 'a', 'c');", """ - max| sum | s| weight - ---------------------- - b| 0 | a| -1 - c| 2 | a| 1"""); + max | sum | s | weight + ------------------------ + b | 0 | a | -1 + c | 2 | a | 1"""); ccs.stepWeightOne("INSERT INTO T VALUES(NULL, NULL, NULL);", """ max | sum | s --------------- @@ -104,12 +105,12 @@ public void testUnnest() { public void testTwoColumns() { var ccs = this.getCCS(""" CREATE TABLE T(x VARCHAR NOT NULL INTERNED, s VARCHAR INTERNED); - CREATE VIEW V AS SELECT * FROM T;"""); + CREATE VIEW V AS SELECT * FROM T;""").withStringTrim(); ccs.stepWeightOne("INSERT INTO T VALUES('x', 'y'), ('z', NULL);", """ - x| y - ------ - x| y - z|NULL"""); + x | y + ------- + x | y + z |NULL"""); } @Test @@ -118,16 +119,17 @@ public void testLeftJoin() { var ccs = this.getCCS(""" CREATE TABLE T(x VARCHAR NOT NULL INTERNED, y VARCHAR NOT NULL INTERNED); CREATE TABLE S(z VARCHAR INTERNED, w VARCHAR INTERNED, a INT); - CREATE VIEW V AS SELECT T.x, S.a FROM T LEFT JOIN S ON T.x = S.z AND T.y = S.w;"""); + CREATE VIEW V AS SELECT T.x, S.a FROM T LEFT JOIN S ON T.x = S.z AND T.y = S.w;""") + .withStringTrim(); ccs.stepWeightOne(""" INSERT INTO T VALUES('a', 'b'), ('a', 'c'); INSERT INTO S VALUES('a', 'b', 1), ('a', 'd', 2), ('b', 'c', 3), ('a', 'b', 4); """, """ x | a ------- - a| 1 - a| 4 - a|NULL"""); + a | 1 + a | 4 + a |NULL"""); } @Test diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression1Tests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression1Tests.java index e7b13e3e739..d7c3560457f 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression1Tests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression1Tests.java @@ -532,11 +532,12 @@ public void issue4264() { public void castBinaryToString() { var ccs = this.getCCS(""" CREATE TABLE T(x BINARY(2)); - CREATE VIEW V AS SELECT CAST(x AS VARCHAR), CAST(x'AB01' AS VARCHAR) FROM T;"""); + CREATE VIEW V AS SELECT CAST(x AS VARCHAR), CAST(x'AB01' AS VARCHAR) FROM T;""") + .withStringTrim(); ccs.stepWeightOne("INSERT INTO T VALUES(x'AB01')", """ - x| y - ---------- - ab01| ab01"""); + x | y + ----------- + ab01 | ab01"""); } @Test @@ -1203,16 +1204,17 @@ WITH ref_profile AS ( SELECT cast(contacts as MAP) contacts FROM user_props ) SELECT key, to_json(contact) - FROM ref_profile profile_0, UNNEST(profile_0.contacts) AS t(key, contact)"""); + FROM ref_profile profile_0, UNNEST(profile_0.contacts) AS t(key, contact)""") + .withStringTrim(); ccs.stepWeightOne(""" INSERT INTO j VALUES('{ "a": "1", "b": 2, "c": [1, 2, 3], "d": null, "e": { "f": 1 } }');""", """ key | contact --------------- - a| "1" - b| 2 - c| [1,2,3] - d| null - e| {"f":1}"""); + a | "1" + b | 2 + c | [1,2,3] + d | null + e | {"f":1}"""); } @Test @@ -1529,13 +1531,13 @@ public void issue5293() { CREATE VIEW V AS WITH FT as (select 'a' as e union all select 'bc') SELECT x, x in (SELECT e from FT) - FROM T;"""); + FROM T;""").withStringTrim(); ccs.stepWeightOne("INSERT INTO T VALUES('a'), ('b'), ('ab');", """ - x | in - ------- - a| true - b|false - ab|false"""); + x | in + --------- + a | true + b | false + ab | false"""); } @Test @@ -1609,12 +1611,12 @@ public void issue5345() { VALUES ('a', 1, ARRAY['by'], true), ('b', 1, ARRAY(), false) - ) AS t (f1, f2, f3, f4);"""); + ) AS t (f1, f2, f3, f4);""").withStringTrim(); ccs.stepWeightOne("", """ - f1 | f2 | f3 | f4 - -------------------- - a| 1 | { by} | true - b| 1 | {} | false"""); + f1 | f2 | f3 | f4 + ---------------------- + a | 1 | { by} | true + b | 1 | {} | false"""); } @Test diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java index 5bed1e0ff41..e07982bf64b 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java @@ -1005,7 +1005,8 @@ public void endVisit() { public void issue5927() { var ccs = this.getCCS(""" CREATE TABLE T(g VARCHAR, x INT); - CREATE VIEW V AS SELECT g, MAX(CASE WHEN x > 2 THEN 1 ELSE 0 END) FROM T GROUP BY g;"""); + CREATE VIEW V AS SELECT g, MAX(CASE WHEN x > 2 THEN 1 ELSE 0 END) FROM T GROUP BY g;""") + .withStringTrim(); // Validated on Postgres ccs.stepWeightOne("", """ g | max @@ -1013,11 +1014,11 @@ public void issue5927() { ccs.stepWeightOne("INSERT INTO T VALUES ('a', 1), ('b', 3), ('c', 3), ('c', 5), ('d', 1), ('d', 5), ('e', NULL);", """ g | max --------- - a| 0 - b| 1 - c| 1 - d| 1 - e| 0"""); + a | 0 + b | 1 + c | 1 + d | 1 + e | 0"""); ccs.visit(this.findLinear(ccs.compiler)); } @@ -1047,7 +1048,8 @@ public void testXxHash() { public void issue5927a() { var ccs = this.getCCS(""" CREATE TABLE T(g VARCHAR, x INT); - CREATE VIEW V AS SELECT g, MAX(CASE WHEN x = 1 THEN 0 ELSE 1 END) FROM T GROUP BY g;"""); + CREATE VIEW V AS SELECT g, MAX(CASE WHEN x = 1 THEN 0 ELSE 1 END) FROM T GROUP BY g;""") + .withStringTrim(); // Validated on Postgres ccs.stepWeightOne("", """ g | max @@ -1066,10 +1068,10 @@ public void issue5927a() { """, """ g | max --------- - a| 1 - b| 0 - c| 1 - d| 1"""); + a | 1 + b | 0 + c | 1 + d | 1"""); ccs.visit(this.findLinear(ccs.compiler)); } @@ -1078,7 +1080,8 @@ public void issue6590() { // MAX(CASE WHEN cond THEN 1 ELSE NULL END) — ELSE NULL variant var ccs = this.getCCS(""" CREATE TABLE T(g VARCHAR, x INT); - CREATE VIEW V AS SELECT g, MAX(CASE WHEN x > 2 THEN 1 ELSE NULL END) FROM T GROUP BY g;"""); + CREATE VIEW V AS SELECT g, MAX(CASE WHEN x > 2 THEN 1 ELSE NULL END) FROM T GROUP BY g;""") + .withStringTrim(); // Validated on Postgres: NULL when no row satisfies cond, 1 otherwise. ccs.stepWeightOne("", """ g | max @@ -1089,11 +1092,11 @@ public void issue6590() { """, """ g | max --------- - a| NULL - b| 1 - c| 1 - d| 1 - e| NULL"""); + a | NULL + b | 1 + c | 1 + d | 1 + e | NULL"""); ccs.visit(this.findLinear(ccs.compiler)); } @@ -1102,7 +1105,8 @@ public void issue6590a() { // MAX(CASE WHEN cond THEN 1 END) — no ELSE clause (equivalent to ELSE NULL) var ccs = this.getCCS(""" CREATE TABLE T(g VARCHAR, x INT); - CREATE VIEW V AS SELECT g, MAX(CASE WHEN x > 2 THEN 1 END) FROM T GROUP BY g;"""); + CREATE VIEW V AS SELECT g, MAX(CASE WHEN x > 2 THEN 1 END) FROM T GROUP BY g;""") + .withStringTrim(); // Validated on Postgres: NULL when no row satisfies cond, 1 otherwise. ccs.stepWeightOne("", """ g | max @@ -1114,11 +1118,11 @@ public void issue6590a() { """, """ g | max --------- - a| NULL - b| 1 - c| 1 - d| 1 - e| NULL"""); + a | NULL + b | 1 + c | 1 + d | 1 + e | NULL"""); ccs.visit(this.findLinear(ccs.compiler)); } @@ -1128,7 +1132,8 @@ public void issue6590b() { // Exercises the post-project index path for untransformed aggregate calls. var ccs = this.getCCS(""" CREATE TABLE T(g VARCHAR, x INT); - CREATE VIEW V AS SELECT g, MAX(CASE WHEN x > 2 THEN 1 ELSE 0 END), SUM(x) FROM T GROUP BY g;"""); + CREATE VIEW V AS SELECT g, MAX(CASE WHEN x > 2 THEN 1 ELSE 0 END), SUM(x) FROM T GROUP BY g;""") + .withStringTrim(); // Validated on Postgres ccs.stepWeightOne("", """ g | max | sum @@ -1140,11 +1145,11 @@ public void issue6590b() { """, """ g | max | sum -------------- - a| 0| 1 - b| 1| 3 - c| 1| 8 - d| 1| 6 - e| 0| NULL"""); + a | 0 | 1 + b | 1 | 3 + c | 1 | 8 + d | 1 | 6 + e | 0 | NULL"""); ccs.visit(this.findLinear(ccs.compiler)); } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/VariantTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/VariantTests.java index 41950046bc6..b1e6c23b4d0 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/VariantTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/VariantTests.java @@ -944,13 +944,13 @@ public void testSparkInline() { CREATE LOCAL VIEW DECODE(rec) AS SELECT jsonstring_as_t_steps(encoded) as steps FROM DATA; -- extract and flatten the arrays from the DECODE view CREATE VIEW OUT(name, "uuid") AS SELECT x.name, x."uuid" FROM DECODE, UNNEST(DECODE.rec.steps) AS x; - """); + """).withStringTrim(); ccs.stepWeightOne("INSERT INTO DATA VALUES (" + data + ")", """ name | uuid ------------- - blah| uuid0 - boo|NULL"""); + blah | uuid0 + boo |NULL"""); } @Test 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 8844fc88db5..4557753161e 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 @@ -3062,20 +3062,20 @@ create table T ( where ( x >= NOW() + INTERVAL 30 DAYS OR y >= NOW() - INTERVAL 30 DAYS);"""; - var ccs = this.getCCS(sql); + var ccs = this.getCCS(sql).withStringTrim(); ccs.step(""" INSERT INTO NOW VALUES('2019-01-01 00:00:00'); INSERT INTO T VALUES('2020-01-11 00:00:00', '2020-01-11 00:00:00', 'z');""", """ site_id | weight ------------------ - z|1"""); + z | 1"""); ccs.step("INSERT INTO NOW VALUES('2020-01-01 00:00:00')", """ site_id | weight ------------------"""); ccs.step("INSERT INTO NOW VALUES('2020-03-01 00:00:00')", """ site_id | weight ------------------ - z| -1"""); + z | -1"""); } @Test @@ -3089,7 +3089,7 @@ public void issue6655() { COUNT(CASE WHEN tt >= NOW() - INTERVAL 1 DAY THEN 1 END) AS c, COUNT(*) AS total FROM T GROUP BY k;"""; - var ccs = this.getCCS(sql); + var ccs = this.getCCS(sql).withStringTrim(); CircuitVisitor visitor = new CircuitVisitor(ccs.compiler) { int window = 0; int aggregate = 0; @@ -3120,14 +3120,14 @@ public void endVisit() { INSERT INTO T VALUES('b', '2019-12-30 00:00:00');""", """ k | s | c | total | weight -------------------------------- - a| 1 | 1 | 1 | 1 - b|NULL | 0 | 1 | 1"""); + a | 1 | 1 | 1 | 1 + b |NULL | 0 | 1 | 1"""); // Two days later a's row leaves the window; b is unchanged ccs.step("INSERT INTO NOW VALUES('2020-01-03 00:00:00')", """ k | s | c | total | weight -------------------------------- - a| 1 | 1 | 1 | -1 - a|NULL | 0 | 1 | 1"""); + a | 1 | 1 | 1 | -1 + a |NULL | 0 | 1 | 1"""); } @Test @@ -3158,26 +3158,27 @@ public void endVisit() { } }; ccs.visit(visitor); + ccs.withStringTrim(); ccs.step(""" INSERT INTO NOW VALUES('2020-01-10 00:00:00'); INSERT INTO T VALUES('a', '2020-01-10 00:00:00'); INSERT INTO T VALUES('a', '2020-01-05 00:00:00');""", """ k | d | w | total | weight ---------------------------------- - a| 1 | 2 | 2 | 1"""); + a | 1 | 2 | 2 | 1"""); // Two days later the newest row leaves the 1-day window; // both rows are still inside the 7-day window ccs.step("INSERT INTO NOW VALUES('2020-01-12 00:00:00')", """ k | d | w | total | weight ---------------------------------- - a| 1 | 2 | 2 | -1 - a|NULL | 2 | 2 | 1"""); + a | 1 | 2 | 2 | -1 + a |NULL | 2 | 2 | 1"""); // Ten days after the first step both rows have left both windows ccs.step("INSERT INTO NOW VALUES('2020-01-20 00:00:00')", """ k | d | w | total | weight ---------------------------------- - a|NULL | 2 | 2 | -1 - a|NULL |NULL | 2 | 1"""); + a |NULL | 2 | 2 | -1 + a |NULL |NULL | 2 | 1"""); } @Test @@ -3292,18 +3293,18 @@ public void issue6655d() { CREATE VIEW V AS SELECT k, ARRAY_AGG(x) FILTER (WHERE tt >= NOW() - INTERVAL 1 DAY) AS agg FROM T GROUP BY k;"""; - var ccs = this.getCCS(sql); + var ccs = this.getCCS(sql).withStringTrim(); ccs.step(""" INSERT INTO NOW VALUES('2020-01-01 00:00:00'); INSERT INTO T VALUES('a', '2020-01-01 00:00:00', 10);""", """ k | agg | weight ------------------- - a|{ 10 } | 1"""); + a |{ 10 } | 1"""); ccs.step("INSERT INTO NOW VALUES('2020-01-03 00:00:00')", """ k | agg | weight ------------------- - a|{ 10 } | -1 - a|{} | 1"""); + a |{ 10 } | -1 + a |{} | 1"""); } static final String issue4909data = """ @@ -3360,7 +3361,7 @@ CREATE TABLE T( lsd >= NOW() - INTERVAL 30 DAYS OR op IS NOT NULL ); - """); + """).withStringTrim(); ccs.visit(new CircuitVisitor(ccs.compiler) { @Override public void postorder(DBSPJoinBaseOperator join) { @@ -3369,12 +3370,12 @@ public void postorder(DBSPJoinBaseOperator join) { }); // Validated using Postgres on the right date ccs.step(issue4909data, """ - s | weight - ---------------- - charlie| 1 - india| 1 - kilo| 1 - mike| 1"""); + s | weight + ------------------- + charlie | 1 + india | 1 + kilo | 1 + mike | 1"""); } @Test @@ -3408,7 +3409,7 @@ CREATE TABLE T( lsd >= NOW() - INTERVAL 30 DAYS OR op IS NOT NULL ); - """); + """).withStringTrim(); ccs.visit(new CircuitVisitor(ccs.compiler) { @Override public void postorder(DBSPJoinBaseOperator join) { @@ -3417,18 +3418,18 @@ public void postorder(DBSPJoinBaseOperator join) { }); // Validated using Postgres on the right date ccs.step(issue4909data, """ - s | weight - ---------------- - alpha| 1 - bravo| 1 - charlie| 1 - delta| 1 - echo| 1 - foxtrot| 1 - india| 1 - kilo| 1 - mike| 1 - sierra| 1"""); + s | weight + ------------------- + alpha | 1 + bravo | 1 + charlie | 1 + delta | 1 + echo | 1 + foxtrot | 1 + india | 1 + kilo | 1 + mike | 1 + sierra | 1"""); } @Test @@ -3462,7 +3463,7 @@ CREATE TABLE T( lsd >= NOW() - INTERVAL 30 DAYS AND op IS NOT NULL ); - """); + """).withStringTrim(); ccs.visit(new CircuitVisitor(ccs.compiler) { @Override public void postorder(DBSPJoinBaseOperator join) { @@ -3471,27 +3472,27 @@ public void postorder(DBSPJoinBaseOperator join) { }); // Validated using Postgres on the right date ccs.step(issue4909data, """ - s | weight - ---------------- - alpha| 1 - bravo| 1 - charlie| 1 - delta| 1 - echo| 1 - foxtrot| 1 - golf| 1 - hotel| 1 - india| 1 - juliet| 1 - kilo| 1 - lima| 1 - mike| 1 - november| 1 - oscar| 1 - papa| 1 - quebec| 1 - romeo| 1 - sierra| 1"""); + s | weight + ------------------- + alpha | 1 + bravo | 1 + charlie | 1 + delta | 1 + echo | 1 + foxtrot | 1 + golf | 1 + hotel | 1 + india | 1 + juliet | 1 + kilo | 1 + lima | 1 + mike | 1 + november | 1 + oscar | 1 + papa | 1 + quebec | 1 + romeo | 1 + sierra | 1"""); } @Test diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/nexmark/NexmarkTest.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/nexmark/NexmarkTest.java index 0d11b49d588..f6cc8f12546 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/nexmark/NexmarkTest.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/suites/nexmark/NexmarkTest.java @@ -586,7 +586,7 @@ CompilerCircuitStream createTest(int query, String... scriptsAndTables) { //noinspection ConstantValue if (debug) previous = Logger.INSTANCE.setLoggingLevel(module, 1); - CompilerCircuitStream ccs = this.getCCS(compiler); + CompilerCircuitStream ccs = this.getCCS(compiler).withStringTrim(); //noinspection ConstantValue if (debug) Logger.INSTANCE.setLoggingLevel(module, previous); @@ -672,19 +672,19 @@ public void q3Test() { INSERT INTO Auction VALUES(452, 'item-name', 'description', 5, 10, '2020-01-01 01:00:00', '2020-01-02 00:00:00', 3, 10, ''); """, """ - name | city | state | id | weight - ----------------------------------------- - CA Seller| Phoenix| CA| 999 | 1 - ID Seller| Phoenix| ID| 452 | 1""", + name | city | state | id | weight + -------------------------------------------- + CA Seller | Phoenix | CA | 999 | 1 + ID Seller | Phoenix | ID | 452 | 1""", """ INSERT INTO Person VALUES(4, 'OR Seller', 'AAABBB@example.com', '1111 2222 3333 4444', 'Phoenix', 'PR', '2020-01-01 00:00:00', ''); INSERT INTO Auction VALUES(999, 'item-name', 'description', 5, 10, '2020-01-01 01:00:00', '2020-01-02 00:00:00', 4, 11, ''); INSERT INTO Person VALUES(5, 'OR Seller', 'AAABBB@example.com', '1111 2222 3333 4444', 'Phoenix', 'OR', '2020-01-01 00:00:00', ''); INSERT INTO Auction VALUES(333, 'item-name', 'description', 5, 10, '2020-01-01 01:00:00', '2020-01-02 00:00:00', 5, 10, '');""", """ - name | city | state | id | weight - ------------------------------------------ - OR Seller| Phoenix| OR| 333 | 1""" + name | city | state | id | weight + -------------------------------------------- + OR Seller | Phoenix | OR | 333 | 1""" ); } From b330f51e56a5b78ac57f60f67b11da4d31da428e Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Wed, 22 Jul 2026 17:22:02 -0700 Subject: [PATCH 3/4] [SQL] Analysis for discovering non-GC-ed state Signed-off-by: Mihai Budiu --- .../DBSPInputMapWithWaterlineOperator.java | 2 +- .../circuit/operator/DBSPWindowOperator.java | 11 +- .../circuit/operator/IInputMapOperator.java | 2 +- .../compiler/backend/ToJsonOuterVisitor.java | 2 + .../backend/dot/ToDotNodesVisitor.java | 4 +- .../compiler/errors/SourceFileContents.java | 9 + .../compiler/errors/SourcePositionRanges.java | 2 +- .../frontend/calciteObject/CalciteObject.java | 3 - .../calciteObject/CalciteRelNode.java | 3 - .../frontend/calciteObject/RelAnd.java | 9 + .../visitors/outer/CircuitOptimizer.java | 1 + .../visitors/outer/CircuitRewriter.java | 2 +- .../visitors/outer/FindSourcePositions.java | 58 ++++ .../visitors/outer/FindUnboundedState.java | 283 ++++++++++++++++++ .../compiler/visitors/outer/Passes.java | 6 +- .../visitors/outer/ToJsonVisitor.java | 60 +--- .../outer/monotonicity/InsertLimiters.java | 8 +- .../visitors/outer/temporal/RewriteNow.java | 8 +- .../visitors/unusedFields/TrimWindows.java | 3 +- .../operator/DBSPWindowOperatorTests.java | 2 +- .../compiler/sql/simple/Regression1Tests.java | 6 +- .../compiler/sql/simple/Regression2Tests.java | 2 +- .../sql/streaming/StreamingTests.java | 48 +-- .../resources/metadataTests-generateDF.json | 1 - .../metadataTests-generateDFRecursive.json | 1 - 25 files changed, 430 insertions(+), 106 deletions(-) create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/FindSourcePositions.java create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/FindUnboundedState.java diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPInputMapWithWaterlineOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPInputMapWithWaterlineOperator.java index a162d9a83a0..a47cea6f999 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPInputMapWithWaterlineOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPInputMapWithWaterlineOperator.java @@ -30,7 +30,7 @@ * with records that missed the lateness. */ public class DBSPInputMapWithWaterlineOperator extends DBSPOperator - implements IMultiOutput, IInputMapOperator, IInputOperator { + implements IMultiOutput, IInputMapOperator, IInputOperator, IStateful { // Fields that belong normally to SourceTableOperators (which we don't derive from) public final ProgramIdentifier tableName; public final DBSPTypeStruct originalRowType; diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWindowOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWindowOperator.java index 8456f890fc1..cd2eb5f3700 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWindowOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWindowOperator.java @@ -27,9 +27,13 @@ public final class DBSPWindowOperator extends DBSPBinaryOperator implements IContainsIntegrator, IIncremental { public final boolean lowerInclusive; public final boolean upperInclusive; + /** True if the window's lower bound is -infinity. + * This property is not used for code generation, but it is difficult to infer later. + * The validity relies on the fact that no optimizations change window bounds once created. */ + public final boolean lowerUnbounded; public DBSPWindowOperator(CalciteRelNode node, boolean lowerInclusive, boolean upperInclusive, - OutputPort data, OutputPort control) { + boolean lowerUnbounded, OutputPort data, OutputPort control) { super(node, "window", null, data.outputType(), data.isMultiset(), data, control); // Check that the left input and output are indexed ZSets. DBSPTypeIndexedZSet indexedType = this.getOutputIndexedZSetType(); @@ -41,6 +45,7 @@ public DBSPWindowOperator(CalciteRelNode node, boolean lowerInclusive, boolean u ", but have type " + control.outputType()); this.lowerInclusive = lowerInclusive; this.upperInclusive = upperInclusive; + this.lowerUnbounded = lowerUnbounded; } @Override @@ -52,6 +57,7 @@ public DBSPSimpleOperator with( if (force || this.inputsDiffer(newInputs)) return new DBSPWindowOperator( this.getRelNode(), this.lowerInclusive, this.upperInclusive, + this.lowerUnbounded, newInputs.get(0), newInputs.get(1)).copyAnnotations(this); } return this; @@ -71,8 +77,9 @@ public static DBSPWindowOperator fromJson(JsonNode node, JsonDecoder decoder) { DBSPSimpleOperator.CommonInfo info = commonInfoFromJson(node, decoder); boolean lowerInclusive = Utilities.getBooleanProperty(node, "lowerInclusive"); boolean upperInclusive = Utilities.getBooleanProperty(node, "upperInclusive"); + boolean lowerUnbounded = Utilities.getBooleanProperty(node, "lowerUnbounded"); return new DBSPWindowOperator(CalciteEmptyRel.INSTANCE, - lowerInclusive, upperInclusive, info.getInput(0), info.getInput(1)) + lowerInclusive, upperInclusive, lowerUnbounded, info.getInput(0), info.getInput(1)) .addAnnotations(info.annotations(), DBSPWindowOperator.class); } } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/IInputMapOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/IInputMapOperator.java index 8d63a54f55b..f6b91c28eb5 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/IInputMapOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/IInputMapOperator.java @@ -16,7 +16,7 @@ import java.util.List; /** Interface for a source operator that has primary keys */ -public interface IInputMapOperator extends IInputOperator { +public interface IInputMapOperator extends IInputOperator, IStateful { TableMetadata getMetadata(); List getKeyFields(); DBSPTypeIndexedZSet getOutputIndexedZSetType(); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java index 364de81bfeb..d8fcea9e8a3 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/ToJsonOuterVisitor.java @@ -333,6 +333,8 @@ public VisitDecision preorder(DBSPWindowOperator operator) { this.stream.append(operator.lowerInclusive); this.property("upperInclusive"); this.stream.append(operator.upperInclusive); + this.property("lowerUnbounded"); + this.stream.append(operator.lowerUnbounded); return VisitDecision.CONTINUE; } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/dot/ToDotNodesVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/dot/ToDotNodesVisitor.java index 5a376659674..20a6ebaf46d 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/dot/ToDotNodesVisitor.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/dot/ToDotNodesVisitor.java @@ -24,8 +24,8 @@ import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRange; import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision; import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor; +import org.dbsp.sqlCompiler.compiler.visitors.outer.FindSourcePositions; import org.dbsp.sqlCompiler.compiler.visitors.outer.LowerCircuitVisitor; -import org.dbsp.sqlCompiler.compiler.visitors.outer.ToJsonVisitor; import org.dbsp.sqlCompiler.ir.IDBSPInnerNode; import org.dbsp.sqlCompiler.ir.IDBSPOuterNode; import org.dbsp.sqlCompiler.ir.expression.DBSPExpression; @@ -153,7 +153,7 @@ static String escapeString(String input) { String getPositions(IDBSPInnerNode node) { StringBuilder result = new StringBuilder(); - ToJsonVisitor.FindSourcePositions finder = new ToJsonVisitor.FindSourcePositions(this.compiler, true); + FindSourcePositions finder = new FindSourcePositions(this.compiler, true); finder.apply(node); for (SourcePositionRange r : finder.getPositions()) { result.append(r.toShortString()); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourceFileContents.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourceFileContents.java index 43220ae1f44..ec5b60eae73 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourceFileContents.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourceFileContents.java @@ -124,6 +124,15 @@ else if (i == endLine) return result.toString(); } + public String getFragments(SourcePositionRanges ranges) { + StringBuilder result = new StringBuilder(); + for (SourcePositionRange r: ranges.positions) { + result.append(this.getFragment(r, true)); + result.append("\n"); + } + return result.toString(); + } + public String getFragment(SqlNode node) { return this.getFragment( new SourcePositionRange(node.getParserPosition()), false); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourcePositionRanges.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourcePositionRanges.java index f8d8d98a1fe..b167e1d4f8e 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourcePositionRanges.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/errors/SourcePositionRanges.java @@ -8,7 +8,7 @@ /** A set of source positions */ public class SourcePositionRanges implements Iterable { - final List positions; + public final List positions; public SourcePositionRanges(Iterable positions) { List pos = Linq.list(positions); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteObject.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteObject.java index 3610c4c90ab..220dcafa7dd 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteObject.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteObject.java @@ -37,9 +37,6 @@ public String toString() { return ""; } - /** Format the object in a way that can displayed in an error message */ - public String getMessage() { return this.toString(); } - public String toInternalString() { return this.toString(); } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteRelNode.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteRelNode.java index bfad040f6fd..151f6d7e942 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteRelNode.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/CalciteRelNode.java @@ -70,9 +70,6 @@ public static String toSqlString(RelNode node) { } } - @Override - public String getMessage() { return ""; } - public abstract IIndentStream asJson(IIndentStream stream, Map idRemap); public abstract CalciteRelNode remove(RelNode node); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/RelAnd.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/RelAnd.java index 7ee925c9f61..dc4e35158c5 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/RelAnd.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteObject/RelAnd.java @@ -32,6 +32,15 @@ public RelAnd() { this.nodes = new HashSet<>(); } + @Override + public List getSourcePositions() { + List result = new ArrayList<>(); + for (LastRel lr: this.nodes) { + result.addAll(lr.getSourcePositions()); + } + return result; + } + @Override public IIndentStream asJson(IIndentStream stream, Map idRemap) { if (this.nodes.size() == 1) { diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java index 768ef268d7a..32948adb7cc 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitOptimizer.java @@ -180,6 +180,7 @@ void createOptimizer() { this.add(new MerkleOuter(compiler, true)); this.add(new MerkleOuter(compiler, false)); this.add(new TagRegions(compiler)); + this.add(new FindUnboundedState(compiler)); this.add(new CircuitStatistics(compiler)); } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitRewriter.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitRewriter.java index c73986c6e6c..c482b5c78e9 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitRewriter.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitRewriter.java @@ -690,7 +690,7 @@ public void postorder(DBSPWindowOperator operator) { DBSPSimpleOperator result = operator; if (Linq.different(sources, operator.inputs)) result = new DBSPWindowOperator(operator.getRelNode(), operator.lowerInclusive, operator.upperInclusive, - sources.get(0), sources.get(1)) + operator.lowerUnbounded, sources.get(0), sources.get(1)) .copyAnnotations(operator); this.map(operator, result); } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/FindSourcePositions.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/FindSourcePositions.java new file mode 100644 index 00000000000..7b3ebbdd388 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/FindSourcePositions.java @@ -0,0 +1,58 @@ +package org.dbsp.sqlCompiler.compiler.visitors.outer; + +import org.dbsp.sqlCompiler.circuit.operator.DBSPOperator; +import org.dbsp.sqlCompiler.compiler.DBSPCompiler; +import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRange; +import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRanges; +import org.dbsp.sqlCompiler.compiler.visitors.inner.InnerVisitor; +import org.dbsp.sqlCompiler.ir.DBSPParameter; +import org.dbsp.sqlCompiler.ir.IDBSPInnerNode; +import org.dbsp.sqlCompiler.ir.expression.DBSPExpression; + +import java.util.HashSet; +import java.util.Set; + +/** Visitor which extracts source position information from the various properties of an operator */ +public class FindSourcePositions extends InnerVisitor { + public final Set positions; + private final boolean reset; + + public FindSourcePositions(DBSPCompiler compiler, boolean reset) { + super(compiler); + this.positions = new HashSet<>(); + this.reset = reset; + } + + @Override + public void postorder(DBSPExpression expression) { + SourcePositionRange positionRange = expression.getNode().getPositionRange(); + if (positionRange.isValid()) + this.positions.add(positionRange); + } + + @Override + public void postorder(DBSPParameter parameter) { + SourcePositionRange positionRange = parameter.getNode().getPositionRange(); + if (positionRange.isValid()) + this.positions.add(positionRange); + } + + @Override + public void startVisit(IDBSPInnerNode node) { + super.startVisit(node); + if (this.reset) + this.positions.clear(); + } + + public SourcePositionRanges getPositions() { + return new SourcePositionRanges(this.positions); + } + + /** Find the source positions associated with the specified operator */ + public static SourcePositionRanges getPositions(DBSPCompiler compiler, DBSPOperator operator) { + FindSourcePositions positions = new FindSourcePositions(compiler, true); + operator.accept(positions); + positions.positions.addAll(operator.getSourcePositions()); + return positions.getPositions(); + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/FindUnboundedState.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/FindUnboundedState.java new file mode 100644 index 00000000000..345392359a8 --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/FindUnboundedState.java @@ -0,0 +1,283 @@ +package org.dbsp.sqlCompiler.compiler.visitors.outer; + +import org.dbsp.sqlCompiler.circuit.DBSPCircuit; +import org.dbsp.sqlCompiler.circuit.ICircuit; +import org.dbsp.sqlCompiler.circuit.OutputPort; +import org.dbsp.sqlCompiler.circuit.operator.DBSPAggregateLinearPostprocessRetainKeysOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPBinaryDistinctOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPBinaryOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPConstantOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPDelayOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPDifferentiateOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPDistinctOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPIndexedTopKOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPIntegrateOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPNestedOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPPartitionedRollingAggregateWithWaterlineOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPPositiveOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPRankOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPRowNumberOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPStreamDistinctOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPUpsertFeedbackOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPViewDeclarationOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPWaterlineOperator; +import org.dbsp.sqlCompiler.circuit.operator.DBSPWindowOperator; +import org.dbsp.sqlCompiler.circuit.operator.IGCOperator; +import org.dbsp.sqlCompiler.circuit.operator.IInputOperator; +import org.dbsp.sqlCompiler.circuit.operator.IJoin; +import org.dbsp.sqlCompiler.circuit.operator.ILinear; +import org.dbsp.sqlCompiler.circuit.operator.INonLinearAggregate; +import org.dbsp.sqlCompiler.circuit.operator.IStateful; +import org.dbsp.sqlCompiler.compiler.DBSPCompiler; +import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRanges; +import org.dbsp.util.Logger; + +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +/** Find operators whose state may grow without bound. + * + *

The analysis computes two properties: + *

    + *
  • "bounded", a property of streams: the integral of + * the stream is bounded.
    • + *
  • "bounded state", a property of operators. An operator may internally + * contain multiple integrators.
    • + *
+ * Stateful operators with unbounded state are collected in {@link #unbounded}. + * + *

Currently no one consumes the output produced by this pass, but it can be obtained + * by turning logging up using the compiler option -TFindUnboundedState=1 */ +public class FindUnboundedState extends Passes { + /** + * An operator whose state may grow without bound. + * + * @param operator The operator holding the state. + * @param unboundedInputs Indexes of the operator inputs that are not bounded. + */ + public record UnboundedOperator(DBSPOperator operator, List unboundedInputs) { } + + /** Streams whose integral is bounded */ + final Set bounded = new HashSet<>(); + /** Streams whose trace is pruned by a GC operator */ + final Set gcedStreams = new HashSet<>(); + /** Operators whose state may grow without bound */ + public final List unbounded = new ArrayList<>(); + + public FindUnboundedState(DBSPCompiler compiler) { + super("FindUnboundedState", compiler); + this.add(new FindGCedStreams(compiler)); + FindBounded findBounded = new FindBounded(compiler); + this.add(findBounded); + // Second run for recursive circuits + this.add(findBounded); + this.add(new CollectUnbounded(compiler)); + } + + @Override + public DBSPCircuit apply(DBSPCircuit circuit) { + this.bounded.clear(); + this.gcedStreams.clear(); + this.unbounded.clear(); + return super.apply(circuit); + } + + /** True if the integral of the stream is bounded */ + boolean isBounded(OutputPort port) { + return this.bounded.contains(port) || this.gcedStreams.contains(port); + } + + /** True if some output stream of the operator has its stream pruned by a GC operator */ + boolean hasGCedOutput(DBSPOperator operator) { + for (OutputPort port : this.gcedStreams) + if (port.operator == operator) + return true; + return false; + } + + /** True for the operators whose output is bounded when all their inputs are bounded. */ + static boolean propagatesBounded(DBSPOperator operator) { + if (operator.is(ILinear.class)) + return !operator.is(DBSPIntegrateOperator.class); + return operator.is(INonLinearAggregate.class) + || operator.is(DBSPDistinctOperator.class) + || operator.is(DBSPStreamDistinctOperator.class) + || operator.is(DBSPBinaryDistinctOperator.class) + || operator.is(DBSPPositiveOperator.class) + || operator.is(DBSPUpsertFeedbackOperator.class) + || operator.is(DBSPIndexedTopKOperator.class) + || operator.is(DBSPRankOperator.class) + || operator.is(DBSPRowNumberOperator.class) + || operator.is(IJoin.class); + } + + boolean allInputsBounded(DBSPOperator operator) { + for (OutputPort input : operator.inputs) + if (!this.isBounded(input)) + return false; + return true; + } + + /** + * Record the streams whose trace is pruned by a GC operator. + */ + class FindGCedStreams extends CircuitVisitor { + FindGCedStreams(DBSPCompiler compiler) { + super(compiler); + } + + @Override + public void postorder(DBSPOperator node) { + if (node.is(IGCOperator.class)) + FindUnboundedState.this.gcedStreams.add(node.to(DBSPBinaryOperator.class).left()); + } + } + + /** Compute the "bounded" stream property. */ + class FindBounded extends CircuitVisitor { + FindBounded(DBSPCompiler compiler) { + super(compiler); + } + + /** True if all output streams of the operator are bounded */ + boolean hasBoundedOutput(DBSPOperator node) { + if (node.is(DBSPWindowOperator.class)) + return !node.to(DBSPWindowOperator.class).lowerUnbounded; + // A waterline is a single value + if (node.is(DBSPWaterlineOperator.class) || node.is(DBSPConstantOperator.class)) + return true; + // Prunes its state and its output using its waterline input + if (node.is(DBSPPartitionedRollingAggregateWithWaterlineOperator.class)) + return true; + // The NOW system table always contains exactly one row + if (node.is(IInputOperator.class) && + node.to(IInputOperator.class).getTableName().equals(DBSPCompiler.NOW_TABLE_NAME)) + return true; + return propagatesBounded(node) && FindUnboundedState.this.allInputsBounded(node); + } + + @Override + public void postorder(DBSPOperator node) { + if (node.is(IGCOperator.class)) + return; + if (this.hasBoundedOutput(node)) + for (int i = 0; i < node.outputCount(); i++) + FindUnboundedState.this.bounded.add(node.getOutput(i)); + } + + @Override + public void postorder(DBSPViewDeclarationOperator node) { + ICircuit parent = this.getParent(); + if (!parent.is(DBSPNestedOperator.class)) + return; + OutputPort port = parent.to(DBSPNestedOperator.class).outputForDeclaration(node); + if (port != null && FindUnboundedState.this.isBounded(port)) + FindUnboundedState.this.bounded.add(node.outputPort()); + } + + @Override + public void postorder(DBSPNestedOperator node) { + for (int i = 0; i < node.outputCount(); i++) { + OutputPort internal = node.internalOutputs.get(i); + if (internal != null && FindUnboundedState.this.isBounded(internal)) + FindUnboundedState.this.bounded.add(node.getOutput(i)); + } + } + } + + /** + * Collect the stateful operators without bounded state; runs after the + * stream properties have been computed. + */ + class CollectUnbounded extends CircuitVisitor { + CollectUnbounded(DBSPCompiler compiler) { + super(compiler); + } + + boolean insideRecursive() { + return this.getParent().is(DBSPNestedOperator.class); + } + + @Override + public void postorder(DBSPDelayOperator node) { + if (this.insideRecursive()) + super.postorder(node); + } + + @Override + public void postorder(DBSPDifferentiateOperator node) { + if (this.insideRecursive()) + super.postorder(node); + } + + @Override + public void postorder(DBSPWaterlineOperator node) { + if (this.insideRecursive()) + super.postorder(node); + } + + @Override + public void postorder(DBSPAggregateLinearPostprocessRetainKeysOperator node) { + if (this.insideRecursive()) + super.postorder(node); + } + + @Override + public void postorder(DBSPPartitionedRollingAggregateWithWaterlineOperator node) { + if (this.insideRecursive()) + super.postorder(node); + } + + @Override + public void postorder(DBSPWindowOperator node) { + if (node.lowerUnbounded || this.insideRecursive()) + super.postorder(node); + } + + void markUnbounded(DBSPOperator operator, List inputs) { + var ub = new UnboundedOperator(operator, inputs); + FindUnboundedState.this.unbounded.add(ub); + + SourcePositionRanges pos = FindSourcePositions.getPositions(this.compiler, operator); + String sources = this.compiler.sources.getFragments(pos); + if (!sources.isEmpty()) + sources += "\n"; + Logger.INSTANCE.belowLevel(FindUnboundedState.class, 1) + .append("Potentially unbounded memory in operator ") + .append(operator.getClass().getSimpleName()) + .newline() + .append(sources); + } + + @Override + public void postorder(DBSPOperator node) { + if (!node.is(IStateful.class)) + return; + // Operators inside recursive circuits store a history of deltas, + // which can grow without bound + if (this.getParent().is(DBSPNestedOperator.class)) { + List all = new ArrayList<>(); + for (int input = 0; input < node.inputs.size(); input++) + all.add(input); + this.markUnbounded(node, all); + return; + } + // Operators whose output trace is pruned by a GC operator have bounded state + if (FindUnboundedState.this.hasGCedOutput(node)) + return; + List unbounded = new ArrayList<>(); + for (int input = 0; input < node.inputs.size(); input++) { + if (!FindUnboundedState.this.isBounded(node.inputs.get(input))) + unbounded.add(input); + } + if (node.inputs.isEmpty()) + // Stateful non-GCed input operator + this.markUnbounded(node, unbounded); + if (!unbounded.isEmpty()) + this.markUnbounded(node, unbounded); + } + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/Passes.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/Passes.java index d66bd1583c4..0ec935f8196 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/Passes.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/Passes.java @@ -80,7 +80,7 @@ public DBSPCircuit apply(DBSPCircuit circuit) { ToDot.dump(this.compiler, name, details, "png", circuit); } long begin = System.currentTimeMillis(); - Logger.INSTANCE.belowLevel(this, 1) + Logger.INSTANCE.belowLevel(this, 2) .append(this.toString()) .append(" starting ") .append(this.passes.size()) @@ -94,7 +94,7 @@ public DBSPCircuit apply(DBSPCircuit circuit) { break; long endId = DBSPNode.outerId; long end = System.currentTimeMillis(); - Logger.INSTANCE.belowLevel(this, 1) + Logger.INSTANCE.belowLevel(this, 2) .append(pass.toString()) .append(" took ") .append(end - start) @@ -109,7 +109,7 @@ public DBSPCircuit apply(DBSPCircuit circuit) { } } long finish = System.currentTimeMillis(); - Logger.INSTANCE.belowLevel(this, 1) + Logger.INSTANCE.belowLevel(this, 2) .decrease() .append(this.toString()) .append(" took ") diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/ToJsonVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/ToJsonVisitor.java index ee58808d9ce..9094be01695 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/ToJsonVisitor.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/ToJsonVisitor.java @@ -13,25 +13,18 @@ import org.dbsp.sqlCompiler.circuit.operator.DBSPSourceTableOperator; import org.dbsp.sqlCompiler.circuit.operator.DBSPViewDeclarationOperator; import org.dbsp.sqlCompiler.compiler.DBSPCompiler; -import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRange; import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRanges; import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteRelNode; import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision; -import org.dbsp.sqlCompiler.compiler.visitors.inner.InnerVisitor; -import org.dbsp.sqlCompiler.ir.DBSPParameter; -import org.dbsp.sqlCompiler.ir.IDBSPInnerNode; import org.dbsp.sqlCompiler.ir.IDBSPOuterNode; -import org.dbsp.sqlCompiler.ir.expression.DBSPExpression; import org.dbsp.util.HashString; import org.dbsp.util.IIndentStream; import org.dbsp.util.Linq; import org.dbsp.util.Utilities; import java.util.ArrayList; -import java.util.HashSet; import java.util.List; import java.util.Map; -import java.util.Set; /** Emit a circuit description as JSON. * Currently only the dataflow graph is emitted. @@ -41,42 +34,6 @@ public class ToJsonVisitor extends CircuitVisitor { final int verbosity; final Map relId; - public static class FindSourcePositions extends InnerVisitor { - public final Set positions; - private final boolean reset; - - public FindSourcePositions(DBSPCompiler compiler, boolean reset) { - super(compiler); - this.positions = new HashSet<>(); - this.reset = reset; - } - - @Override - public void postorder(DBSPExpression expression) { - SourcePositionRange positionRange = expression.getNode().getPositionRange(); - if (positionRange.isValid()) - this.positions.add(positionRange); - } - - @Override - public void postorder(DBSPParameter parameter) { - SourcePositionRange positionRange = parameter.getNode().getPositionRange(); - if (positionRange.isValid()) - this.positions.add(positionRange); - } - - @Override - public void startVisit(IDBSPInnerNode node) { - super.startVisit(node); - if (this.reset) - this.positions.clear(); - } - - public SourcePositionRanges getPositions() { - return new SourcePositionRanges(this.positions); - } - } - public ToJsonVisitor(DBSPCompiler compiler, IIndentStream builder, int verbosity, Map id) { super(compiler); @@ -85,13 +42,6 @@ public ToJsonVisitor(DBSPCompiler compiler, IIndentStream builder, int verbosity this.relId = id; } - SourcePositionRanges getPositions(DBSPOperator operator) { - FindSourcePositions positions = new FindSourcePositions(this.compiler, true); - operator.accept(positions); - positions.positions.addAll(operator.getSourcePositions()); - return positions.getPositions(); - } - void emitPort(OutputPort port) { String inputName = port.operator.getCompactName(); this.builder.append("{ ") @@ -160,12 +110,14 @@ void process(DBSPSimpleOperator operator) { this.builder.append(",").newline(); this.builder.appendJsonLabelAndColon("positions") .append("["); - var list = Linq.list(this.getPositions(operator)); + var list = Linq.list(FindSourcePositions.getPositions(this.compiler, operator)); if (operator.is(DBSPSourceTableOperator.class) || operator.is(DBSPSinkOperator.class)) { if (operator.getSourcePosition().isValid()) list.add(operator.getSourcePosition()); } - List strings = Linq.map(list, p -> p.asJson().toString()); + // Deduplicate positions + SourcePositionRanges ranges = new SourcePositionRanges(list); + List strings = Linq.map(ranges.positions, p -> p.asJson().toString()); if (!strings.isEmpty()) { this.builder .increase() @@ -211,7 +163,7 @@ void processWithError(DBSPOperatorWithError operator) { this.builder.append(",").newline(); this.builder.appendJsonLabelAndColon("positions") .append("["); - var list = Linq.list(this.getPositions(operator)); + var list = Linq.list(FindSourcePositions.getPositions(this.compiler, operator)); List strings = Linq.map(list, p -> p.asJson().toString()); if (!strings.isEmpty()) { this.builder @@ -258,7 +210,7 @@ void processInputMapWithWaterline(DBSPInputMapWithWaterlineOperator operator) { this.builder.append(",").newline(); this.builder.appendJsonLabelAndColon("positions") .append("["); - var list = Linq.list(this.getPositions(operator)); + var list = Linq.list(FindSourcePositions.getPositions(this.compiler, operator)); List strings = Linq.map(list, p -> p.asJson().toString()); if (!strings.isEmpty()) { this.builder diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/InsertLimiters.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/InsertLimiters.java index 5df2101530c..c68368a9601 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/InsertLimiters.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/monotonicity/InsertLimiters.java @@ -1899,7 +1899,9 @@ public void postorder(DBSPSourceMultisetOperator operator) { fields.get(0).getType(), dataType), true, replacement.getOutput(0)); this.addOperator(ix); DBSPWindowOperator window = new DBSPWindowOperator( - operator.getRelNode(), true, true, ix.outputPort(), apply.outputPort()); + operator.getRelNode(), true, true, + // -infinity + true, ix.outputPort(), apply.outputPort()); this.addOperator(window); replacement = new DBSPDeindexOperator(operator.getRelNode(), operator.getNode(), window.outputPort()); } @@ -2193,7 +2195,9 @@ public void postorder(DBSPSinkOperator operator) { this.addOperator(ix); // The upper bound must be exclusive DBSPWindowOperator window = new DBSPWindowOperator( - operator.getRelNode(), true, false, ix.outputPort(), apply.outputPort()); + operator.getRelNode(), true, false, + // -infinity + true, ix.outputPort(), apply.outputPort()); this.addOperator(window); // GC for window: the waterline delayed PartiallyMonotoneTuple projection = new PartiallyMonotoneTuple( diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/temporal/RewriteNow.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/temporal/RewriteNow.java index a1e2fb022cf..37e7f73d18b 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/temporal/RewriteNow.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/temporal/RewriteNow.java @@ -30,7 +30,7 @@ import org.dbsp.sqlCompiler.compiler.visitors.inner.ReferenceMap; import org.dbsp.sqlCompiler.compiler.visitors.inner.ResolveReferences; import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitCloneVisitor; -import org.dbsp.sqlCompiler.compiler.visitors.outer.ToJsonVisitor; +import org.dbsp.sqlCompiler.compiler.visitors.outer.FindSourcePositions; import org.dbsp.sqlCompiler.compiler.visitors.outer.monotonicity.InsertLimiters; import org.dbsp.sqlCompiler.ir.DBSPParameter; import org.dbsp.sqlCompiler.ir.IDBSPDeclaration; @@ -377,11 +377,13 @@ DBSPSimpleOperator implementTemporalFilter(DBSPFilterOperator operator, boolean lowerInclusive = bounds.lower() == null || bounds.lower().inclusive(); boolean upperInclusive = bounds.upper() == null || bounds.upper().inclusive(); CalciteRelNode windowNode = relNode.copy(); - ToJsonVisitor.FindSourcePositions finder = new ToJsonVisitor.FindSourcePositions(this.compiler, false); + FindSourcePositions finder = new FindSourcePositions(this.compiler, false); finder.apply(makeWindow); windowNode.addSourcePositions(finder.positions); DBSPSimpleOperator window = new DBSPWindowOperator( - windowNode, lowerInclusive, upperInclusive, diffIndex.outputPort(), windowBounds.outputPort()); + windowNode, lowerInclusive, upperInclusive, + // -infinity if not specified + bounds.lower() == null, diffIndex.outputPort(), windowBounds.outputPort()); this.addOperator(window); DBSPSimpleOperator winInt = new DBSPIntegrateOperator(relNode, window.outputPort()); this.addOperator(winInt); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/unusedFields/TrimWindows.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/unusedFields/TrimWindows.java index a863d289050..e5585e48abb 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/unusedFields/TrimWindows.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/unusedFields/TrimWindows.java @@ -44,7 +44,8 @@ public void postorder(DBSPMapIndexOperator operator) { this.addOperator(pre); DBSPWindowOperator newWindow = new DBSPWindowOperator( - window.getRelNode(), window.lowerInclusive, window.upperInclusive, pre.outputPort(), window.right()); + window.getRelNode(), window.lowerInclusive, window.upperInclusive, + window.lowerUnbounded, pre.outputPort(), window.right()); this.addOperator(newWindow); DBSPUnaryOperator postProj = new DBSPMapIndexOperator( diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWindowOperatorTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWindowOperatorTests.java index 9c0b0018c3f..5768502a382 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWindowOperatorTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/circuit/operator/DBSPWindowOperatorTests.java @@ -44,7 +44,7 @@ private static OutputPort controlInput( private static DBSPWindowOperator window( DBSPExpression lower, DBSPExpression upper) { - return new DBSPWindowOperator(CalciteEmptyRel.INSTANCE, true, true, + return new DBSPWindowOperator(CalciteEmptyRel.INSTANCE, true, true, false, dataInput().outputPort(), controlInput(lower, upper)); } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression1Tests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression1Tests.java index d7c3560457f..b63b0986ced 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression1Tests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression1Tests.java @@ -536,7 +536,7 @@ public void castBinaryToString() { .withStringTrim(); ccs.stepWeightOne("INSERT INTO T VALUES(x'AB01')", """ x | y - ----------- + ------------- ab01 | ab01"""); } @@ -1534,7 +1534,7 @@ SELECT x, x in (SELECT e from FT) FROM T;""").withStringTrim(); ccs.stepWeightOne("INSERT INTO T VALUES('a'), ('b'), ('ab');", """ x | in - --------- + ----------- a | true b | false ab | false"""); @@ -1614,7 +1614,7 @@ public void issue5345() { ) AS t (f1, f2, f3, f4);""").withStringTrim(); ccs.stepWeightOne("", """ f1 | f2 | f3 | f4 - ---------------------- + ------------------------ a | 1 | { by} | true b | 1 | {} | false"""); } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java index e07982bf64b..4d1a4963cc1 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/Regression2Tests.java @@ -1144,7 +1144,7 @@ public void issue6590b() { ('e', NULL); """, """ g | max | sum - -------------- + ---------------- a | 0 | 1 b | 1 | 3 c | 1 | 8 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 4557753161e..b18bb184108 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 @@ -24,6 +24,7 @@ import org.dbsp.sqlCompiler.compiler.sql.tools.InputOutputChange; import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision; import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor; +import org.dbsp.sqlCompiler.compiler.visitors.outer.FindUnboundedState; import org.dbsp.sqlCompiler.ir.expression.DBSPTupleExpression; import org.dbsp.sqlCompiler.ir.expression.literal.DBSPDateLiteral; import org.dbsp.sqlCompiler.ir.expression.literal.DBSPDoubleLiteral; @@ -51,12 +52,12 @@ CREATE TABLE t1( x INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE TABLE t2( y INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE VIEW v WITH ('emit_final' = 'ts') AS SELECT t1.ts @@ -259,7 +260,7 @@ CREATE TABLE t ( ) WITH ( 'append_only' = 'true' ); - + create view v1 AS SELECT TIMESTAMP_TRUNC(ts, DAY) as d, @@ -327,12 +328,12 @@ CREATE TABLE t1( x INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE TABLE t2( y INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE VIEW v WITH ('emit_final' = 'ts') AS SELECT @@ -397,7 +398,7 @@ CREATE TABLE T ( 'materialized' = 'true', 'append_only' = 'true' ); - + CREATE VIEW V WITH ('emit_final' = 'ts') AS SELECT * FROM T @@ -674,13 +675,13 @@ create table TRANSACTION ( id bigint NOT NULL, unix_time BIGINT LATENESS 100 ); - + create table FEEDBACK ( id bigint, status int, unix_time bigint NOT NULL LATENESS 100 ); - + CREATE VIEW TRANSACT AS SELECT feedback.*, transaction.* FROM @@ -2044,9 +2045,9 @@ CREATE TABLE data ( t0 TIMESTAMP NOT NULL LATENESS INTERVAL '2' HOURS, location INT NOT NULL ); - + CREATE LOCAL VIEW IT AS SELECT (t0 - TIMESTAMP '2020-01-01 00:00:00') HOURS AS t, location FROM data; - + CREATE VIEW V AS SELECT *, @@ -3056,7 +3057,7 @@ create table T ( y TIMESTAMP, site_id varchar ); - + create view V as select site_id from T where ( x >= NOW() + INTERVAL 30 DAYS @@ -3342,7 +3343,7 @@ CREATE TABLE T( lp VARCHAR, lsd TIMESTAMP ); - + create view V as SELECT s @@ -3390,7 +3391,7 @@ CREATE TABLE T( lp VARCHAR, lsd TIMESTAMP ); - + create view V as SELECT s @@ -3444,7 +3445,7 @@ CREATE TABLE T( lp VARCHAR, lsd TIMESTAMP ); - + create view V as SELECT s @@ -3502,7 +3503,7 @@ create table T ( properties variant, site_id varchar ); - + create view V as (select site_id from T where CAST(properties['x'] AS TIMESTAMP) >= NOW() + INTERVAL 30 DAYS) @@ -3520,12 +3521,12 @@ CREATE TABLE t1( x INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE TABLE t2( y INT, ts TIMESTAMP NOT NULL LATENESS INTERVAL 1 HOUR ); - + CREATE VIEW v WITH ('emit_final' = 'ts') AS SELECT t1.ts @@ -3537,11 +3538,7 @@ CREATE TABLE t2( public void changeLog() { // TLOG is a log of insertions and deletions applied to a table with // primary key t_key; view T reconstructs the current table contents - // from the log entries of the last 25 hours: the latest entry per key - // wins, and a key whose latest entry is a deletion is absent. The op - // filter must sit outside the TOP-1, otherwise a deletion would - // resurrect the previous insertion. The temporal filter makes rows - // age out of T once their latest entry falls behind the window. + // from the log entries of the last 25 hours. String sql = """ CREATE TABLE TLOG ( t_key INT NOT NULL, @@ -3563,6 +3560,13 @@ CREATE TABLE TLOG ( ) latest WHERE rn = 1 AND op = 'insert';"""; var ccs = this.getCCS(sql).withStringTrim(); + FindUnboundedState gc = new FindUnboundedState(ccs.compiler); + ccs.visit(gc); + // The temporal filter's window operator bounds its own state; its + // bounded output propagates to the TOP-1 operator, and the NOW-derived + // window-bound computation is bounded because the NOW table holds one + // row. The circuit has no unbounded state. + Assert.assertTrue(gc.unbounded.toString(), gc.unbounded.isEmpty()); ccs.step(""" INSERT INTO NOW VALUES('2020-01-01 01:00:00'); INSERT INTO TLOG VALUES(1, 'aaa', 'insert', '2020-01-01 00:00:00'); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDF.json b/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDF.json index 734ca856549..e6d832f6412 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDF.json +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDF.json @@ -174,7 +174,6 @@ "final": 3 }, "positions": [ - {"start_line_number":2,"start_column":1,"end_line_number":2,"end_column":40}, {"start_line_number":2,"start_column":1,"end_line_number":2,"end_column":40} ], "persistent_id": "cf472d4e3713ed0e8c91ebbf46cd813eaff94a440f8779390a9c96b745c19bab" diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json b/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json index 116820e0c51..24df341a121 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/resources/metadataTests-generateDFRecursive.json @@ -454,7 +454,6 @@ "final": 10 }, "positions": [ - {"start_line_number":4,"start_column":1,"end_line_number":21,"end_column":1}, {"start_line_number":4,"start_column":1,"end_line_number":21,"end_column":1} ], "persistent_id": "08b99a30ddf054bdadc7c1e3588e18db716ca20bdccad6761e24181bc61ccc63" From 4794994c04cd9ca7bc47009be33f568f0f54d39b Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Tue, 4 Aug 2026 15:28:34 -0700 Subject: [PATCH 4/4] [DOCS] Example combining soft-deletes, temporal windows, and window aggregates Signed-off-by: Mihai Budiu --- docs.feldera.com/docs/connectors/index.mdx | 34 +----- docs.feldera.com/docs/sql/streaming.md | 112 +++++++++++++++++- .../sql/streaming/StreamingTests.java | 63 ++++++++++ 3 files changed, 178 insertions(+), 31 deletions(-) diff --git a/docs.feldera.com/docs/connectors/index.mdx b/docs.feldera.com/docs/connectors/index.mdx index 1048e03c149..0443e36c941 100644 --- a/docs.feldera.com/docs/connectors/index.mdx +++ b/docs.feldera.com/docs/connectors/index.mdx @@ -260,39 +260,13 @@ message that deletes the old value of a record and inserts the new one, so those two changes carry one timestamp; ranking the insertion first keeps the updated record, where ranking the deletion first would drop it. -{/* Revise after implementing the append-only optimization for soft-delete tables. - -#### Bounding the state of the query - As written, `live` needs unbounded state. A change carrying any timestamp can arrive at any moment and displace the record that is currently the latest one for its key, so Feldera has to keep every insertion and deletion the stream -ever reported. - -An application that only needs values from a bounded time frame can filter the -changes with a -[temporal filter](/tutorials/time-series#now-and-temporal-filters) before -ranking them, which leaves Feldera storing only the changes newer than the -lower bound of the interval: - -```sql -CREATE MATERIALIZED VIEW live_recent AS -SELECT id, s, ts -FROM ( - SELECT *, ROW_NUMBER() OVER ( - PARTITION BY id ORDER BY ts DESC, is_delete NULLS FIRST - ) AS rn - FROM changes - WHERE ts >= NOW() - INTERVAL 7 DAYS -) -WHERE rn = 1 AND is_delete IS NOT TRUE; -``` - -A record whose latest change ages out of the window leaves `live_recent`, so -this variant reports what is live among the keys the stream touched in the last -seven days rather than what is live overall. - -*/} +ever reported. An application that only needs values from a bounded time frame +can filter the changes with a temporal filter before ranking them, which +bounds the state of the query; see +[Soft deletes with temporal filters](/sql/streaming#soft-deletes-with-temporal-filters). Notes and restrictions: diff --git a/docs.feldera.com/docs/sql/streaming.md b/docs.feldera.com/docs/sql/streaming.md index bf10e08cc4d..f269be2a0a1 100644 --- a/docs.feldera.com/docs/sql/streaming.md +++ b/docs.feldera.com/docs/sql/streaming.md @@ -54,4 +54,114 @@ The `emit_final` annotation on a view instructs Feldera to only output its final i.e., rows that are guaranteed to never get deleted or updated. See the [Time Series Analysis Guide](/tutorials/time-series#emitting-final-values-of-a-view-with-emit_final) -for details. \ No newline at end of file +for details. + +## Soft deletes with temporal filters + +An input connector can be configured with the [`soft_delete`](/connectors#soft_delete) +property to transform deletions into insertions; in this case the `is_delete` metadata +attribute records the kind of change (insert/delete), essentially converting a table +into a log. Note that the table does *not* declare a `PRIMARY KEY` column, although +the data may contain one. Since the connector transforms every change into an +insertion, the table only receives insertions, so it can be declared +[`append_only`](#append_only-tables), which enables additional optimizations. + +The [Soft deletes](/connectors#soft-deletes) section shows how one can write a query +to recover the current contents of the table from this log: group the changes +on the columns forming the primary key, rank them by time, keep the latest one, +and return it only when it is an insertion. That query returns one row for each +primary key, but it must remember every change in the log, so its state grows +without bound. + +However, in some cases only a bounded window of the table is necessary for +computing the desired results. When a +[temporal filter](/tutorials/time-series#now-and-temporal-filters) can be used +to describe the window, the entire computation can be performed using finite +state, by sequencing the computation as follows: + +``` +[connector with soft deletes] -> [temporal filter] -> [reconstruct table] -> [views] +``` + +The following program reconstructs only the recent contents +of a change stream while never storing more than the last seven +days of changes: + +```sql +-- The 'soft_delete' connector property converts this +-- table into a log of changes to the table. +CREATE TABLE input_log ( + id BIGINT, -- not declared as primary key + s VARCHAR, + ts TIMESTAMP, + -- Is the change a deletion? Produced by the connector + is_delete BOOLEAN DEFAULT CAST(CONNECTOR_METADATA()['is_delete'] AS BOOLEAN) +) WITH ( + -- A soft-delete table only receives insertions + 'append_only' = 'true', + 'connectors' = '[{ + "name": "changes", + "soft_delete": true, + "transport": { + "name": "kafka_input", + "config": { + "topic": "changes", + "start_from": "earliest", + "bootstrap.servers": "example.com:9092", + "include_timestamp": true + } + }, + "format": { + "name": "json", + "config": { "update_format": "insert_delete" } + } + }]' +); + +-- Contains only changes to 'input_log' from the last 7 days +CREATE LOCAL VIEW recent AS +SELECT * FROM input_log +WHERE ts >= NOW() - INTERVAL 7 DAYS AND ts <= NOW(); + +-- The contents of the 'input' table limited to the last 7 days +CREATE LOCAL VIEW input AS +SELECT id, s, ts +FROM ( + SELECT *, ROW_NUMBER() OVER ( + PARTITION BY id ORDER BY ts DESC, is_delete NULLS FIRST + ) AS rn + FROM recent +) +WHERE rn = 1 AND is_delete IS NOT TRUE; + +-- Rolling aggregates over the reconstructed table: for each record, +-- the number of records with a timestamp in the preceding minute, hour, and day. +CREATE VIEW input_stats AS +SELECT + id, s, ts, + COUNT(*) OVER minute_window AS rows_last_minute, + COUNT(*) OVER hour_window AS rows_last_hour, + COUNT(*) OVER day_window AS rows_last_day +FROM input +WINDOW + minute_window AS (ORDER BY ts RANGE BETWEEN INTERVAL 1 MINUTE PRECEDING AND CURRENT ROW), + hour_window AS (ORDER BY ts RANGE BETWEEN INTERVAL 1 HOUR PRECEDING AND CURRENT ROW), + day_window AS (ORDER BY ts RANGE BETWEEN INTERVAL 1 DAY PRECEDING AND CURRENT ROW); +``` + +The view `input_stats` consumes the reconstructed table using +[rolling aggregates](/tutorials/time-series#rolling-aggregates) over three +shorter intervals. The aggregates see the reconstructed table rather than the +log, so a deleted record stops contributing to the counts as soon as its +deletion arrives. + +Two details of the query matter for correctness: + +* The `is_delete` term in the `ORDER BY` clause ranks an insertion ahead of a + deletion that carries the same timestamp, which keeps the new value of a + record that a CDC stream updates with a single delete-insert message pair. + See [Soft deletes](/connectors#soft-deletes) for details. + +* The polarity filter `is_delete IS NOT TRUE` must be outside the subquery + that ranks the changes. Filtering out the deletions before ranking would + "resurrect" the previous insertion of a deleted key. 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 b18bb184108..799f94321d5 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 @@ -3617,4 +3617,67 @@ CREATE TABLE TLOG ( -------------------------- 2 | ddd | -1"""); } + + @Test + public void softDeleteLog() { + // The example from docs/sql/streaming.md: reconstruct the current + // contents of a soft-deleted stream with bounded state and aggregate + // over it. + String sql = """ + CREATE TABLE input_log ( + id BIGINT, + s VARCHAR, + ts TIMESTAMP, + is_delete BOOLEAN DEFAULT CAST(CONNECTOR_METADATA()['is_delete'] AS BOOLEAN) + ) WITH ( + 'append_only' = 'true', + 'connectors' = '[{ + "name": "changes", + "soft_delete": true, + "transport": { + "name": "kafka_input", + "config": { + "topic": "changes", + "start_from": "earliest", + "bootstrap.servers": "example.com:9092", + "include_timestamp": true + } + }, + "format": { + "name": "json", + "config": { "update_format": "insert_delete" } + } + }]' + ); + + CREATE LOCAL VIEW recent AS + SELECT * FROM input_log + WHERE ts >= NOW() - INTERVAL 7 DAYS AND ts <= NOW(); + + CREATE LOCAL VIEW input AS + SELECT id, s, ts + FROM ( + SELECT *, ROW_NUMBER() OVER ( + PARTITION BY id ORDER BY ts DESC, is_delete NULLS FIRST + ) AS rn + FROM recent + ) + WHERE rn = 1 AND is_delete IS NOT TRUE; + + CREATE VIEW input_stats AS + SELECT + id, s, ts, + COUNT(*) OVER minute_window AS rows_last_minute, + COUNT(*) OVER hour_window AS rows_last_hour, + COUNT(*) OVER day_window AS rows_last_day + FROM input + WINDOW + minute_window AS (ORDER BY ts RANGE BETWEEN INTERVAL 1 MINUTE PRECEDING AND CURRENT ROW), + hour_window AS (ORDER BY ts RANGE BETWEEN INTERVAL 1 HOUR PRECEDING AND CURRENT ROW), + day_window AS (ORDER BY ts RANGE BETWEEN INTERVAL 1 DAY PRECEDING AND CURRENT ROW);"""; + var ccs = this.getCCS(sql); + FindUnboundedState gc = new FindUnboundedState(ccs.compiler); + ccs.visit(gc); + Assert.assertTrue(gc.unbounded.toString(), gc.unbounded.isEmpty()); + } }