From a0d01b33b7982ef4b59ac7e08c0bdc2020f01ece Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Wed, 26 Jun 2024 23:31:41 -0700 Subject: [PATCH 1/7] Support materialized views in the catalog API. Signed-off-by: Leonid Ryzhyk --- crates/adapters/src/static_compile/catalog.rs | 192 +++++++++++++++++- crates/adapters/src/test/mod.rs | 4 +- demo/project_demo00-SecOps/project.sql | 5 +- docs/tour/tour.md | 2 +- docs/tutorials/basics/part1.md | 13 +- docs/tutorials/debugging-sql.md | 6 +- docs/tutorials/rest_api/index.md | 2 +- docs/tutorials/writing-sql.md | 4 +- 8 files changed, 215 insertions(+), 13 deletions(-) diff --git a/crates/adapters/src/static_compile/catalog.rs b/crates/adapters/src/static_compile/catalog.rs index 334c099cdaa..4f6d5505e4d 100644 --- a/crates/adapters/src/static_compile/catalog.rs +++ b/crates/adapters/src/static_compile/catalog.rs @@ -44,6 +44,7 @@ impl Catalog { )) }) } + /// Add an input stream of Z-sets to the catalog. /// /// Adds a `DeCollectionHandle` to the catalog, which will deserialize @@ -78,6 +79,37 @@ impl Catalog { self.register_output_zset(stream, schema); } + /// Like `register_input_zset`, but additionally materializes the integral + /// of the stream and makes it queryable. + pub fn register_materialized_input_zset( + &mut self, + stream: Stream, + handle: ZSetHandle, + schema: &str, + ) where + D: for<'de> DeserializeWithContext<'de, SqlSerdeConfig> + + SerializeWithContext + + From + + Clone + + Debug + + Send + + 'static, + Z: ZSet + Debug + Send + Sync, + Z::InnerBatch: Send, + Z::Key: Sync + From, + { + let relation_schema: Relation = Self::parse_relation_schema(schema).unwrap(); + + self.register_input_collection_handle(InputCollectionHandle::new( + relation_schema, + DeZSetHandle::new(handle), + )) + .unwrap(); + + // Inputs are also outputs. + self.register_materialized_output_zset(stream, schema); + } + /// Add an input stream created using `add_input_set` to catalog. /// /// Adds a `DeCollectionHandle` to the catalog, which will deserialize @@ -112,6 +144,37 @@ impl Catalog { self.register_output_zset(stream, schema); } + /// Like `register_input_set`, but additionally materializes the integral + /// of the stream and makes it queryable. + pub fn register_materialized_input_set( + &mut self, + stream: Stream, + handle: SetHandle, + schema: &str, + ) where + D: for<'de> DeserializeWithContext<'de, SqlSerdeConfig> + + SerializeWithContext + + From + + Clone + + Debug + + Send + + 'static, + Z: ZSet + Debug + Send + Sync, + Z::InnerBatch: Send, + Z::Key: Sync + From, + { + let relation_schema: Relation = Self::parse_relation_schema(schema).unwrap(); + + self.register_input_collection_handle(InputCollectionHandle::new( + relation_schema, + DeSetHandle::new(handle), + )) + .unwrap(); + + // Inputs are also outputs. + self.register_materialized_output_zset(stream, schema); + } + /// Register an input handle created using `add_input_map`. /// /// Elements are inserted by value and deleted by key. On insert, the @@ -173,6 +236,52 @@ impl Catalog { self.register_output_map(stream, value_key_func, schema); } + /// Like `register_input_map`, but additionally materializes the integral + /// of the stream and makes it queryable. + pub fn register_materialized_input_map( + &mut self, + stream: Stream>, + handle: MapHandle, + value_key_func: VF, + update_key_func: UF, + schema: &str, + ) where + VF: Fn(&V) -> K + Clone + Send + Sync + 'static, + UF: Fn(&U) -> K + Clone + Send + Sync + 'static, + KD: for<'de> DeserializeWithContext<'de, SqlSerdeConfig> + + SerializeWithContext + + From + + Send + + 'static, + VD: for<'de> DeserializeWithContext<'de, SqlSerdeConfig> + + SerializeWithContext + + From + + Clone + + Debug + + Default + + Send + + 'static, + UD: for<'de> DeserializeWithContext<'de, SqlSerdeConfig> + + SerializeWithContext + + From + + Send + + 'static, + K: DBData + Sync + From, + V: DBData + Sync + From, + U: DBData + Sync + From, + { + let relation_schema: Relation = Self::parse_relation_schema(schema).unwrap(); + + self.register_input_collection_handle(InputCollectionHandle::new( + relation_schema, + DeMapHandle::new(handle, value_key_func.clone(), update_key_func.clone()), + )) + .unwrap(); + + // Inputs are also outputs. + self.register_materialized_output_map(stream, value_key_func, schema); + } + /// Add an output stream of Z-sets to the catalog. pub fn register_output_zset(&mut self, stream: Stream, schema: &str) where @@ -189,6 +298,44 @@ impl Catalog { { let schema: Relation = Self::parse_relation_schema(schema).unwrap(); + // Create handle for the stream itself. + let delta_handle = stream.output(); + + let handles = OutputCollectionHandles { + schema, + delta_handle: Box::new(>::new(delta_handle)) + as Box, + + neighborhood_descr_handle: None, + neighborhood_handle: None, + neighborhood_snapshot_handle: None, + num_quantiles_handle: None, + quantiles_handle: None, + }; + + self.register_output_batch_handles(handles).unwrap(); + } + + /// Like `register_output_zset`, but additionally materializes the integral + /// of the stream and makes it queryable. + pub fn register_materialized_output_zset( + &mut self, + stream: Stream, + schema: &str, + ) where + D: for<'de> DeserializeWithContext<'de, SqlSerdeConfig> + + SerializeWithContext + + From + + Clone + + Debug + + Send + + 'static, + Z: ZSet + Debug + Send + Sync, + Z::InnerBatch: Send, + Z::Key: Sync + From, + { + let schema: Relation = Self::parse_relation_schema(schema).unwrap(); + let circuit = stream.circuit(); // Create handle for the stream itself. @@ -287,6 +434,49 @@ impl Catalog { /// streams contain values only. Clients, e.g., the web console, can /// work with maps and z-sets in the same way. pub fn register_output_map( + &mut self, + stream: Stream>, + _key_func: F, + schema: &str, + ) where + F: Fn(&V) -> K + Clone + Send + Sync + 'static, + KD: for<'de> DeserializeWithContext<'de, SqlSerdeConfig> + + SerializeWithContext + + From, + VD: for<'de> DeserializeWithContext<'de, SqlSerdeConfig> + + SerializeWithContext + + From + + Default + + Debug + + Clone + + Send + + 'static, + K: DBData + Send + Sync + From + Default, + V: DBData + Send + Sync + From + Default, + { + let schema: Relation = Self::parse_relation_schema(schema).unwrap(); + + // Create handle for the stream itself. + let delta_handle = stream.map(|(_k, v)| v.clone()).output(); + + let handles = OutputCollectionHandles { + schema, + delta_handle: Box::new(>::new(delta_handle)) + as Box, + + neighborhood_descr_handle: None, + neighborhood_handle: None, + neighborhood_snapshot_handle: None, + num_quantiles_handle: None, + quantiles_handle: None, + }; + + self.register_output_batch_handles(handles).unwrap(); + } + + /// Like `register_output_map`, but additionally materializes the integral + /// of the stream and makes it queryable. + pub fn register_materialized_output_map( &mut self, stream: Stream>, key_func: F, @@ -493,7 +683,7 @@ mod test { let (input, hinput) = circuit.add_input_map::(|v, u| *v = u.clone()); - catalog.register_input_map::( + catalog.register_materialized_input_map::( input.clone(), hinput, |test_struct| test_struct.id, diff --git a/crates/adapters/src/test/mod.rs b/crates/adapters/src/test/mod.rs index ad6c3948186..4d67f30d39d 100644 --- a/crates/adapters/src/test/mod.rs +++ b/crates/adapters/src/test/mod.rs @@ -160,8 +160,8 @@ where let output_schema = serde_json::to_string(&Relation::new("test_output1", false, schema)).unwrap(); - catalog.register_input_zset(input.clone(), hinput, &input_schema); - catalog.register_output_zset(input, &output_schema); + catalog.register_materialized_input_zset(input.clone(), hinput, &input_schema); + catalog.register_materialized_output_zset(input, &output_schema); Ok(catalog) }) diff --git a/demo/project_demo00-SecOps/project.sql b/demo/project_demo00-SecOps/project.sql index 88a0a0f190c..9d26f56b3a0 100644 --- a/demo/project_demo00-SecOps/project.sql +++ b/demo/project_demo00-SecOps/project.sql @@ -40,7 +40,10 @@ create table vulnerability ( vulnerability_reference_id varchar not null, severity int, priority varchar -); +) + -- Instruct Feldera to store the snapshot of the table, allowing the + -- user to browse it via the UI or API. + with ('materialized' = 'true'); -- K8s clusters. create table k8scluster ( diff --git a/docs/tour/tour.md b/docs/tour/tour.md index 620032f41b8..28c5d9c99be 100644 --- a/docs/tour/tour.md +++ b/docs/tour/tour.md @@ -128,7 +128,7 @@ changing. The view of the running pipeline should look something like this: ![Running pipeline](running-pipeline.png) -Each row that lists a SQL table or view includes, in addition to names +Rows for materialized tables and view includes, in addition to names and metrics, an eye icon for an action to view data received or sent through the connector. Click on the eye for some row to see how it works: diff --git a/docs/tutorials/basics/part1.md b/docs/tutorials/basics/part1.md index 14e66af1414..82e4494ec34 100644 --- a/docs/tutorials/basics/part1.md +++ b/docs/tutorials/basics/part1.md @@ -49,7 +49,7 @@ create table VENDOR ( id bigint not null primary key, name varchar, address varchar -); +) with ('materialized' = 'true'); create table PART ( id bigint not null primary key, @@ -74,6 +74,10 @@ could arrive from a Kafka stream, a database, or an HTTP request. Below we will see how our SQL program can be instantiated with any of these data sources, or even multiple data sources connected to the same table. +Finally, note the `'materialized' = 'true'` attribute on the `VENDOR` +table. This annotation instructs Feldera to store the entire contents of the table, +so that the user can browse it at any time. + ## Step 2. Write queries We would like to compute the lowest price for each part @@ -88,7 +92,7 @@ create view LOW_PRICE ( select part, MIN(price) as price from PRICE group by part; -- Lowest available price for each part along with part and vendor details. -create view PREFERRED_VENDOR ( +create materialized view PREFERRED_VENDOR ( part_id, part_name, vendor_id, @@ -118,6 +122,11 @@ tables and other views, making it possible to express deeply nested queries. In this example, the `PREFERRED_VENDOR` view is expressed in terms of the `LOW_PRICE` view. +We declare `PREFERRED_VENDOR` as a **materialized** view, instructing Feldera to +store the entire contents of the view, so that the user can browse it at any time. +This is in contrast to regular views, for which the user can only observe a stream +of **changes** to the view, but cannot inspect its current contents. + ## Step 3. Run the program In order to run our SQL program, we must instantiate it as part of a _pipeline_. diff --git a/docs/tutorials/debugging-sql.md b/docs/tutorials/debugging-sql.md index b3cb00770d0..e2fd7c527cf 100644 --- a/docs/tutorials/debugging-sql.md +++ b/docs/tutorials/debugging-sql.md @@ -9,8 +9,8 @@ CREATE TABLE Person name VARCHAR, age INT, present BOOLEAN -); -CREATE VIEW Adult AS SELECT Person.name, Person.age FROM Person WHERE Person.age > 18; +) with ('materialized' = 'true'); +CREATE MATERIALIZED VIEW Adult AS SELECT Person.name, Person.age FROM Person WHERE Person.age > 18; ``` Enter the code in the SQL editor. Once the program compiled successfully, (as @@ -39,7 +39,7 @@ data is generated for every field by opening the `RNG Settings`: Once you are happy with the generated rows (you can edit them after generation by double clicking the cells in the table), press `INSERT ROWS` which persists the your rows in the table. Switch the tab from `INSERT NEW ROWS` to `BROWSE -USERS` to see the content you just added in the table. If you have multiple +PERSON` to see the content you just added in the table. If you have multiple tables in the program, you can repeat this process until you filled all tables with content. diff --git a/docs/tutorials/rest_api/index.md b/docs/tutorials/rest_api/index.md index e05dc6deadc..4ccc125c571 100644 --- a/docs/tutorials/rest_api/index.md +++ b/docs/tutorials/rest_api/index.md @@ -143,7 +143,7 @@ CREATE VIEW low_price \n FROM price \n GROUP BY price.part; \n \n -CREATE VIEW preferred_vendor \n +CREATE MATERIALIZED VIEW preferred_vendor \n (part_id, part_name, vendor_id, vendor_name, price) \n AS \n SELECT \n diff --git a/docs/tutorials/writing-sql.md b/docs/tutorials/writing-sql.md index 1f63cf66cf0..f1fe0112d7a 100644 --- a/docs/tutorials/writing-sql.md +++ b/docs/tutorials/writing-sql.md @@ -12,7 +12,7 @@ CREATE TABLE Person age INT, present BOOLEAN ); -CREATE VIEW Adult AS SELECT Person.name, Person.age FROM Person WHERE Person.age > 18; +CREATE MATERIALIZED VIEW Adult AS SELECT Person.name, Person.age FROM Person WHERE Person.age > 18; ``` Statements need to be separated by semicolons. @@ -27,7 +27,7 @@ formed by a query on other tables or views. For example, the following query defines a view: ```sql -CREATE VIEW Adult AS SELECT Person.name FROM Person WHERE Person.age > 18 +CREATE MATERIALIZED VIEW Adult AS SELECT Person.name FROM Person WHERE Person.age > 18 ``` In order to interpret this query the compiler needs to have been given From 8fdfa92ee581bdea63dd090758e24422b4cf617c Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Wed, 26 Jun 2024 17:25:43 -0700 Subject: [PATCH 2/7] [SQL] support MATERIALIZED views and tables. Signed-off-by: Mihai Budiu --- CHANGELOG.md | 3 + docs/sql/grammar.md | 175 ++++++++++++------ docs/sql/types.md | 4 +- .../SQL-compiler/src/main/codegen/config.fmpp | 1 + .../src/main/codegen/includes/ddl.ftl | 7 +- .../operator/DBSPDelayOutputOperator.java | 2 +- .../circuit/operator/DBSPSinkOperator.java | 4 +- .../operator/DBSPSourceBaseOperator.java | 24 ++- .../operator/DBSPSourceMapOperator.java | 27 +-- .../operator/DBSPSourceMultisetOperator.java | 25 +-- .../operator/DBSPSourceTableOperator.java | 25 +-- .../operator/DBSPViewBaseOperator.java | 8 +- .../circuit/operator/DBSPViewOperator.java | 9 +- ...tTableMetadata.java => TableMetadata.java} | 6 +- .../sqlCompiler/compiler/ViewMetadata.java | 25 +++ .../compiler/backend/rust/ToRustVisitor.java | 27 ++- .../frontend/CalciteToDBSPCompiler.java | 26 +-- .../calciteCompiler/AvroSchemaWrapper.java | 2 +- .../calciteCompiler/CalciteCompiler.java | 2 +- .../frontend/parser/SqlCreateLocalView.java | 26 ++- .../statements/CreateRelationStatement.java | 15 +- .../statements/CreateTableStatement.java | 21 ++- .../statements/CreateViewStatement.java | 18 +- .../frontend/statements/HasSchema.java | 2 +- .../frontend/statements/IHasSchema.java | 6 +- .../visitors/outer/CircuitRewriter.java | 6 +- .../visitors/outer/IncrementalizeVisitor.java | 6 +- .../visitors/outer/IndexedInputs.java | 3 +- .../compiler/visitors/outer/Monotonicity.java | 2 +- .../compiler/sql/CatalogTests.java | 32 +++- .../sqlCompiler/compiler/sql/OtherTests.java | 13 +- .../sqlCompiler/compiler/sql/ParserTests.java | 8 +- .../compiler/sql/simple/RegresssionTests.java | 24 +++ 33 files changed, 397 insertions(+), 187 deletions(-) rename sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/{InputTableMetadata.java => TableMetadata.java} (79%) create mode 100644 sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/ViewMetadata.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 0a3a15190b1..751ec6383ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +- [SQL] Added `MATERIALIZED` views + ([#1959](https://github.com/feldera/feldera/pull/1959)) + ## [0.19.0] - 2024-06-25 - [SQL] Preliminary support for MAP-typed values diff --git a/docs/sql/grammar.md b/docs/sql/grammar.md index afac02a21a6..96ab65d31b1 100644 --- a/docs/sql/grammar.md +++ b/docs/sql/grammar.md @@ -8,7 +8,7 @@ form. - Uppercase words (`FUNCTION`) and single-quoted text (`')'`) indicate grammar terminals. - Parentheses `()` are used for grouping productions together. -- The vertical bar `|` indicates alternation. +- The vertical bar `|` indicates choice between two constructs. ``` statementList: @@ -21,38 +21,33 @@ statement | createTypeStatement | latenessStatement -generalType - : type [NOT NULL] +columnDecl + : column generalType +``` -createFunctionStatement - : CREATE FUNCTION name '(' [ columnDecl [, columnDecl ]* ] ')' RETURNS generalType +## Creating user-defined types -latenessStatement - : LATENESS view '.' column expression - -createTableStatement - : CREATE TABLE name - '(' tableElement [, tableElement ]* ')' - [ 'WITH' keyValueList ] +``` +generalType + : type [NOT NULL] createTypeStatement : CREATE TYPE name AS '(' typedef ')' -keyValueList - : '(' keyValue ( ',' keyValue )* ')' - -keyValue - : stringLiteral '=' stringLiteral - typedef : generalType | name generalType [, name type ]* +``` -createViewStatement - : CREATE [ LOCAL ] VIEW name - [ '(' columnName [, columnName ]* ')' ] +See [user-defined structures](types.md#user-defined-structures) + +## Creating tables + +``` +createTableStatement + : CREATE TABLE name + '(' tableElement [, tableElement ]* ')' [ 'WITH' keyValueList ] - AS query tableElement : columnName generalType ( columnConstraint )* @@ -66,9 +61,6 @@ columnConstraint | WATERMARK expression | DEFAULT expression -parensColumnList - : '(' columnName [, columnName ]* ')' - tableConstraint : [ CONSTRAINT name ] { @@ -77,6 +69,95 @@ tableConstraint } | FOREIGN KEY parensColumnList REFERENCES identifier parensColumnList +parensColumnList + : '(' columnName [, columnName ]* ')' + +keyValueList + : '(' keyValue ( ',' keyValue )* ')' + +keyValue + : stringLiteral '=' stringLiteral +``` + +Note: `FOREIGN KEY` information is parsed, but it is not validated, +and is currently ignored. + +`CREATE TABLE` is used to declare tables. Tables correspond to input +data sources. A table declaration must list the table columns and +their types. Here is an example: + +```sql +CREATE TABLE empsalary ( + depname varchar, + empno bigint, + salary int, + enroll_date date +); +``` + +A table declaration can have an optional `WITH` clause which is used +to specify properties of the connector that provides the source data. +The properties are specified as key-value pairs, each written as a +string. Here is an example: + +```sql +CREATE TABLE empsalary ( + depname varchar, + empno bigint, + salary int, + enroll_date date +) WITH ( + 'source' = 'kafka', + 'url' = 'localhost:8080', + 'materialized' = 'false' +); +``` + +Unlike a database, Feldera does normally not maintain the contents of +tables; it will only store as much data as necessary to compute future +outputs. By specifying the property `'materialized' = 'true'` a user +instructs Feldera to also maintain the complete contents of a table. +The contents of the table can be queried using the `http`-based API +described elsewhere. + +### LATENESS + +``` +latenessStatement + : LATENESS view '.' column expression +``` + +See [Streaming SQL Extensions](streaming.md#lateness-expressions) + +### WATERMARKS + +See [Streaming SQL Extensions](streaming.md#watermark-expressions) + +## Creating user-defined functions. + +`CREATE FUNCTION` is used to declare [user-defined functions](udf.md). + +``` +createFunctionStatement + : CREATE FUNCTION name '(' [ columnDecl [, columnDecl ]* ] ')' RETURNS generalType + [ AS expression ] +``` + +## Creating views + +`CREATE VIEW` is used to declare a view. The optional `LOCAL` +keyword can be used to indicate that the declared view is not exposed +to the outside world as an output of the computation. This is useful +for modularizing the SQL code, by declaring intermediate views that +are used in the implementation of other views. + +``` +createViewStatement + : CREATE [ LOCAL | MATERIALIZED ] VIEW name + [ '(' columnName [, columnName ]* ')' ] + [ 'WITH' keyValueList ] + AS query + query : values | WITH withItem [ , withItem ]* query @@ -92,6 +173,11 @@ query [ LIMIT { count | ALL } ] +withItem + : name + [ '(' column [, column ]* ')' ] + AS '(' query ')' + values : { VALUES | VALUE } expression [, expression ]* @@ -104,8 +190,7 @@ select [ HAVING booleanExpression ] tablePrimary - : [ [ catalogName . ] schemaName . ] tableName - '(' TABLE [ [ catalogName . ] schemaName . ] tableName ')' + : tableName '(' TABLE tableName ')' | tablePrimary '(' columnDecl [, columnDecl ]* ')' | UNNEST '(' expression ')' [ WITH ORDINALITY ] | TABLE '(' functionName '(' expression [, expression ]* ')' ')' @@ -117,18 +202,10 @@ groupItem: | CUBE '(' expression [, expression ]* ')' | ROLLUP '(' expression [, expression ]* ')' -columnDecl - : column generalType - selectWithoutFrom : SELECT [ ALL | DISTINCT ] { * | projectItem [, projectItem ]* } -withItem - : name - [ '(' column [, column ]* ')' ] - AS '(' query ')' - orderItem : expression [ ASC | DESC ] [ NULLS FIRST | NULLS LAST ] @@ -175,20 +252,14 @@ exprOrList | '(' expr [, expr ]* ')' ``` -Note: `FOREIGN KEY` information is parsed, but it is not validated, -and is currently ignored. - In `orderItem`, if expression is a positive integer n, it denotes the nth item in the `SELECT` clause. -SQL `CREATE FUNCTION` can be used to declare [user-defined -functions](udf.md). +If a view is marked as `MATERIALIZED`, the implementation will +maintain a full copy of the view's output in addition to producing the +expected changes. -SQL `CREATE VIEW` is used to declare a view. The optional `LOCAL` -keyword can be used to indicate that the declared view is not exposed -to the outside world as an output of the computation. This is useful -for modularizing the SQL code, by declaring intermediate views that -are used in the implementation of other views. +### Aggregate queries An aggregate query is a query that contains a `GROUP BY` or a `HAVING` clause, or aggregate functions in the `SELECT` clause. In the @@ -201,6 +272,8 @@ aggregate query, and only in a `SELECT`, `HAVING` or `ORDER BY` clause. Aggregate functions are described in [this section](aggregates.md#standard-aggregate-operations). +## Sub-queries + A scalar sub-query is a sub-query used as an expression. If the sub-query returns no rows, the value is `NULL`; if it returns more than one row, it is an error. @@ -267,7 +340,7 @@ group by cube(deptno, job); +--------+-----------+----+---+---+---+ ``` -### Window aggregates +## Window aggregates One type of expression that can appear in a `SELECT` statement is a window aggregate. The grammar for window aggregates is: @@ -301,15 +374,7 @@ on aggregation](aggregates.md#window-aggregate-functions). Currently we require window ranges to have constant values. This precludes ranges such as `INTERVAL 1 YEAR`, which have variable sizes. -### LATENESS - -See [Streaming SQL Extensions](streaming.md#lateness-expressions) - -### WATERMARKS - -See [Streaming SQL Extensions](streaming.md#watermark-expressions) - -### Table functions +## Table functions Table functions are invoked using the syntax `TABLE(function(arguments))`. diff --git a/docs/sql/types.md b/docs/sql/types.md index e64e82dd4a3..796934ddd2a 100644 --- a/docs/sql/types.md +++ b/docs/sql/types.md @@ -1,4 +1,4 @@ -# Supported Data Types +# Built-in Data Types The compiler supports the following SQL data types: @@ -56,7 +56,7 @@ value. ## User-defined structures Users can declare new structure types. Such types can be used for -columns, record fields, user-defined function parameters or result. +columns, record fields, user-defined function parameters or results. For example, we can declare types `address_typ` and `employee_typ`: diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/codegen/config.fmpp b/sql-to-dbsp-compiler/SQL-compiler/src/main/codegen/config.fmpp index c46d6838f0c..7dfd644da5d 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/codegen/config.fmpp +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/codegen/config.fmpp @@ -28,6 +28,7 @@ data: { "DISCARD" "IF" "LATENESS" + "MATERIALIZED" "WATERMARK" "PLANS" "REMOVE" diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/codegen/includes/ddl.ftl b/sql-to-dbsp-compiler/SQL-compiler/src/main/codegen/includes/ddl.ftl index aa7495c8b43..12b38d53dde 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/codegen/includes/ddl.ftl +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/codegen/includes/ddl.ftl @@ -299,16 +299,17 @@ SqlCreate SqlCreateView(Span s, boolean replace) : final SqlIdentifier id; SqlNodeList columnList = null; final SqlNode query; - boolean local = false; + SqlCreateLocalView.ViewKind kind = SqlCreateLocalView.ViewKind.STANDARD; SqlNodeList connector = null; } { - [ { local = true; } ] + [ { kind = SqlCreateLocalView.ViewKind.LOCAL; } + | { kind = SqlCreateLocalView.ViewKind.MATERIALIZED; } ] id = CompoundIdentifier() [ columnList = ParenthesizedSimpleIdentifierList() ] [ connector = KeyValueList() ] query = OrderedQueryOrExpr(ExprContext.ACCEPT_QUERY) { - return new SqlCreateLocalView(s.end(this), replace, local, id, columnList, connector, query); + return new SqlCreateLocalView(s.end(this), replace, kind, id, columnList, connector, query); } } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPDelayOutputOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPDelayOutputOperator.java index 3055ec2f74a..0e3d94d354a 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPDelayOutputOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPDelayOutputOperator.java @@ -22,7 +22,7 @@ public final class DBSPDelayOutputOperator extends DBSPSourceBaseOperator { public DBSPDelayOutputOperator(CalciteObject node, DBSPType outputType, boolean isMultiset, @Nullable String comment) { - super(node, outputType, isMultiset, comment, new NameGen("delay").nextName()); + super(node, outputType, isMultiset, new NameGen("delay").nextName(), comment); } @Override diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSinkOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSinkOperator.java index bc49de7e8b2..560b2768085 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSinkOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSinkOperator.java @@ -23,7 +23,7 @@ package org.dbsp.sqlCompiler.circuit.operator; -import org.dbsp.sqlCompiler.compiler.ViewColumnMetadata; +import org.dbsp.sqlCompiler.compiler.ViewMetadata; import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject; import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision; import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor; @@ -35,7 +35,7 @@ public final class DBSPSinkOperator extends DBSPViewBaseOperator { public DBSPSinkOperator(CalciteObject node, String viewName, String query, DBSPTypeStruct originalRowType, - List metadata, + ViewMetadata metadata, DBSPOperator input) { super(node, "inspect", null, viewName, query, originalRowType, metadata, input); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceBaseOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceBaseOperator.java index c37cfb0c2b2..93e547aa9a4 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceBaseOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceBaseOperator.java @@ -33,19 +33,17 @@ public abstract class DBSPSourceBaseOperator extends DBSPOperator { public final String tableName; - /** - * Create a DBSP operator that is a source to the dataflow graph. - * @param node Calcite node for the statement creating the table - * that this node is created from. - * @param isMultiset True if the source data can be a multiset. - * @param outputType Type of table. - * @param comment A comment describing the operator. - * @param tableName The name of the table that this operator is created from. - */ - public DBSPSourceBaseOperator( - CalciteObject node, - DBSPType outputType, boolean isMultiset, @Nullable String comment, - String tableName) { + /** Create a DBSP operator that is a source to the dataflow graph. + * + * @param node Calcite node for the statement creating the table + * that this node is created from. + * @param outputType Type of table. + * @param isMultiset True if the source data can be a multiset. + * @param tableName The name of the table that this operator is created from. + * @param comment A comment describing the operator. */ + protected DBSPSourceBaseOperator( + CalciteObject node, DBSPType outputType, boolean isMultiset, + String tableName, @Nullable String comment) { super(node, "source " + tableName, null, outputType, isMultiset, comment); this.tableName = tableName; } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceMapOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceMapOperator.java index 92344bf7921..cb8d978ec36 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceMapOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceMapOperator.java @@ -1,6 +1,6 @@ package org.dbsp.sqlCompiler.circuit.operator; -import org.dbsp.sqlCompiler.compiler.InputTableMetadata; +import org.dbsp.sqlCompiler.compiler.TableMetadata; import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject; import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision; import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor; @@ -24,19 +24,20 @@ public final class DBSPSourceMapOperator extends DBSPSourceTableOperator { * Create a DBSP operator that is a source to the dataflow graph. * The table has a primary key, so the data forms a set. * The data is represented as an indexed zset, hence the name "MapOperator". - * @param node Calcite node for the statement creating the table - * that this node is created from. - * @param sourceName Calcite node for the identifier naming the table. - * @param outputType Type of output produced. - * @param keyFields Fields of the input row which compose the key. - * @param comment A comment describing the operator. - * @param name The name of the table that this operator is created from. + * + * @param node Calcite node for the statement creating the table + * that this node is created from. + * @param sourceName Calcite node for the identifier naming the table. + * @param keyFields Fields of the input row which compose the key. + * @param outputType Type of output produced. + * @param name The name of the table that this operator is created from. + * @param comment A comment describing the operator. */ public DBSPSourceMapOperator( CalciteObject node, CalciteObject sourceName, List keyFields, - DBSPTypeIndexedZSet outputType, DBSPTypeStruct originalRowType, @Nullable String comment, - InputTableMetadata metadata, String name) { - super(node, sourceName, outputType, originalRowType, false, comment, metadata, name); + DBSPTypeIndexedZSet outputType, DBSPTypeStruct originalRowType, + TableMetadata metadata, String name, @Nullable String comment) { + super(node, sourceName, outputType, originalRowType, false, metadata, name, comment); this.keyFields = keyFields; } @@ -53,7 +54,7 @@ public void accept(CircuitVisitor visitor) { public DBSPOperator withFunction(@Nullable DBSPExpression unused, DBSPType outputType) { return new DBSPSourceMapOperator(this.getNode(), this.sourceName, this.keyFields, outputType.to(DBSPTypeIndexedZSet.class), this.originalRowType, - this.comment, this.metadata, this.tableName); + this.metadata, this.tableName, this.comment); } @Override @@ -61,7 +62,7 @@ public DBSPOperator withInputs(List newInputs, boolean force) { if (force || this.inputsDiffer(newInputs)) return new DBSPSourceMapOperator(this.getNode(), this.sourceName, this.keyFields, this.getOutputIndexedZSetType(), this.originalRowType, - this.comment, this.metadata, this.tableName); + this.metadata, this.tableName, this.comment); return this; } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceMultisetOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceMultisetOperator.java index 75faa9c1c91..7a696541b57 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceMultisetOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceMultisetOperator.java @@ -3,7 +3,7 @@ import org.dbsp.sqlCompiler.compiler.IHasColumnsMetadata; import org.dbsp.sqlCompiler.compiler.IHasLateness; import org.dbsp.sqlCompiler.compiler.IHasWatermark; -import org.dbsp.sqlCompiler.compiler.InputTableMetadata; +import org.dbsp.sqlCompiler.compiler.TableMetadata; import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject; import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision; import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor; @@ -23,18 +23,19 @@ public final class DBSPSourceMultisetOperator /** * Create a DBSP operator that is a source to the dataflow graph. * The table has *no* primary key, so the data can form a multiset. - * @param node Calcite node for the statement creating the table - * that this node is created from. - * @param sourceName Calcite node for the identifier naming the table. - * @param outputType Type of table. - * @param comment A comment describing the operator. - * @param name The name of the table that this operator is created from. + * + * @param node Calcite node for the statement creating the table + * that this node is created from. + * @param sourceName Calcite node for the identifier naming the table. + * @param outputType Type of table. + * @param name The name of the table that this operator is created from. + * @param comment A comment describing the operator. */ public DBSPSourceMultisetOperator( CalciteObject node, CalciteObject sourceName, - DBSPTypeZSet outputType, DBSPTypeStruct originalRowType, @Nullable String comment, - InputTableMetadata metadata, String name) { - super(node, sourceName, outputType, originalRowType, true, comment, metadata, name); + DBSPTypeZSet outputType, DBSPTypeStruct originalRowType, + TableMetadata metadata, String name, @Nullable String comment) { + super(node, sourceName, outputType, originalRowType, true, metadata, name, comment); assert metadata.getColumnCount() == originalRowType.fields.size(); assert metadata.getColumnCount() == outputType.elementType.to(DBSPTypeTuple.class).size(); } @@ -52,7 +53,7 @@ public void accept(CircuitVisitor visitor) { public DBSPOperator withFunction(@Nullable DBSPExpression unused, DBSPType outputType) { return new DBSPSourceMultisetOperator(this.getNode(), this.sourceName, outputType.to(DBSPTypeZSet.class), this.originalRowType, - this.comment, this.metadata, this.tableName); + this.metadata, this.tableName, this.comment); } @Override @@ -60,7 +61,7 @@ public DBSPOperator withInputs(List newInputs, boolean force) { if (force || this.inputsDiffer(newInputs)) return new DBSPSourceMultisetOperator( this.getNode(), this.sourceName, this.getOutputZSetType(), this.originalRowType, - this.comment, this.metadata, this.tableName); + this.metadata, this.tableName, this.comment); return this; } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceTableOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceTableOperator.java index d7806858416..381c8506bb9 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceTableOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPSourceTableOperator.java @@ -1,6 +1,6 @@ package org.dbsp.sqlCompiler.circuit.operator; -import org.dbsp.sqlCompiler.compiler.InputTableMetadata; +import org.dbsp.sqlCompiler.compiler.TableMetadata; import org.dbsp.sqlCompiler.compiler.errors.SourcePositionRange; import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject; import org.dbsp.sqlCompiler.ir.type.DBSPType; @@ -15,23 +15,24 @@ public abstract class DBSPSourceTableOperator extends DBSPSourceBaseOperator { public final CalciteObject sourceName; // Note: the metadata is not transformed after being set. // In particular, types are not rewritten. - public final InputTableMetadata metadata; + public final TableMetadata metadata; /** * Create a DBSP operator that is a source to the dataflow graph. - * @param node Calcite node for the statement creating the table - * that this node is created from. - * @param sourceName Calcite node for the identifier naming the table. - * @param outputType Type of table. - * @param isMultiset True if the source can produce multiset values. - * @param comment A comment describing the operator. - * @param name The name of the table that this operator is created from. + * + * @param node Calcite node for the statement creating the table + * that this node is created from. + * @param sourceName Calcite node for the identifier naming the table. + * @param outputType Type of table. + * @param isMultiset True if the source can produce multiset values. + * @param name The name of the table that this operator is created from. + * @param comment A comment describing the operator. */ - public DBSPSourceTableOperator( + protected DBSPSourceTableOperator( CalciteObject node, CalciteObject sourceName, DBSPType outputType, DBSPTypeStruct originalRowType, boolean isMultiset, - @Nullable String comment, InputTableMetadata metadata, String name) { - super(node, outputType, isMultiset, comment, name); + TableMetadata metadata, String name, @Nullable String comment) { + super(node, outputType, isMultiset, name, comment); this.originalRowType = originalRowType; this.sourceName = sourceName; this.metadata = metadata; diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPViewBaseOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPViewBaseOperator.java index f6b5606f0d7..ccbc26fb2bb 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPViewBaseOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPViewBaseOperator.java @@ -1,13 +1,12 @@ package org.dbsp.sqlCompiler.circuit.operator; -import org.dbsp.sqlCompiler.compiler.ViewColumnMetadata; +import org.dbsp.sqlCompiler.compiler.ViewMetadata; import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject; import org.dbsp.sqlCompiler.ir.expression.DBSPExpression; import org.dbsp.sqlCompiler.ir.type.DBSPTypeStruct; import org.dbsp.util.IIndentStream; import javax.annotation.Nullable; -import java.util.List; /** Base class for an operator representing a view declared by the user. * If the view is an output then it is represented by a Sink operator. @@ -16,13 +15,12 @@ public abstract class DBSPViewBaseOperator extends DBSPUnaryOperator { public final String viewName; public final String query; public final DBSPTypeStruct originalRowType; - public final List metadata; + public final ViewMetadata metadata; protected DBSPViewBaseOperator( CalciteObject node, String operation, @Nullable DBSPExpression function, String viewName, String query, DBSPTypeStruct originalRowType, - List metadata, - DBSPOperator input) { + ViewMetadata metadata, DBSPOperator input) { super(node, operation, function, input.outputType, input.isMultiset, input); this.metadata = metadata; this.query = query; diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPViewOperator.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPViewOperator.java index 903d7eca906..f6dd3a21412 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPViewOperator.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/circuit/operator/DBSPViewOperator.java @@ -3,7 +3,7 @@ import org.dbsp.sqlCompiler.compiler.IHasColumnsMetadata; import org.dbsp.sqlCompiler.compiler.IHasLateness; import org.dbsp.sqlCompiler.compiler.IHasWatermark; -import org.dbsp.sqlCompiler.compiler.ViewColumnMetadata; +import org.dbsp.sqlCompiler.compiler.ViewMetadata; import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject; import org.dbsp.sqlCompiler.compiler.visitors.VisitDecision; import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor; @@ -25,8 +25,7 @@ public final class DBSPViewOperator public DBSPViewOperator( CalciteObject node, String viewName, String query, DBSPTypeStruct originalRowType, - List metadata, - DBSPOperator input) { + ViewMetadata metadata, DBSPOperator input) { super(node, "map", DBSPClosureExpression.id(), viewName, query, originalRowType, metadata, input); assert metadata.size() == originalRowType.fields.size(); @@ -34,7 +33,7 @@ public DBSPViewOperator( /** True if any column has LATENESS information */ public boolean hasLateness() { - return Linq.any(this.metadata, m -> m.lateness != null); + return this.metadata.hasLateness(); } @Override @@ -63,7 +62,7 @@ public DBSPOperator withInputs(List newInputs, boolean force) { @Override public Iterable getLateness() { - return this.metadata; + return this.metadata.columns; } @Override diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/InputTableMetadata.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/TableMetadata.java similarity index 79% rename from sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/InputTableMetadata.java rename to sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/TableMetadata.java index 2c61ded42d8..0a383fe3f03 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/InputTableMetadata.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/TableMetadata.java @@ -7,11 +7,13 @@ import java.util.List; /** Metadata describing an input table. */ -public class InputTableMetadata { +public class TableMetadata { final LinkedHashMap columnMetadata; + public final boolean materialized; - public InputTableMetadata(List columns) { + public TableMetadata(List columns, boolean materialized) { this.columnMetadata = new LinkedHashMap<>(); + this.materialized = materialized; for (InputColumnMetadata meta: columns) { Utilities.putNew(this.columnMetadata, meta.name, meta); } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/ViewMetadata.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/ViewMetadata.java new file mode 100644 index 00000000000..8f4c0bdca5a --- /dev/null +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/ViewMetadata.java @@ -0,0 +1,25 @@ +package org.dbsp.sqlCompiler.compiler; + +import org.dbsp.sqlCompiler.compiler.frontend.parser.SqlCreateLocalView; +import org.dbsp.util.Linq; + +import java.util.List; + +public class ViewMetadata { + public final List columns; + public final SqlCreateLocalView.ViewKind viewKind; + + public ViewMetadata(List columns, SqlCreateLocalView.ViewKind viewKind) { + this.columns = columns; + this.viewKind = viewKind; + } + + public int size() { + return this.columns.size(); + } + + /** True if any column has LATENESS information */ + public boolean hasLateness() { + return Linq.any(this.columns, m -> m.lateness != null); + } +} diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java index 1d9ce222c5f..a07906e0b36 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java @@ -47,7 +47,7 @@ import org.dbsp.sqlCompiler.compiler.CompilerOptions; import org.dbsp.sqlCompiler.compiler.IErrorReporter; import org.dbsp.sqlCompiler.compiler.InputColumnMetadata; -import org.dbsp.sqlCompiler.compiler.InputTableMetadata; +import org.dbsp.sqlCompiler.compiler.TableMetadata; import org.dbsp.sqlCompiler.compiler.ProgramMetadata; import org.dbsp.sqlCompiler.compiler.errors.InternalCompilerError; import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject; @@ -228,7 +228,7 @@ protected void generateFromTrait(DBSPTypeStruct type) { * @param metadata Metadata for the input columns (null for an output view). */ protected void generateRenameMacro(DBSPTypeStruct type, - @Nullable InputTableMetadata metadata) { + @Nullable TableMetadata metadata) { this.builder.append("deserialize_table_record!("); this.builder.append(type.sanitizedName) .append("[") @@ -468,7 +468,7 @@ void generateStructDeclarations(DBSPTypeStruct struct) { item.accept(this.innerVisitor); } - void generateStructHelpers(DBSPTypeStruct type, @Nullable InputTableMetadata metadata) { + void generateStructHelpers(DBSPTypeStruct type, @Nullable TableMetadata metadata) { List nested = new ArrayList<>(); findNestedStructs(type, nested); for (DBSPTypeStruct s: nested) { @@ -499,7 +499,11 @@ public VisitDecision preorder(DBSPSourceMultisetOperator operator) { zsetType.elementType.accept(this.innerVisitor); this.builder.append(">();").newline(); if (!this.useHandles) { - this.builder.append("catalog.register_input_zset::<_, "); + String registerFunction = operator.metadata.materialized ? + "register_materialized_input_zset" : "register_input_zset"; + this.builder.append("catalog.") + .append(registerFunction) + .append("::<_, "); IHasSchema tableDescription = this.metadata.getTableDescription(operator.tableName); DBSPStrLiteral json = new DBSPStrLiteral(tableDescription.asJson().toString(), false, true); operator.originalRowType.accept(this.innerVisitor); @@ -581,7 +585,11 @@ public VisitDecision preorder(DBSPSourceMapOperator operator) { if (!this.useHandles) { IHasSchema tableDescription = this.metadata.getTableDescription(operator.tableName); DBSPStrLiteral json = new DBSPStrLiteral(tableDescription.asJson().toString(), false, true); - this.builder.append("catalog.register_input_map::<"); + String registerFunction = operator.metadata.materialized ? + "register_materialized_input_map" : "register_input_map"; + this.builder.append("catalog.") + .append(registerFunction) + .append("::<"); keyStructType.toTuple().accept(this.innerVisitor); this.builder.append(", "); keyStructType.accept(this.innerVisitor); @@ -723,7 +731,14 @@ public VisitDecision preorder(DBSPSinkOperator operator) { if (!this.useHandles) { IHasSchema description = this.metadata.getViewDescription(operator.viewName); DBSPStrLiteral json = new DBSPStrLiteral(description.asJson().toString(), false, true); - this.builder.append("catalog.register_output_zset::<_, "); + String registerFunction = switch (operator.metadata.viewKind) { + case MATERIALIZED -> "register_materialized_output_zset"; + case LOCAL -> throw new InternalCompilerError("Sink operator for local view " + operator); + case STANDARD -> "register_output_zset"; + }; + this.builder.append("catalog.") + .append(registerFunction) + .append("::<_, "); operator.originalRowType.accept(this.innerVisitor); this.builder.append(">(") .append(operator.input().getOutputName()) diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java index a8223335f93..fe76f5a9183 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java @@ -97,15 +97,17 @@ import org.dbsp.sqlCompiler.compiler.DBSPCompiler; import org.dbsp.sqlCompiler.compiler.ICompilerComponent; import org.dbsp.sqlCompiler.compiler.InputColumnMetadata; -import org.dbsp.sqlCompiler.compiler.InputTableMetadata; +import org.dbsp.sqlCompiler.compiler.TableMetadata; import org.dbsp.sqlCompiler.compiler.ProgramMetadata; import org.dbsp.sqlCompiler.compiler.ViewColumnMetadata; +import org.dbsp.sqlCompiler.compiler.ViewMetadata; import org.dbsp.sqlCompiler.compiler.errors.InternalCompilerError; import org.dbsp.sqlCompiler.compiler.errors.UnimplementedException; import org.dbsp.sqlCompiler.compiler.errors.UnsupportedException; import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.CalciteCompiler; import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.RelColumnMetadata; import org.dbsp.sqlCompiler.compiler.frontend.calciteObject.CalciteObject; +import org.dbsp.sqlCompiler.compiler.frontend.parser.SqlCreateLocalView; import org.dbsp.sqlCompiler.compiler.frontend.parser.SqlCreateTable; import org.dbsp.sqlCompiler.compiler.frontend.statements.CreateFunctionStatement; import org.dbsp.sqlCompiler.compiler.frontend.statements.CreateTableStatement; @@ -749,12 +751,12 @@ void visitScan(TableScan scan, boolean create) { DBSPType rowType = originalRowType.toTuple(); HasSchema withSchema = new HasSchema(CalciteObject.EMPTY, tableName, false, tableRowType); this.metadata.addTable(withSchema); - InputTableMetadata tableMeta = new InputTableMetadata( - Linq.map(withSchema.getColumns(), this::convertMetadata)); + TableMetadata tableMeta = new TableMetadata( + Linq.map(withSchema.getColumns(), this::convertMetadata), false); source = new DBSPSourceMultisetOperator( node, CalciteObject.EMPTY, this.makeZSet(rowType), originalRowType, - null, tableMeta, tableName); + tableMeta, tableName, null); this.circuit.addOperator(source); Utilities.putNew(this.nodeOperator, scan, source); } @@ -2120,23 +2122,24 @@ DBSPNode compileCreateView(CreateViewStatement view) { } } - if (this.generateOutputForNextView && !view.local) { + ViewMetadata meta = new ViewMetadata(additionalMetadata, view.kind); + if (this.generateOutputForNextView && view.kind != SqlCreateLocalView.ViewKind.LOCAL) { this.metadata.addView(view); // Create two operators chained, a ViewOperator and a SinkOperator. DBSPViewOperator vo = new DBSPViewOperator( view.getCalciteObject(), view.relationName, view.statement, - struct, additionalMetadata, op); + struct, meta, op); this.circuit.addOperator(vo); o = new DBSPSinkOperator( view.getCalciteObject(), view.relationName, - view.statement, struct, additionalMetadata, vo); + view.statement, struct, meta, vo); + this.circuit.addOperator(o); } else { // We may already have a node for this output DBSPOperator previous = this.circuit.getView(view.relationName); if (previous != null) return previous; - o = new DBSPViewOperator(view.getCalciteObject(), view.relationName, view.statement, - struct, additionalMetadata, op); + o = new DBSPViewOperator(view.getCalciteObject(), view.relationName, view.statement, struct, meta, op); } this.circuit.addOperator(o); return o; @@ -2232,10 +2235,11 @@ DBSPNode compileCreateTable(CreateTableStatement create) { identifier = CalciteObject.create(sct.name); } List metadata = Linq.map(create.columns, this::convertMetadata); - InputTableMetadata tableMeta = new InputTableMetadata(metadata); + boolean materialized = create.isMaterialized(); + TableMetadata tableMeta = new TableMetadata(metadata, materialized); DBSPSourceMultisetOperator result = new DBSPSourceMultisetOperator( create.getCalciteObject(), identifier, this.makeZSet(rowType), originalRowType, - def.statement, tableMeta, tableName); + tableMeta, tableName, def.statement); this.circuit.addOperator(result); this.metadata.addTable(create); return null; diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/AvroSchemaWrapper.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/AvroSchemaWrapper.java index 4bbe592a39b..17d3de44ace 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/AvroSchemaWrapper.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/AvroSchemaWrapper.java @@ -55,7 +55,7 @@ public List getColumns() { @Nullable @Override - public Map getConnectorProperties() { + public Map getProperties() { return null; } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/CalciteCompiler.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/CalciteCompiler.java index b8db134bcb9..3ec0556203c 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/CalciteCompiler.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/calciteCompiler/CalciteCompiler.java @@ -516,7 +516,7 @@ public RelDataType specToRel(SqlDataTypeSpec spec) { if (result.containsKey(keyString)) { SqlNode prev = Utilities.getExists(previous, keyString); this.errorReporter.reportError(new SourcePositionRange(key.getParserPosition()), - "Duplicate key", "connector property " + Utilities.singleQuote(keyString) + + "Duplicate key", "property " + Utilities.singleQuote(keyString) + " already declared"); this.errorReporter.reportError(new SourcePositionRange(prev.getParserPosition()), "Duplicate key", "Previous declaration"); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/parser/SqlCreateLocalView.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/parser/SqlCreateLocalView.java index 9852a4776fe..fc24fd7253f 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/parser/SqlCreateLocalView.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/parser/SqlCreateLocalView.java @@ -20,8 +20,17 @@ // but the constructor isn't public. // We just need an extra 'local' field. public class SqlCreateLocalView extends SqlCreate { + public enum ViewKind { + /** For materialized views the DBSP program will keep the full contents */ + MATERIALIZED, + /** Local views are not program outputs */ + LOCAL, + /** Standard views only produce deltas */ + STANDARD, + } + public final SqlIdentifier name; - public final boolean isLocal; + public final ViewKind kind; public final @Nullable SqlNodeList columnList; public final SqlNode query; @Nullable public final SqlNodeList connectorProperties; @@ -29,14 +38,14 @@ public class SqlCreateLocalView extends SqlCreate { private static final SqlOperator OPERATOR = new SqlSpecialOperator("CREATE VIEW", SqlKind.CREATE_VIEW); - public SqlCreateLocalView(SqlParserPos pos, boolean replace, boolean local, SqlIdentifier name, + public SqlCreateLocalView(SqlParserPos pos, boolean replace, ViewKind kind, SqlIdentifier name, @Nullable SqlNodeList columnList, @Nullable SqlNodeList connectorProperties, SqlNode query) { super(OPERATOR, pos, replace, false); this.name = Objects.requireNonNull(name, "name"); this.columnList = columnList; // may be null this.query = Objects.requireNonNull(query, "query"); - this.isLocal = local; + this.kind = kind; this.connectorProperties = connectorProperties; } @@ -51,8 +60,15 @@ public SqlCreateLocalView(SqlParserPos pos, boolean replace, boolean local, SqlI } else { writer.keyword("CREATE"); } - if (this.isLocal) { - writer.keyword("LOCAL"); + switch (this.kind) { + case LOCAL: + writer.keyword("LOCAL"); + break; + case MATERIALIZED: + writer.keyword("MATERIALIZED"); + break; + default: + break; } writer.keyword("VIEW"); name.unparse(writer, leftPrec, rightPrec); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateRelationStatement.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateRelationStatement.java index b266b443f9a..ca84f78cf46 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateRelationStatement.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateRelationStatement.java @@ -39,17 +39,17 @@ public abstract class CreateRelationStatement public final String relationName; public final boolean nameIsQuoted; public final List columns; - @Nullable final Map connectorProperties; + @Nullable final Map properties; protected CreateRelationStatement(SqlNode node, String statement, String relationName, boolean nameIsQuoted, List columns, - @Nullable Map connectorProperties) { + @Nullable Map properties) { super(node, statement); this.nameIsQuoted = nameIsQuoted; this.relationName = relationName; this.columns = columns; - this.connectorProperties = connectorProperties; + this.properties = properties; } public AbstractTable getEmulatedTable() { @@ -64,7 +64,14 @@ public List getColumns() { return this.columns; } - @Nullable public Map getConnectorProperties() { return this.connectorProperties; } + @Nullable public Map getProperties() { return this.properties; } + + @Nullable + public String getPropertyValue(String property) { + if (this.properties == null) + return null; + return this.properties.get(property); + } @Override public String toString() { diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateTableStatement.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateTableStatement.java index 455c0226044..8632e0f9fc8 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateTableStatement.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateTableStatement.java @@ -23,6 +23,8 @@ package org.dbsp.sqlCompiler.compiler.frontend.statements; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.calcite.sql.SqlNode; import org.dbsp.sqlCompiler.compiler.frontend.calciteCompiler.RelColumnMetadata; @@ -35,7 +37,22 @@ public class CreateTableStatement extends CreateRelationStatement { public CreateTableStatement(SqlNode node, String statement, String tableName, boolean nameIsQuoted, List columns, - @Nullable Map connectorProperties) { - super(node, statement, tableName, nameIsQuoted, columns, connectorProperties); + @Nullable Map properties) { + super(node, statement, tableName, nameIsQuoted, columns, properties); + } + + public boolean isMaterialized() { + String mat = this.getPropertyValue("materialized"); + if (mat == null) + return false; + return mat.equalsIgnoreCase("true"); + } + + @Override + public JsonNode asJson() { + JsonNode node = super.asJson(); + ObjectNode object = (ObjectNode) node; + object.put("materialized", this.isMaterialized()); + return object; } } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateViewStatement.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateViewStatement.java index 011e70ee987..1716ae55abe 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateViewStatement.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/CreateViewStatement.java @@ -23,6 +23,8 @@ package org.dbsp.sqlCompiler.compiler.frontend.statements; +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.calcite.rel.RelNode; import org.apache.calcite.rel.RelRoot; import org.apache.calcite.sql.SqlNode; @@ -39,15 +41,15 @@ public class CreateViewStatement extends CreateRelationStatement { /** Compiled and optimized query. */ private final RelRoot compiled; public final SqlNode query; - public final boolean local; + public final SqlCreateLocalView.ViewKind kind; public CreateViewStatement(SqlCreateLocalView node, String statement, String tableName, boolean nameIsQuoted, List columns, SqlNode query, RelRoot compiled, - @Nullable Map connectorProperties) { - super(node, statement, tableName, nameIsQuoted, columns, connectorProperties); - this.local = node.isLocal; + @Nullable Map properties) { + super(node, statement, tableName, nameIsQuoted, columns, properties); + this.kind = node.kind; this.query = query; this.compiled = compiled; } @@ -59,4 +61,12 @@ public RelNode getRelNode() { public RelRoot getRoot() { return this.compiled; } + + @Override + public JsonNode asJson() { + JsonNode node = super.asJson(); + ObjectNode object = (ObjectNode) node; + object.put("materialized", kind == SqlCreateLocalView.ViewKind.MATERIALIZED); + return object; + } } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/HasSchema.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/HasSchema.java index 13015ea3a03..404e061d4c4 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/HasSchema.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/HasSchema.java @@ -53,7 +53,7 @@ public List getColumns() { @Nullable @Override - public Map getConnectorProperties() { + public Map getProperties() { return null; } } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/IHasSchema.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/IHasSchema.java index 62ed3d990ee..3fbd4e74e91 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/IHasSchema.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/statements/IHasSchema.java @@ -32,7 +32,7 @@ public interface IHasSchema extends IHasCalciteObject { List getColumns(); /** Properties describing the connector attached to this object */ @Nullable - Map getConnectorProperties(); + Map getProperties(); /** Return the index of the specified column. */ default int getColumnIndex(SqlIdentifier id) { @@ -73,13 +73,13 @@ default JsonNode asJson() { } if (hasKey) result.set("primary_key", keyFields); - Map props = this.getConnectorProperties(); + Map props = this.getProperties(); if (props != null) { ObjectNode properties = mapper.createObjectNode(); for (Map.Entry entry: props.entrySet()) { properties.put(entry.getKey(), entry.getValue()); } - result.set("connector", properties); + result.set("properties", properties); } return result; } 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 eb05f4fec40..8c86d647135 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 @@ -129,8 +129,8 @@ public void postorder(DBSPSourceMultisetOperator operator) { if (!originalRowType.sameType(operator.originalRowType) || !outputType.sameType(operator.outputType)) { result = new DBSPSourceMultisetOperator(operator.getNode(), operator.sourceName, - outputType.to(DBSPTypeZSet.class), originalRowType, operator.comment, - operator.metadata, operator.getTableName()); + outputType.to(DBSPTypeZSet.class), originalRowType, + operator.metadata, operator.getTableName(), operator.comment); } this.map(operator, result); } @@ -144,7 +144,7 @@ public void postorder(DBSPSourceMapOperator operator) { || !outputType.sameType(operator.outputType)) { result = new DBSPSourceMapOperator(operator.getNode(), operator.sourceName, operator.keyFields, outputType.to(DBSPTypeIndexedZSet.class), originalRowType, - operator.comment, operator.metadata, operator.getTableName()); + operator.metadata, operator.getTableName(), operator.comment); } this.map(operator, result); } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/IncrementalizeVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/IncrementalizeVisitor.java index 0981c6ca753..4c47482853c 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/IncrementalizeVisitor.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/IncrementalizeVisitor.java @@ -26,12 +26,10 @@ import org.dbsp.sqlCompiler.circuit.operator.*; import org.dbsp.sqlCompiler.compiler.IErrorReporter; -/** - * This visitor converts a DBSPCircuit into a new circuit which +/** This visitor converts a DBSPCircuit into a new circuit which * computes the incremental version of the same query. * The generated circuit is not efficient, though, it should be - * further optimized. - */ + * further optimized. */ public class IncrementalizeVisitor extends CircuitCloneVisitor { public IncrementalizeVisitor(IErrorReporter reporter) { super(reporter, false); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/IndexedInputs.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/IndexedInputs.java index fde3d158c51..6eff6a9ec60 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/IndexedInputs.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/IndexedInputs.java @@ -44,8 +44,7 @@ public void postorder(DBSPSourceMultisetOperator node) { DBSPTypeIndexedZSet ix = new DBSPTypeIndexedZSet(node.getNode(), keyType, inputType.elementType); DBSPSourceMapOperator set = new DBSPSourceMapOperator( node.getNode(), node.sourceName, keyColumnFields, - ix, node.originalRowType, node.comment, - node.metadata, node.tableName); + ix, node.originalRowType, node.metadata, node.tableName, node.comment); this.addOperator(set); DBSPDeindexOperator deindex = new DBSPDeindexOperator(node.getNode(), set); this.map(node, deindex); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/Monotonicity.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/Monotonicity.java index b760c6d3e8b..1a135bcada2 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/Monotonicity.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/Monotonicity.java @@ -208,7 +208,7 @@ public void postorder(DBSPViewOperator node) { // Trust the annotations, and forget what we know about the input. // This code parallels DBSPSourceMultisetOperator List fields = new ArrayList<>(); - for (ViewColumnMetadata metadata: node.metadata) { + for (ViewColumnMetadata metadata: node.metadata.columns) { IMaybeMonotoneType columnType = new NonMonotoneType(metadata.getType()); if (metadata.lateness != null) columnType = new MonotoneType(metadata.getType()); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/CatalogTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/CatalogTests.java index 7d68c47d465..f385f95e505 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/CatalogTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/CatalogTests.java @@ -272,16 +272,32 @@ public void primaryKeyTest() { } @Test - public void updateTest() { + public void primaryKeyTest2() { String sql = """ create table t1( - id1 bigint not null, - id2 bigint, - str1 varchar not null, - str2 varchar, - int1 bigint not null, - int2 bigint, - primary key(id1, id2))"""; + id1 bigint not null, + id2 bigint, + str1 varchar not null, + str2 varchar, + int1 bigint not null, + int2 bigint, + primary key(id1, id2) + )"""; + DBSPCompiler compiler = testCompiler(); + compiler.compileStatements(sql); + CompilerCircuitStream ccs = new CompilerCircuitStream(compiler); + this.addRustTestCase(sql, ccs); + } + + @Test + public void materializedTest2() { + String sql = """ + create table T( + I int not null + ) with ( + 'materialized' = 'true' + ); + create materialized view V as SELECT * FROM T;"""; DBSPCompiler compiler = testCompiler(); compiler.compileStatements(sql); CompilerCircuitStream ccs = new CompilerCircuitStream(compiler); diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/OtherTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/OtherTests.java index 5188db4b6c8..a62079c7151 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/OtherTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/OtherTests.java @@ -174,7 +174,7 @@ CREATE VIEW V WITH ( JsonNode inputs = meta.get("inputs"); Assert.assertNotNull(inputs); Assert.assertTrue(inputs.isArray()); - JsonNode c = inputs.get(0).get("connector"); + JsonNode c = inputs.get(0).get("properties"); Assert.assertNotNull(c); String str = c.toPrettyString(); Assert.assertEquals(""" @@ -186,7 +186,7 @@ CREATE VIEW V WITH ( JsonNode outputs = meta.get("outputs"); Assert.assertNotNull(inputs); Assert.assertTrue(outputs.isArray()); - c = outputs.get(0).get("connector"); + c = outputs.get(0).get("properties"); Assert.assertNotNull(c); str = c.toPrettyString(); Assert.assertEquals(""" @@ -846,7 +846,8 @@ CREATE TABLE T ( } } } ], - "primary_key" : [ "COL3" ] + "primary_key" : [ "COL3" ], + "materialized" : false } ], "outputs" : [ { "name" : "V", @@ -858,7 +859,8 @@ CREATE TABLE T ( "nullable" : false, "type" : "INTEGER" } - } ] + } ], + "materialized" : false }, { "name" : "V1", "case_sensitive" : false, @@ -869,7 +871,8 @@ CREATE TABLE T ( "nullable" : false, "type" : "INTEGER" } - } ] + } ], + "materialized" : false } ] }""", jsonContents); } diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/ParserTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/ParserTests.java index 91aec7e0a25..1185be79738 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/ParserTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/ParserTests.java @@ -57,6 +57,7 @@ public void ddlTest() throws SqlParseException { ")"; String ddl1 = "CREATE VIEW V AS SELECT * FROM T"; String ddl2 = "CREATE LOCAL VIEW V2 AS SELECT * FROM T"; + String ddl3 = "CREATE MATERIALIZED VIEW V3 AS SELECT * FROM T"; SqlNode node = calcite.parse(ddl); Assert.assertNotNull(node); @@ -73,7 +74,12 @@ public void ddlTest() throws SqlParseException { node = calcite.parse(ddl2); Assert.assertNotNull(node); Assert.assertTrue(node instanceof SqlCreateLocalView); - Assert.assertTrue(((SqlCreateLocalView) node).isLocal); + Assert.assertSame(SqlCreateLocalView.ViewKind.LOCAL, ((SqlCreateLocalView) node).kind); + + node = calcite.parse(ddl3); + Assert.assertNotNull(node); + Assert.assertTrue(node instanceof SqlCreateLocalView); + Assert.assertSame(SqlCreateLocalView.ViewKind.MATERIALIZED, ((SqlCreateLocalView) node).kind); } @Test diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/RegresssionTests.java b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/RegresssionTests.java index 5f1df536e82..bf3fcfa0177 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/RegresssionTests.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/test/java/org/dbsp/sqlCompiler/compiler/sql/simple/RegresssionTests.java @@ -9,6 +9,7 @@ import org.dbsp.sqlCompiler.compiler.sql.SqlIoTest; import org.dbsp.sqlCompiler.compiler.visitors.outer.CircuitVisitor; import org.junit.Assert; +import org.junit.Ignore; import org.junit.Test; public class RegresssionTests extends SqlIoTest { @@ -138,6 +139,29 @@ public void missingCast() { this.compileRustTestCase(sql); } + @Test @Ignore("Calcite decorrelator fails") + public void issue1956() { + String sql = """ + CREATE TABLE auctions ( + id INT PRIMARY KEY, + seller INT, + item TEXT + ); + + CREATE TABLE bids ( + id INT PRIMARY KEY, + buyer INT, + auction_id INT, + amount INT + ); + + CREATE VIEW V AS SELECT id, (SELECT array_agg(buyer) FROM ( + SELECT buyer FROM bids WHERE auction_id = auctions.id + ORDER BY buyer LIMIT 10 + )) FROM auctions;"""; + this.compileRustTestCase(sql); + } + @Test public void issue1957() { String sql = """ From 3ea42e79d627345e5e7ca057f22eba3affe7a1d0 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Thu, 27 Jun 2024 11:45:13 -0700 Subject: [PATCH 3/7] Materialized view-related fixups. - Docs - Demos - Python API Signed-off-by: Leonid Ryzhyk --- Cargo.lock | 1 + crates/adapters/src/format/json/output.rs | 4 ++-- crates/adapters/src/format/parquet/test.rs | 2 +- crates/adapters/src/test/kafka.rs | 2 +- crates/adapters/src/test/mod.rs | 9 +++---- crates/pipeline-types/src/program_schema.rs | 5 +++- .../src/transport/delta_table.rs | 5 +++- crates/pipeline_manager/src/db/test.rs | 14 +++++------ .../pipeline_manager/src/integration_test.rs | 16 ++++++------- demo/project_demo00-SecOps/project.sql | 2 +- .../notebook.ipynb | 4 ++-- .../run.py | 2 +- .../1_feature_pipeline.ipynb | 14 +++++------ docs/sql/grammar.md | 7 +++--- .../fraud_detection/fraud_detection.md | 2 +- openapi.json | 3 +++ python/docs/examples.rst | 8 +++---- python/docs/introduction.rst | 22 ++++++++--------- python/feldera/_sql_view.py | 21 ++++++++++++---- python/feldera/sql_context.py | 19 +++++++++++---- python/tests/test_wireframes.py | 24 +++++++++---------- 21 files changed, 110 insertions(+), 76 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 43351798f3c..d514c1f9650 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8811,6 +8811,7 @@ dependencies = [ "geo", "geo-types", "hex", + "itertools 0.13.0", "lazy_static", "like", "num", diff --git a/crates/adapters/src/format/json/output.rs b/crates/adapters/src/format/json/output.rs index a359779f4ad..2e7313f86a7 100644 --- a/crates/adapters/src/format/json/output.rs +++ b/crates/adapters/src/format/json/output.rs @@ -494,7 +494,7 @@ mod test { let mut encoder = JsonEncoder::new( Box::new(consumer), config, - &Relation::new("TestStruct", false, TestStruct::schema()), + &Relation::new("TestStruct", false, TestStruct::schema(), false), ); let zsets = batches .iter() @@ -671,7 +671,7 @@ mod test { let mut encoder = JsonEncoder::new( Box::new(consumer), config, - &Relation::new("TestStruct", false, TestStruct::schema()), + &Relation::new("TestStruct", false, TestStruct::schema(), false), ); let zset = OrdZSet::from_keys((), test_data()[0].clone()); diff --git a/crates/adapters/src/format/parquet/test.rs b/crates/adapters/src/format/parquet/test.rs index c62b2645bf1..c0a3c9eaea9 100644 --- a/crates/adapters/src/format/parquet/test.rs +++ b/crates/adapters/src/format/parquet/test.rs @@ -115,7 +115,7 @@ fn parquet_output() { let mut encoder = ParquetEncoder::new( Box::new(consumer), config, - Relation::new("TestStruct2", false, TestStruct2::schema()), + Relation::new("TestStruct2", false, TestStruct2::schema(), false), ) .expect("Can't create encoder"); let zset = OrdZSet::from_keys( diff --git a/crates/adapters/src/test/kafka.rs b/crates/adapters/src/test/kafka.rs index 22ae737a515..a8bb48938a5 100644 --- a/crates/adapters/src/test/kafka.rs +++ b/crates/adapters/src/test/kafka.rs @@ -245,7 +245,7 @@ impl BufferConsumer { let buffer = MockDeZSet::new(); // Input parsers don't care about schema yet. - let schema = Relation::new("mock_schema", false, vec![]); + let schema = Relation::new("mock_schema", false, vec![], false); let mut parser = format .new_parser( diff --git a/crates/adapters/src/test/mod.rs b/crates/adapters/src/test/mod.rs index 4d67f30d39d..7274d6ee605 100644 --- a/crates/adapters/src/test/mod.rs +++ b/crates/adapters/src/test/mod.rs @@ -97,7 +97,7 @@ where { let input_handle = >::new(); // Input parsers don't care about schema yet. - let schema = Relation::new("mock_schema", false, vec![]); + let schema = Relation::new("mock_schema", false, vec![], false); let consumer = MockInputConsumer::from_handle( &InputCollectionHandle::new(schema, input_handle.clone()), config, @@ -155,10 +155,11 @@ where let (input, hinput) = circuit.add_input_zset::(); let input_schema = - serde_json::to_string(&Relation::new("test_input1", false, schema.clone())).unwrap(); + serde_json::to_string(&Relation::new("test_input1", false, schema.clone(), false)) + .unwrap(); let output_schema = - serde_json::to_string(&Relation::new("test_output1", false, schema)).unwrap(); + serde_json::to_string(&Relation::new("test_output1", false, schema, false)).unwrap(); catalog.register_materialized_input_zset(input.clone(), hinput, &input_schema); catalog.register_materialized_output_zset(input, &output_schema); @@ -196,7 +197,7 @@ where let buffer = MockDeZSet::::new(); // Input parsers don't care about schema yet. - let schema = Relation::new("mock_schema", false, vec![]); + let schema = Relation::new("mock_schema", false, vec![], false); let mut parser = format .new_parser( diff --git a/crates/pipeline-types/src/program_schema.rs b/crates/pipeline-types/src/program_schema.rs index 8ef8c265288..be4e331aa96 100644 --- a/crates/pipeline-types/src/program_schema.rs +++ b/crates/pipeline-types/src/program_schema.rs @@ -50,14 +50,17 @@ pub struct Relation { pub case_sensitive: bool, #[cfg_attr(feature = "testing", proptest(value = "Vec::new()"))] pub fields: Vec, + #[serde(default)] + pub materialized: bool, } impl Relation { - pub fn new(name: &str, case_sensitive: bool, fields: Vec) -> Self { + pub fn new(name: &str, case_sensitive: bool, fields: Vec, materialized: bool) -> Self { Self { name: name.to_string(), case_sensitive, fields, + materialized, } } diff --git a/crates/pipeline-types/src/transport/delta_table.rs b/crates/pipeline-types/src/transport/delta_table.rs index ceda7a7f04a..125a9e3906e 100644 --- a/crates/pipeline-types/src/transport/delta_table.rs +++ b/crates/pipeline-types/src/transport/delta_table.rs @@ -170,7 +170,10 @@ fn test_delta_reader_config_serde() { let expected = r#"{"uri":"protocol:/path/to/somewhere","timestamp_column":"ts","mode":"follow","snapshot_filter":"ts BETWEEN '2005-01-01 00:00:00' AND '2010-12-31 23:59:59'","version":null,"datetime":"2010-12-31 00:00:00Z","customoption1":"val1","customoption2":"val2"}"#; - assert_eq!(serialized_config, expected); + assert_eq!( + serde_json::from_str::(&serialized_config).unwrap(), + serde_json::from_str::(&expected).unwrap() + ); } impl DeltaTableReaderConfig { diff --git a/crates/pipeline_manager/src/db/test.rs b/crates/pipeline_manager/src/db/test.rs index 52ae9789540..87bcb548880 100644 --- a/crates/pipeline_manager/src/db/test.rs +++ b/crates/pipeline_manager/src/db/test.rs @@ -862,8 +862,8 @@ async fn versioning() { tenant_id, program_id, ProgramSchema { - inputs: vec![Relation::new("t1", false, vec![])], - outputs: vec![Relation::new("v1", false, vec![])], + inputs: vec![Relation::new("t1", false, vec![], false)], + outputs: vec![Relation::new("v1", false, vec![], false)], }, ) .await @@ -941,10 +941,10 @@ async fn versioning() { program_id, ProgramSchema { inputs: vec![ - Relation::new("t1", false, vec![]), - Relation::new("t2", false, vec![]), + Relation::new("t1", false, vec![], false), + Relation::new("t2", false, vec![], false), ], - outputs: vec![Relation::new("v1", false, vec![])], + outputs: vec![Relation::new("v1", false, vec![], false)], }, ) .await @@ -963,8 +963,8 @@ async fn versioning() { tenant_id, program_id, ProgramSchema { - inputs: vec![Relation::new("tnew1", false, vec![])], - outputs: vec![Relation::new("vnew1", false, vec![])], + inputs: vec![Relation::new("tnew1", false, vec![], false)], + outputs: vec![Relation::new("vnew1", false, vec![], false)], }, ) .await diff --git a/crates/pipeline_manager/src/integration_test.rs b/crates/pipeline_manager/src/integration_test.rs index 5b9493ffcaa..38d6c941b16 100644 --- a/crates/pipeline_manager/src/integration_test.rs +++ b/crates/pipeline_manager/src/integration_test.rs @@ -755,7 +755,7 @@ async fn deploy_pipeline() { let config = setup().await; let _ = deploy_pipeline_without_connectors( &config, - "create table t1(c1 integer); create view v1 as select * from t1;", + "create table t1(c1 integer) with ('materialized' = 'true'); create view v1 as select * from t1;", ) .await; @@ -930,7 +930,7 @@ async fn json_ingress() { let config = setup().await; let id = deploy_pipeline_without_connectors( &config, - "create table t1(c1 integer, c2 bool, c3 varchar); create view v1 as select * from t1;", + "create table t1(c1 integer, c2 bool, c3 varchar) with ('materialized' = 'true'); create materialized view v1 as select * from t1;", ) .await; @@ -1101,7 +1101,7 @@ async fn map_column() { let config = setup().await; let id = deploy_pipeline_without_connectors( &config, - "create table t1(c1 integer, c2 bool, c3 MAP); create view v1 as select * from t1;", + "create table t1(c1 integer, c2 bool, c3 MAP) with ('materialized' = 'true'); create view v1 as select * from t1;", ) .await; @@ -1159,7 +1159,7 @@ async fn parse_datetime() { let config = setup().await; let _ = deploy_pipeline_without_connectors( &config, - "create table t1(t TIME, ts TIMESTAMP, d DATE);", + "create table t1(t TIME, ts TIMESTAMP, d DATE) with ('materialized' = 'true');", ) .await; @@ -1208,7 +1208,7 @@ async fn quoted_columns() { let config = setup().await; let _ = deploy_pipeline_without_connectors( &config, - r#"create table t1("c1" integer not null, "C2" bool not null, "😁❤" varchar not null, "αβγ" boolean not null, ΔΘ boolean not null)"#, + r#"create table t1("c1" integer not null, "C2" bool not null, "😁❤" varchar not null, "αβγ" boolean not null, ΔΘ boolean not null) with ('materialized' = 'true')"#, ) .await; @@ -1258,7 +1258,7 @@ async fn primary_keys() { let config = setup().await; let _ = deploy_pipeline_without_connectors( &config, - r#"create table t1(id bigint not null, s varchar not null, primary key (id))"#, + r#"create table t1(id bigint not null, s varchar not null, primary key (id)) with ('materialized' = 'true')"#, ) .await; @@ -1371,8 +1371,8 @@ async fn case_sensitive_tables() { &config, r#"create table "TaBle1"(id bigint not null); create table table1(id bigint); -create view "V1" as select * from "TaBle1"; -create view "v1" as select * from table1;"#, +create materialized view "V1" as select * from "TaBle1"; +create materialized view "v1" as select * from table1;"#, ) .await; diff --git a/demo/project_demo00-SecOps/project.sql b/demo/project_demo00-SecOps/project.sql index 9d26f56b3a0..ab5bc3e59e2 100644 --- a/demo/project_demo00-SecOps/project.sql +++ b/demo/project_demo00-SecOps/project.sql @@ -138,7 +138,7 @@ create view k8scluster_vulnerability ( -- Per-cluster statistics: -- * Number of vulnerabilities. -- * Most severe vulnerability. -create view k8scluster_vulnerability_stats ( +create materialized view k8scluster_vulnerability_stats ( k8scluster_id, k8scluster_name, total_vulnerabilities, diff --git a/demo/project_demo10-FraudDetectionDeltaLake/notebook.ipynb b/demo/project_demo10-FraudDetectionDeltaLake/notebook.ipynb index 4cc48895a0d..cc190990342 100644 --- a/demo/project_demo10-FraudDetectionDeltaLake/notebook.ipynb +++ b/demo/project_demo10-FraudDetectionDeltaLake/notebook.ipynb @@ -186,7 +186,7 @@ " window_30_day AS (PARTITION BY t.cc_num ORDER BY unix_time RANGE BETWEEN 2592000 PRECEDING AND CURRENT ROW);\n", " \"\"\"\n", "\n", - " sql.register_output_view(\"FEATURE\", query)\n", + " sql.register_view(\"FEATURE\", query)\n", " return sql\n" ] }, @@ -390,7 +390,7 @@ "\n", " # eval_metrics(y_inf, predictions_inf)\n", " fraud = [index for index, value in enumerate(predictions_inf) if value != 0]\n", - " \n", + "\n", " GREEN = \"\\033[92m\"\n", " RED = \"\\033[91m\"\n", " RESET = \"\\033[0m\"\n", diff --git a/demo/project_demo10-FraudDetectionDeltaLake/run.py b/demo/project_demo10-FraudDetectionDeltaLake/run.py index 2881af96f70..d2261817ae2 100644 --- a/demo/project_demo10-FraudDetectionDeltaLake/run.py +++ b/demo/project_demo10-FraudDetectionDeltaLake/run.py @@ -279,7 +279,7 @@ def build_program(client, pipeline_name): window_30_day AS (PARTITION BY t.cc_num ORDER BY unix_time RANGE BETWEEN 2592000 PRECEDING AND CURRENT ROW); """ - sql.register_output_view("FEATURE", query) + sql.register_view("FEATURE", query) return sql # Split input dataframe into train and test sets diff --git a/demo/project_demo11-Hopsworks/1_feature_pipeline.ipynb b/demo/project_demo11-Hopsworks/1_feature_pipeline.ipynb index e6b0c82dac9..04feb9644c0 100644 --- a/demo/project_demo11-Hopsworks/1_feature_pipeline.ipynb +++ b/demo/project_demo11-Hopsworks/1_feature_pipeline.ipynb @@ -59,7 +59,7 @@ "# Use Feldera online sandbox\n", "# client = FelderaClient(\"https://try.feldera.com\", api_key = get_secret('FELDERA_API_KEY'))\n", "\n", - "# Use local Feldera instance \n", + "# Use local Feldera instance\n", "client = FelderaClient(\"http://localhost:8080\")\n", "\n", "sql = SQLContext(\"hopsworks_kafka\", client).get_or_create()" @@ -97,7 +97,7 @@ "\n", "# Create feature groups to store Feldera outputs.\n", "\n", - "# COMBINED - features that extend credit card transaction records with attributes extracted from the card \n", + "# COMBINED - features that extend credit card transaction records with attributes extracted from the card\n", "# holder's profile, such as their age at the time of the transaction and the number of days until the credit card expires.\n", "combined_fg = fs.get_or_create_feature_group(\n", " name=KAFKA_OUTPUT_TOPICS[0],\n", @@ -236,7 +236,7 @@ "# Convert credit card expiration date from MM/YY formatted string to a TIMESTAMP,\n", "# so that we can perform computations on it.\n", "sql.register_local_view(\n", - " \"cc_expiration\", \n", + " \"cc_expiration\",\n", " f\"\"\"\n", " SELECT\n", " cc_num,\n", @@ -262,8 +262,8 @@ "\n", "# Compute the age of the individual during the transaction, and the number of days until the\n", "# credit card expires from `PROFILES` and `TRANSACTIONS` tables.\n", - "sql.register_output_view(\n", - " \"combined\", \n", + "sql.register_view(\n", + " \"combined\",\n", " f\"\"\"\n", " SELECT\n", " T1.*,\n", @@ -320,7 +320,7 @@ "\n", "\n", "# Final output view\n", - "sql.register_output_view(\n", + "sql.register_view(\n", " \"windowed\",\n", " \"\"\"\n", " SELECT\n", @@ -588,7 +588,7 @@ " {\"name\": \"days_until_card_expires\", \"description\": \"Card validity days left when the transaction was made\"},\n", "]\n", "\n", - "for desc in feature_descriptions: \n", + "for desc in feature_descriptions:\n", " combined_fg.update_feature_description(desc[\"name\"], desc[\"description\"])" ] } diff --git a/docs/sql/grammar.md b/docs/sql/grammar.md index 96ab65d31b1..4def39a0e98 100644 --- a/docs/sql/grammar.md +++ b/docs/sql/grammar.md @@ -150,6 +150,9 @@ keyword can be used to indicate that the declared view is not exposed to the outside world as an output of the computation. This is useful for modularizing the SQL code, by declaring intermediate views that are used in the implementation of other views. +The `MATERIALIZED` keyword instructs Feldera to maintain a full copy +of the view's output in addition to producing the +stream of changes. ``` createViewStatement @@ -255,10 +258,6 @@ exprOrList In `orderItem`, if expression is a positive integer n, it denotes the nth item in the `SELECT` clause. -If a view is marked as `MATERIALIZED`, the implementation will -maintain a full copy of the view's output in addition to producing the -expected changes. - ### Aggregate queries An aggregate query is a query that contains a `GROUP BY` or a `HAVING` diff --git a/docs/use_cases/fraud_detection/fraud_detection.md b/docs/use_cases/fraud_detection/fraud_detection.md index 7cccd29eeff..678f3386123 100644 --- a/docs/use_cases/fraud_detection/fraud_detection.md +++ b/docs/use_cases/fraud_detection/fraud_detection.md @@ -168,7 +168,7 @@ def build_program(client, pipeline_name): window_30_day AS (PARTITION BY t.cc_num ORDER BY unix_time RANGE BETWEEN 2592000 PRECEDING AND CURRENT ROW); """ - sql.register_output_view("FEATURE", query) + sql.register_view("FEATURE", query) return sql ``` diff --git a/openapi.json b/openapi.json index 8e9710f0d3b..48c82d91cda 100644 --- a/openapi.json +++ b/openapi.json @@ -4898,6 +4898,9 @@ "$ref": "#/components/schemas/Field" } }, + "materialized": { + "type": "boolean" + }, "name": { "type": "string" } diff --git a/python/docs/examples.rst b/python/docs/examples.rst index dbc5595f40f..0bad0b05c63 100644 --- a/python/docs/examples.rst +++ b/python/docs/examples.rst @@ -6,7 +6,7 @@ Using Pandas DataFrames as Input / Output You can use :meth:`.SQLContext.input_pandas` to connect a -DataFrame to a feldera table as the data source. +DataFrame to a feldera table as the data source. To listen for response from feldera, in the form of DataFrames call :meth:`.SQLContext.listen`. @@ -45,7 +45,7 @@ To ensure all data is received start listening before calling # here, we provide a query, that gets registered as a view in feldera # this query will be executed on the data in the table query = f"SELECT name, ((science + maths + art) / 3) as average FROM {TBL_NAMES[0]} JOIN {TBL_NAMES[1]} on id = student_id ORDER BY average DESC" - sql.register_output_view(view_name, query) + sql.register_view(view_name, query) # listen for the output of the view here in the notebook # you do not need to call this if you are forwarding the data to a sink @@ -132,7 +132,7 @@ More on Kafka as the output connector at: https://www.feldera.com/docs/connector sql = SQLContext('kafka', 'http://localhost:8080').get_or_create() sql.register_table(TABLE_NAME, SQLSchema({"id": "INT NOT NULL PRIMARY KEY"})) - sql.register_output_view(VIEW_NAME, f"SELECT COUNT(*) as num_rows FROM {TABLE_NAME}") + sql.register_view(VIEW_NAME, f"SELECT COUNT(*) as num_rows FROM {TABLE_NAME}") source_config = { "topics": ["example_topic"], @@ -186,7 +186,7 @@ More on the HTTP GET connector at: https://www.feldera.com/docs/connectors/sourc sql.register_table(TBL_NAME, SQLSchema({"id": "INT", "name": "STRING"})) - sql.register_output_view(VIEW_NAME, f"SELECT * FROM {TBL_NAME}") + sql.register_view(VIEW_NAME, f"SELECT * FROM {TBL_NAME}") path = "https://feldera-basics-tutorial.s3.amazonaws.com/part.json" diff --git a/python/docs/introduction.rst b/python/docs/introduction.rst index 1e4ca894899..c325d1c8692 100644 --- a/python/docs/introduction.rst +++ b/python/docs/introduction.rst @@ -1,11 +1,11 @@ Introduction ============ -The Feldera Python SDK is meant to provide an easy and convenient way of -interacting with Feldera. +The Feldera Python SDK is meant to provide an easy and convenient way of +interacting with Feldera. -Please submit any feature request / bug reports to: +Please submit any feature request / bug reports to: https://github.com/feldera/feldera @@ -30,7 +30,7 @@ Key Concepts ************ * :class:`.FelderaClient` - - This is the actual HTTP client used to make requests to your Feldera + - This is the actual HTTP client used to make requests to your Feldera instance. - creating an instance of :class:`.FelderaClient` is usually the first thing you will do while working with Feldera. @@ -42,13 +42,13 @@ Key Concepts from feldera import FelderaClient client = FelderaClient("https://try.feldera.com", api_key="YOUR_API_KEY") - + - The API key may not be required if you are running Feldera locally. * :class:`.SQLContext` - - This represents the current context of your SQL program, data sources + - This represents the current context of your SQL program, data sources and sinks. In Feldera terminology, this represents both a Program and a Pipeline. @@ -73,7 +73,7 @@ Key Concepts - Example: .. code-block:: python - + from feldera import SQLSchema tbl_name = "user_data" @@ -84,7 +84,7 @@ Key Concepts # Register Views based on your queries query = f"SELECT * FROM {tbl_name}" - sql.register_output_view(view_name, query) + sql.register_view(view_name, query) # name for this connector in_con = "delta_input_conn" @@ -105,12 +105,12 @@ Key Concepts sql.wait_for_completion(shutdown=True) - Here, we register a data table which receives data from input sources. - - Then, we register a view that performs operations on this input data. + - Then, we register a view that performs operations on this input data. You can also register other views on top of existing views. - Then, we connect a source delta table to the previously defined table. - Then, we connect a sink delta table to the previously defined view. - Finally, we run the pipeline to completion. Feldera will fetch data from - the source, perform the query you supplied and passes this data to the + the source, perform the query you supplied and passes this data to the sink delta table. .. warning:: @@ -122,7 +122,7 @@ Key Concepts - Example: .. code-block:: python - + sql.start() - This tells Feldera to go ahead and start processing the data. diff --git a/python/feldera/_sql_view.py b/python/feldera/_sql_view.py index bb3df691a06..a5a76fc1eae 100644 --- a/python/feldera/_sql_view.py +++ b/python/feldera/_sql_view.py @@ -1,9 +1,16 @@ +from enum import Enum from typing import List +class ViewKind(Enum): + DEFAULT = 1 + LOCAL = 2 + MATERIALIZED = 3 + + class SQLView: - def __init__(self, name: str, local: bool, query: str): + def __init__(self, name: str, kind: ViewKind, query: str): self.name: str = name - self.local = local + self.kind = kind query = query.strip() if query[-1] != ';': @@ -16,7 +23,13 @@ def add_lateness(self, timestamp_column: str, lateness_expr: str): self.lateness.append(f"LATENESS {self.name}.{timestamp_column} {lateness_expr};") def build_ddl(self): - local = " LOCAL" if self.local else "" - view = f"CREATE{local} VIEW {self.name} AS {self.query}" + match self.kind: + case ViewKind.DEFAULT: + kind = "" + case ViewKind.LOCAL: + kind = " LOCAL" + case ViewKind.MATERIALIZED: + kind = " MATERIALIZED" + view = f"CREATE{kind} VIEW {self.name} AS {self.query}" statements = self.lateness + [view] return "\n".join(statements) \ No newline at end of file diff --git a/python/feldera/sql_context.py b/python/feldera/sql_context.py index 13e9545f796..c420823ae4b 100644 --- a/python/feldera/sql_context.py +++ b/python/feldera/sql_context.py @@ -14,7 +14,7 @@ from feldera.rest.pipeline import Pipeline from feldera.rest.connector import Connector from feldera._sql_table import SQLTable -from feldera._sql_view import SQLView +from feldera._sql_view import SQLView, ViewKind from feldera.sql_schema import SQLSchema from feldera.output_handler import OutputHandler from feldera._callback_runner import CallbackRunner, _CallbackRunnerInstruction @@ -301,9 +301,9 @@ def register_local_view(self, name: str, query: str): :param query: The query to be used to create the view. """ - self.views[name] = SQLView(name, True, query) + self.views[name] = SQLView(name, ViewKind.LOCAL, query) - def register_output_view(self, name: str, query: str): + def register_view(self, name: str, query: str): """ Register a Feldera View based on the provided query. Auto inserts the trailing semicolon if not present. @@ -312,7 +312,18 @@ def register_output_view(self, name: str, query: str): :param query: The query to be used to create the view. """ - self.views[name] = SQLView(name, False, query) + self.views[name] = SQLView(name, ViewKind.DEFAULT, query) + + def register_materialized_view(self, name: str, query: str): + """ + Register a Feldera materialized View based on the provided query. + Auto inserts the trailing semicolon if not present. + + :param name: The name of the view. + :param query: The query to be used to create the view. + """ + + self.views[name] = SQLView(name, ViewKind.MATERIALIZED, query) def register_type(self, name: str, spec: str): """ diff --git a/python/tests/test_wireframes.py b/python/tests/test_wireframes.py index d258eacd783..cacda59e57c 100644 --- a/python/tests/test_wireframes.py +++ b/python/tests/test_wireframes.py @@ -27,7 +27,7 @@ def test_local(self): })) query = f"SELECT name, ((science + maths + art) / 3) as average FROM {TBL_NAMES[0]} JOIN {TBL_NAMES[1]} on id = student_id ORDER BY average DESC" - sql.register_output_view(view_name, query) + sql.register_view(view_name, query) out = sql.listen(view_name) sql.start() @@ -61,7 +61,7 @@ def test_local_listen_after_start(self): })) query = f"SELECT name, ((science + maths + art) / 3) as average FROM {TBL_NAMES[0]} JOIN {TBL_NAMES[1]} on id = student_id ORDER BY average DESC" - sql.register_output_view(view_name, query) + sql.register_view(view_name, query) sql.start() out = sql.listen(view_name) @@ -96,8 +96,8 @@ def test_two_SQLContexts(self): "art": "INT" })) - sql.register_output_view(VIEW_NAMES[0], f"SELECT * FROM {TBL_NAMES[0]}") - sql2.register_output_view(VIEW_NAMES[1], f"SELECT * FROM {TBL_NAMES[1]}") + sql.register_view(VIEW_NAMES[0], f"SELECT * FROM {TBL_NAMES[0]}") + sql2.register_view(VIEW_NAMES[1], f"SELECT * FROM {TBL_NAMES[1]}") out = sql.listen(VIEW_NAMES[0]) out2 = sql2.listen(VIEW_NAMES[1]) @@ -140,7 +140,7 @@ def callback(df: pd.DataFrame, seq_no: int): })) query = f"SELECT name, ((science + maths + art) / 3) as average FROM {TBL_NAMES[0]} JOIN {TBL_NAMES[1]} on id = student_id ORDER BY average DESC" - sql.register_output_view(view_name, query) + sql.register_view(view_name, query) sql.start() sql.foreach_chunk(view_name, callback) @@ -158,7 +158,7 @@ def test_df_without_columns(self): df = pd.DataFrame([(1, "a"), (2, "b"), (3, "c")]) sql.register_table(TBL_NAME, SQLSchema({"id": "INT", "name": "STRING"})) - sql.register_output_view("s", f"SELECT * FROM {TBL_NAME}") + sql.register_view("s", f"SELECT * FROM {TBL_NAME}") sql.start() @@ -172,7 +172,7 @@ def test_sql_error(self): sql = SQLContext('sql_error', TEST_CLIENT).get_or_create() TBL_NAME = "student" sql.register_table(TBL_NAME, SQLSchema({"id": "INT", "name": "STRING"})) - sql.register_output_view("s", f"SELECT FROM blah") + sql.register_view("s", f"SELECT FROM blah") _ = sql.listen("s") with self.assertRaises(Exception): @@ -222,7 +222,7 @@ def test_kafka(self): sql = SQLContext('kafka_test', TEST_CLIENT).get_or_create() sql.register_table(TABLE_NAME, SQLSchema({"id": "INT NOT NULL PRIMARY KEY"})) - sql.register_output_view(VIEW_NAME, f"SELECT COUNT(*) as num_rows FROM {TABLE_NAME}") + sql.register_view(VIEW_NAME, f"SELECT COUNT(*) as num_rows FROM {TABLE_NAME}") PIPELINE_TO_KAFKA_SERVER = "redpanda:9092" @@ -260,7 +260,7 @@ def test_http_get(self): sql.register_table(TBL_NAME, SQLSchema({"id": "INT", "name": "STRING"})) - sql.register_output_view(VIEW_NAME, f"SELECT * FROM {TBL_NAME}") + sql.register_view(VIEW_NAME, f"SELECT * FROM {TBL_NAME}") path = "https://feldera-basics-tutorial.s3.amazonaws.com/part.json" @@ -301,7 +301,7 @@ def test_avro_format(self): VIEW_NAME = "s" sql.register_table(TBL_NAME, SQLSchema({"id": "INT", "name": "STRING"})) - sql.register_output_view(VIEW_NAME, f"SELECT * FROM {TBL_NAME}") + sql.register_view(VIEW_NAME, f"SELECT * FROM {TBL_NAME}") sink_config = { "topic": TOPIC, @@ -361,7 +361,7 @@ def test_pipeline_resource_config(self): sql.register_table(TBL_NAME, SQLSchema({"id": "INT", "name": "STRING"})) - sql.register_output_view(VIEW_NAME, f"SELECT * FROM {TBL_NAME}") + sql.register_view(VIEW_NAME, f"SELECT * FROM {TBL_NAME}") path = "https://feldera-basics-tutorial.s3.amazonaws.com/part.json" @@ -390,7 +390,7 @@ def test_timestamp_pandas(self): # backend doesn't support TIMESTAMP of format: "2024-06-06T18:06:28.443" sql.register_table(TBL_NAME, SQLSchema({"id": "INT", "name": "STRING", "birthdate": "TIMESTAMP"})) - sql.register_output_view(VIEW_NAME, f"SELECT * FROM {TBL_NAME}") + sql.register_view(VIEW_NAME, f"SELECT * FROM {TBL_NAME}") df = pd.DataFrame({"id": [1, 2, 3], "name": ["a", "b", "c"], "birthdate": [ pd.Timestamp.now(), pd.Timestamp.now(), pd.Timestamp.now() From e5401cf24169566cc6c42391d7cebf47989b9445 Mon Sep 17 00:00:00 2001 From: George Date: Fri, 28 Jun 2024 17:56:35 +0000 Subject: [PATCH 4/7] WebConsole: disable browsing data for non-materialized relations Update API typings Signed-off-by: George --- web-console/openapi-fixes.patch | 3 +- web-console/package.json | 2 +- .../(app)/streaming/inspection/page.tsx | 18 +- .../streaming/import/SQLValueInput.tsx | 4 + .../streaming/import/randomData/generators.ts | 3 +- .../inspection/TableInspectionTab.tsx | 23 +- .../streaming/management/PipelineTable.tsx | 64 ++-- .../streaming/import/useDefaultRows.ts | 1 + .../src/lib/compositions/useHashPart.ts | 2 +- web-console/src/lib/functions/sqlValue.ts | 9 + .../src/lib/services/manager/core/OpenAPI.ts | 2 +- .../src/lib/services/manager/core/request.ts | 12 +- .../src/lib/services/manager/customRequest.ts | 329 ++++++++++++++++++ .../lib/services/manager/models/ColumnType.ts | 2 + .../lib/services/manager/models/Relation.ts | 1 + .../lib/services/manager/models/SqlType.ts | 1 + 16 files changed, 432 insertions(+), 44 deletions(-) create mode 100644 web-console/src/lib/services/manager/customRequest.ts diff --git a/web-console/openapi-fixes.patch b/web-console/openapi-fixes.patch index 931270011ab..385d566ba9a 100644 --- a/web-console/openapi-fixes.patch +++ b/web-console/openapi-fixes.patch @@ -2,8 +2,7 @@ diff --git a/web-console/src/lib/services/manager/models/ColumnType.ts b/web-con index 8484d2bea..953d53760 100644 --- a/web-console/src/lib/services/manager/models/ColumnType.ts +++ b/web-console/src/lib/services/manager/models/ColumnType.ts -@@ -57,5 +57,5 @@ export type ColumnType = { - * - `DECIMAL(1,2)` sets scale to `2`. +@@ -59,4 +59,4 @@ * - `DECIMAL(1,2)` sets scale to `2`. */ scale?: number | null - type?: SqlType diff --git a/web-console/package.json b/web-console/package.json index 848838517b4..06d303e3e7f 100644 --- a/web-console/package.json +++ b/web-console/package.json @@ -22,7 +22,7 @@ "lint": "eslint --max-warnings 0 --fix \"src/**/*.{js,jsx,ts,tsx}\"", "format": "prettier --write \"{src,tests}/**/*.{js,jsx,ts,tsx}\"", "format-check": "prettier --check \"src/**/*.{js,jsx,ts,tsx}\"", - "generate-openapi": "openapi --input ../openapi.json --request ./src/lib/services/manager/core/request.ts --output ./src/lib/services/manager && yarn format && patch -p2 < openapi-fixes.patch", + "generate-openapi": "openapi --input ../openapi.json --request ./src/lib/services/manager/customRequest.ts --output ./src/lib/services/manager && yarn format && patch -p2 < openapi-fixes.patch", "build-openapi": "cd .. && cargo run --bin pipeline-manager -- --dump-openapi", "test": "PLAYWRIGHT_API_ORIGIN=http://localhost:8080/ PLAYWRIGHT_APP_ORIGIN=http://localhost:8080/ DISPLAY= yarn playwright test", "test-ui": "PLAYWRIGHT_API_ORIGIN=http://localhost:8080/ PLAYWRIGHT_APP_ORIGIN=http://localhost:8080/ DISPLAY= yarn playwright test --ui-host=0.0.0.0", diff --git a/web-console/src/app/(spa)/(root)/(authenticated)/(app)/streaming/inspection/page.tsx b/web-console/src/app/(spa)/(root)/(authenticated)/(app)/streaming/inspection/page.tsx index 4218b78abaf..793ed632dd5 100644 --- a/web-console/src/app/(spa)/(root)/(authenticated)/(app)/streaming/inspection/page.tsx +++ b/web-console/src/app/(spa)/(root)/(authenticated)/(app)/streaming/inspection/page.tsx @@ -12,10 +12,10 @@ import { useSearchParams } from 'next/navigation' import { useEffect } from 'react' import { nonNull } from 'src/lib/functions/common/function' -import { useHash } from '@mantine/hooks' import { Alert, AlertTitle, Autocomplete, Box, FormControl, Link, MenuItem, TextField } from '@mui/material' import Grid from '@mui/material/Grid' import { useQuery } from '@tanstack/react-query' +import { useHashPart } from 'src/lib/compositions/useHashPart' const TablesBreadcrumb = (props: { pipeline: Pipeline @@ -24,10 +24,10 @@ const TablesBreadcrumb = (props: { tables: Relation[] views: Relation[] }) => { - const [tab] = useHash() + const [tab] = useHashPart() const options = props.tables - .map(relation => ({ type: 'Tables', name: getCaseIndependentName(relation) })) - .concat(props.views.map(relation => ({ type: 'Views', name: getCaseIndependentName(relation) }))) + .map(relation => ({ type: 'Tables', name: getCaseIndependentName(relation), relation })) + .concat(props.views.map(relation => ({ type: 'Views', name: getCaseIndependentName(relation), relation }))) return ( @@ -42,14 +42,18 @@ const TablesBreadcrumb = (props: { slotProps={{ popupIndicator: { 'data-testid': 'button-expand-relations' } as any }} ListboxProps={{ 'data-testid': 'box-relation-options' } as any} renderInput={params => } - value={{ name: props.caseIndependentName, type: props.relationType === 'table' ? 'Tables' : 'Views' }} + value={{ + name: props.caseIndependentName, + type: props.relationType === 'table' ? 'Tables' : 'Views', + relation: undefined! + }} renderOption={(_props, item) => ( @@ -63,7 +67,7 @@ const TablesBreadcrumb = (props: { } export default () => { - const [tab, setTab] = (([tab, setTab]) => [tab.slice(1) || 'browse', setTab])(useHash()) + const [tab, setTab] = (([tab, setTab]) => [tab || 'browse', setTab])(useHashPart()) // Parse config, view, tab arguments from router query const query = useSearchParams() diff --git a/web-console/src/lib/components/streaming/import/SQLValueInput.tsx b/web-console/src/lib/components/streaming/import/SQLValueInput.tsx index 35e129657b0..90f4c52de70 100644 --- a/web-console/src/lib/components/streaming/import/SQLValueInput.tsx +++ b/web-console/src/lib/components/streaming/import/SQLValueInput.tsx @@ -221,6 +221,10 @@ export const SQLValueInput = ({ type: 'string', ...props })) + .with('MAP', () => ({ + type: 'string', + ...props + })) .with('NULL', () => ({ type: 'string', ...props, diff --git a/web-console/src/lib/components/streaming/import/randomData/generators.ts b/web-console/src/lib/components/streaming/import/randomData/generators.ts index 4cc35d3250d..ea04b773481 100644 --- a/web-console/src/lib/components/streaming/import/randomData/generators.ts +++ b/web-console/src/lib/components/streaming/import/randomData/generators.ts @@ -119,6 +119,7 @@ const getDefaultRngMethodName = (sqlType: ColumnType): string => { .with({ type: 'VARBINARY' }, () => 'VARBINARY type not implemented') .with({ type: { Interval: P._ } }, () => 'INTERVAL type not supported') .with({ type: 'STRUCT' }, () => 'STRUCT type not supported') + .with({ type: 'MAP' }, () => 'MAP type not supported') .with({ type: 'NULL' }, () => 'NULL type not supported') .exhaustive() } @@ -192,7 +193,7 @@ export const columnTypeToRngOptions = (type: ColumnType): IRngGenMethod[] => { invariant(type.component) return transformToArrayGenerator(type, columnTypeToRngOptions(type.component)) }) - .with({ Interval: P._ }, 'BINARY', 'VARBINARY', 'STRUCT', 'NULL', () => UNSUPPORTED_TYPE_GENERATORS) + .with({ Interval: P._ }, 'BINARY', 'VARBINARY', 'STRUCT', 'NULL', 'MAP', () => UNSUPPORTED_TYPE_GENERATORS) .exhaustive() .map(({ generator, ...rng }) => ({ ...rng, diff --git a/web-console/src/lib/components/streaming/inspection/TableInspectionTab.tsx b/web-console/src/lib/components/streaming/inspection/TableInspectionTab.tsx index 6e56b6872f1..1e28fa5b343 100644 --- a/web-console/src/lib/components/streaming/inspection/TableInspectionTab.tsx +++ b/web-console/src/lib/components/streaming/inspection/TableInspectionTab.tsx @@ -10,7 +10,7 @@ import { ErrorBoundary } from 'react-error-boundary' import TabContext from '@mui/lab/TabContext' import TabList from '@mui/lab/TabList' import TabPanel from '@mui/lab/TabPanel' -import { Alert, AlertTitle } from '@mui/material' +import { Alert, AlertTitle, Box, Tooltip } from '@mui/material' import Tab from '@mui/material/Tab' type Tab = 'browse' | 'insert' @@ -65,11 +65,28 @@ export const TableInspectionTab = ({ onChange={(_e, tab) => setTab(tab)} aria-label='tabs to insert and browse relations' > - + {relation.materialized ? ( + + ) : ( + + + + + + )} - + {!!relation.materialized && } {pipeline.state.current_status === PipelineStatus.RUNNING ? ( diff --git a/web-console/src/lib/components/streaming/management/PipelineTable.tsx b/web-console/src/lib/components/streaming/management/PipelineTable.tsx index 91879c79896..f972d45ada2 100644 --- a/web-console/src/lib/components/streaming/management/PipelineTable.tsx +++ b/web-console/src/lib/components/streaming/management/PipelineTable.tsx @@ -229,34 +229,54 @@ const DetailPanelContent = (props: { row: Pipeline }) => { headerName: 'Action', flex: 0.15, display: 'flex', - renderCell: params => ( - - - { + const materialized = params.row.relation.materialized + return ( + + - - - - {direction === 'input' && state.current_status == PipelineStatus.RUNNING && ( - - + - )} - - ) + {direction === 'input' && state.current_status == PipelineStatus.RUNNING && ( + + + + + + )} + + ) + } } ] } diff --git a/web-console/src/lib/compositions/streaming/import/useDefaultRows.ts b/web-console/src/lib/compositions/streaming/import/useDefaultRows.ts index af44b6e2daa..78eff5ae638 100644 --- a/web-console/src/lib/compositions/streaming/import/useDefaultRows.ts +++ b/web-console/src/lib/compositions/streaming/import/useDefaultRows.ts @@ -23,6 +23,7 @@ export const getDefaultValue = (columntype: ColumnType): SQLValueJS => .with({ type: 'VARBINARY' }, () => invariant(false, 'VARBINARY not implemented') as never) .with({ type: { Interval: P._ } }, () => invariant(false, 'INTERVAL not supported for ingress') as never) .with({ type: 'STRUCT' }, () => new Map()) + .with({ type: 'MAP' }, () => new Map()) .with({ type: 'NULL' }, () => invariant(false, 'NULL not supported for ingress') as never) .exhaustive() diff --git a/web-console/src/lib/compositions/useHashPart.ts b/web-console/src/lib/compositions/useHashPart.ts index 1d066dc1fd5..3fdc5c1ea43 100644 --- a/web-console/src/lib/compositions/useHashPart.ts +++ b/web-console/src/lib/compositions/useHashPart.ts @@ -4,7 +4,7 @@ import { useCallback, useEffect, useState } from 'react' // https://stackoverflow.com/questions/69343932/how-to-detect-change-in-the-url-hash-in-next-js export const useHashPart = () => { - const [hash, setHash] = useState(('window' in globalThis && window.location.hash) || '') + const [hash, setHash] = useState(('window' in globalThis && window.location.hash.slice(1)) || '') // https://github.com/vercel/next.js/discussions/49465#discussioncomment-5845312 const params = useParams() diff --git a/web-console/src/lib/functions/sqlValue.ts b/web-console/src/lib/functions/sqlValue.ts index 4b92b9833a3..2b9dcfa0ded 100644 --- a/web-console/src/lib/functions/sqlValue.ts +++ b/web-console/src/lib/functions/sqlValue.ts @@ -125,6 +125,9 @@ export const sqlValueToXgressJSON = (type: ColumnType, value: SQLValueJS): JSONX .with({ type: 'NULL' }, () => { invariant(false, 'NULL type is not supported for ingress') }) + .with({ type: 'MAP' }, () => { + invariant(false, 'MAP type is not supported for ingress') + }) .exhaustive() } @@ -210,6 +213,9 @@ export const xgressJSONToSQLValue = (type: ColumnType, value: JSONXgressValue): .with({ type: 'NULL' }, () => { invariant(false, 'NULL type is not supported for ingress') }) + .with({ type: 'MAP' }, () => { + invariant(false, 'MAP type is not supported for ingress') + }) .exhaustive() } @@ -272,6 +278,7 @@ export const numericRange = (sqlType: ColumnType) => { type: { Interval: P._ } }, { type: 'STRUCT' }, { type: 'NULL' }, + { type: 'MAP' }, () => { throw new Error(`Not a numeric type: ${sqlType.type}`) } @@ -304,6 +311,7 @@ export const dateTimeRange = (sqlType: ColumnType): Dayjs[] => 'ARRAY', 'STRUCT', 'NULL', + 'MAP', () => { throw new Error('Not a date/time type') } @@ -345,6 +353,7 @@ export const sqlValueComparator = (sqlType: ColumnType) => { .with('VARBINARY', () => () => 0) .with('STRUCT', () => () => 0) .with('NULL', () => () => 0) + .with('MAP', () => () => 0) .exhaustive() return (a: SQLValueJS, b: SQLValueJS) => { diff --git a/web-console/src/lib/services/manager/core/OpenAPI.ts b/web-console/src/lib/services/manager/core/OpenAPI.ts index b0b24890465..f2b0360e447 100644 --- a/web-console/src/lib/services/manager/core/OpenAPI.ts +++ b/web-console/src/lib/services/manager/core/OpenAPI.ts @@ -21,7 +21,7 @@ export type OpenAPIConfig = { export const OpenAPI: OpenAPIConfig = { BASE: '', - VERSION: '0.18.0', + VERSION: '0.19.0', WITH_CREDENTIALS: false, CREDENTIALS: 'include', TOKEN: undefined, diff --git a/web-console/src/lib/services/manager/core/request.ts b/web-console/src/lib/services/manager/core/request.ts index e5b3c45e40f..85873ea55f2 100644 --- a/web-console/src/lib/services/manager/core/request.ts +++ b/web-console/src/lib/services/manager/core/request.ts @@ -4,13 +4,13 @@ import JSONbig from 'true-json-bigint' /* eslint-disable */ -import { ApiError } from './ApiError' -import { CancelablePromise } from './CancelablePromise' +import { ApiError } from '$lib/services/manager/core/ApiError' +import { CancelablePromise } from '$lib/services/manager/core/CancelablePromise' -import type { ApiRequestOptions } from './ApiRequestOptions' -import type { ApiResult } from './ApiResult' -import type { OnCancel } from './CancelablePromise' -import type { OpenAPIConfig } from './OpenAPI' +import type { ApiRequestOptions } from '$lib/services/manager/core/ApiRequestOptions' +import type { ApiResult } from '$lib/services/manager/core/ApiResult' +import type { OnCancel } from '$lib/services/manager/core/CancelablePromise' +import type { OpenAPIConfig } from '$lib/services/manager/core/OpenAPI' export const isDefined = (value: T | null | undefined): value is Exclude => { return value !== undefined && value !== null } diff --git a/web-console/src/lib/services/manager/customRequest.ts b/web-console/src/lib/services/manager/customRequest.ts new file mode 100644 index 00000000000..85873ea55f2 --- /dev/null +++ b/web-console/src/lib/services/manager/customRequest.ts @@ -0,0 +1,329 @@ +/* generated using openapi-typescript-codegen -- do not edit */ +/* istanbul ignore file */ +/* tslint:disable */ +import JSONbig from 'true-json-bigint' + +/* eslint-disable */ +import { ApiError } from '$lib/services/manager/core/ApiError' +import { CancelablePromise } from '$lib/services/manager/core/CancelablePromise' + +import type { ApiRequestOptions } from '$lib/services/manager/core/ApiRequestOptions' +import type { ApiResult } from '$lib/services/manager/core/ApiResult' +import type { OnCancel } from '$lib/services/manager/core/CancelablePromise' +import type { OpenAPIConfig } from '$lib/services/manager/core/OpenAPI' +export const isDefined = (value: T | null | undefined): value is Exclude => { + return value !== undefined && value !== null +} + +export const isString = (value: any): value is string => { + return typeof value === 'string' +} + +export const isStringWithValue = (value: any): value is string => { + return isString(value) && value !== '' +} + +export const isBlob = (value: any): value is Blob => { + return ( + typeof value === 'object' && + typeof value.type === 'string' && + typeof value.stream === 'function' && + typeof value.arrayBuffer === 'function' && + typeof value.constructor === 'function' && + typeof value.constructor.name === 'string' && + /^(Blob|File)$/.test(value.constructor.name) && + /^(Blob|File)$/.test(value[Symbol.toStringTag]) + ) +} + +export const isFormData = (value: any): value is FormData => { + return value instanceof FormData +} + +export const base64 = (str: string): string => { + try { + return btoa(str) + } catch (err) { + // @ts-ignore + return Buffer.from(str).toString('base64') + } +} + +export const getQueryString = (params: Record): string => { + const qs: string[] = [] + + const append = (key: string, value: any) => { + qs.push(`${encodeURIComponent(key)}=${encodeURIComponent(String(value))}`) + } + + const process = (key: string, value: any) => { + if (isDefined(value)) { + if (Array.isArray(value)) { + value.forEach(v => { + process(key, v) + }) + } else if (typeof value === 'object') { + Object.entries(value).forEach(([k, v]) => { + process(`${key}[${k}]`, v) + }) + } else { + append(key, value) + } + } + } + + Object.entries(params).forEach(([key, value]) => { + process(key, value) + }) + + if (qs.length > 0) { + return `?${qs.join('&')}` + } + + return '' +} + +const getUrl = (config: OpenAPIConfig, options: ApiRequestOptions): string => { + const encoder = config.ENCODE_PATH || encodeURI + + const path = options.url + .replace('{api-version}', config.VERSION) + .replace(/{(.*?)}/g, (substring: string, group: string) => { + if (options.path?.hasOwnProperty(group)) { + return encoder(String(options.path[group])) + } + return substring + }) + + const url = `${config.BASE}${path}` + if (options.query) { + return `${url}${getQueryString(options.query)}` + } + return url +} + +export const getFormData = (options: ApiRequestOptions): FormData | undefined => { + if (options.formData) { + const formData = new FormData() + + const process = (key: string, value: any) => { + if (isString(value) || isBlob(value)) { + formData.append(key, value) + } else { + formData.append(key, JSONbig.stringify(value)) + } + } + + Object.entries(options.formData) + .filter(([_, value]) => isDefined(value)) + .forEach(([key, value]) => { + if (Array.isArray(value)) { + value.forEach(v => process(key, v)) + } else { + process(key, value) + } + }) + + return formData + } + return undefined +} + +type Resolver = (options: ApiRequestOptions) => Promise + +export const resolve = async (options: ApiRequestOptions, resolver?: T | Resolver): Promise => { + if (typeof resolver === 'function') { + return (resolver as Resolver)(options) + } + return resolver +} + +export const getHeaders = async (config: OpenAPIConfig, options: ApiRequestOptions): Promise => { + const [token, username, password, additionalHeaders] = await Promise.all([ + resolve(options, config.TOKEN), + resolve(options, config.USERNAME), + resolve(options, config.PASSWORD), + resolve(options, config.HEADERS) + ]) + + const headers = Object.entries({ + Accept: 'application/json', + ...additionalHeaders, + ...options.headers + }) + .filter(([_, value]) => isDefined(value)) + .reduce( + (headers, [key, value]) => ({ + ...headers, + [key]: String(value) + }), + {} as Record + ) + + if (isStringWithValue(token)) { + headers['Authorization'] = `Bearer ${token}` + } + + if (isStringWithValue(username) && isStringWithValue(password)) { + const credentials = base64(`${username}:${password}`) + headers['Authorization'] = `Basic ${credentials}` + } + + if (options.body !== undefined) { + if (options.mediaType) { + headers['Content-Type'] = options.mediaType + } else if (isBlob(options.body)) { + headers['Content-Type'] = options.body.type || 'application/octet-stream' + } else if (isString(options.body)) { + headers['Content-Type'] = 'text/plain' + } else if (!isFormData(options.body)) { + headers['Content-Type'] = 'application/json' + } + } + + return new Headers(headers) +} + +export const getRequestBody = (options: ApiRequestOptions): any => { + if (options.body !== undefined) { + if (options.mediaType?.includes('/json')) { + return JSONbig.stringify(options.body) + } else if (isString(options.body) || isBlob(options.body) || isFormData(options.body)) { + return options.body + } else { + return JSONbig.stringify(options.body) + } + } + return undefined +} + +export const sendRequest = async ( + config: OpenAPIConfig, + options: ApiRequestOptions, + url: string, + body: any, + formData: FormData | undefined, + headers: Headers, + onCancel: OnCancel +): Promise => { + const controller = new AbortController() + + const request: RequestInit = { + headers, + body: body ?? formData, + method: options.method, + signal: controller.signal + } + + if (config.WITH_CREDENTIALS) { + request.credentials = config.CREDENTIALS + } + + onCancel(() => controller.abort()) + + return await fetch(url, request) +} + +export const getResponseHeader = (response: Response, responseHeader?: string): string | undefined => { + if (responseHeader) { + const content = response.headers.get(responseHeader) + if (isString(content)) { + return content + } + } + return undefined +} + +export const getResponseBody = async (response: Response): Promise => { + if (response.status !== 204) { + try { + const contentType = response.headers.get('Content-Type') + if (contentType) { + const jsonTypes = ['application/json', 'application/problem+json'] + const isJSON = jsonTypes.some(type => contentType.toLowerCase().startsWith(type)) + if (isJSON) { + return JSONbig.parse(await response.text()) + } else { + return await response.text() + } + } + } catch (error) { + console.error(error) + } + } + return undefined +} + +export const catchErrorCodes = (options: ApiRequestOptions, result: ApiResult): void => { + const errors: Record = { + 400: 'Bad Request', + 401: 'Unauthorized', + 403: 'Forbidden', + 404: 'Not Found', + 500: 'Internal Server Error', + 502: 'Bad Gateway', + 503: 'Service Unavailable', + ...options.errors + } + + const error = errors[result.status] + if (error) { + throw new ApiError(options, result, error) + } + + if (!result.ok) { + const errorStatus = result.status ?? 'unknown' + const errorStatusText = result.statusText ?? 'unknown' + const errorBody = (() => { + try { + return JSONbig.stringify(result.body, null, 2) + } catch (e) { + return undefined + } + })() + + throw new ApiError( + options, + result, + `Generic Error: status: ${errorStatus}; status text: ${errorStatusText}; body: ${errorBody}` + ) + } +} + +/** + * Request method + * @param config The OpenAPI configuration object + * @param options The request options from the service + * @returns CancelablePromise + * @throws ApiError + */ +export const request = (config: OpenAPIConfig, options: ApiRequestOptions): CancelablePromise => { + return new CancelablePromise(async (resolve, reject, onCancel) => { + try { + const url = getUrl(config, options) + const formData = getFormData(options) + const body = getRequestBody(options) + const headers = await getHeaders(config, options) + + if (!onCancel.isCancelled) { + const response = await sendRequest(config, options, url, body, formData, headers, onCancel) + const responseBody = await getResponseBody(response) + const responseHeader = getResponseHeader(response, options.responseHeader) + + const result: ApiResult = { + url, + ok: response.ok, + status: response.status, + statusText: response.statusText, + body: responseHeader ?? responseBody + } + + catchErrorCodes(options, result) + + resolve(result.body) + } + } catch (error) { + reject(error) + } + }) +} diff --git a/web-console/src/lib/services/manager/models/ColumnType.ts b/web-console/src/lib/services/manager/models/ColumnType.ts index 953d53760a5..42be89fc58f 100644 --- a/web-console/src/lib/services/manager/models/ColumnType.ts +++ b/web-console/src/lib/services/manager/models/ColumnType.ts @@ -35,6 +35,7 @@ export type ColumnType = { * ``` */ fields?: Array | null + key?: ColumnType | null /** * Does the type accept NULL values? */ @@ -58,4 +59,5 @@ export type ColumnType = { */ scale?: number | null type: SqlType + value?: ColumnType | null } diff --git a/web-console/src/lib/services/manager/models/Relation.ts b/web-console/src/lib/services/manager/models/Relation.ts index 658fa492564..5e76405d2c7 100644 --- a/web-console/src/lib/services/manager/models/Relation.ts +++ b/web-console/src/lib/services/manager/models/Relation.ts @@ -11,5 +11,6 @@ import type { Field } from './Field' export type Relation = { case_sensitive?: boolean fields: Array + materialized?: boolean name: string } diff --git a/web-console/src/lib/services/manager/models/SqlType.ts b/web-console/src/lib/services/manager/models/SqlType.ts index 9b730a155ed..f2c82e9c59d 100644 --- a/web-console/src/lib/services/manager/models/SqlType.ts +++ b/web-console/src/lib/services/manager/models/SqlType.ts @@ -27,4 +27,5 @@ export type SqlType = } | 'ARRAY' | 'STRUCT' + | 'MAP' | 'NULL' From bb2191d0e9b01a78bf8d6684ec6d8a41b87e17d5 Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Fri, 28 Jun 2024 11:16:55 -0700 Subject: [PATCH 5/7] Doc section on materialized views. Add a section on materialized tables and views in SQL compiler docs. Signed-off-by: Leonid Ryzhyk --- docs/sidebars.js | 1 + docs/sql/grammar.md | 13 ++++---- docs/sql/intro.mdx | 7 ++--- docs/sql/materialized.md | 65 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 76 insertions(+), 10 deletions(-) create mode 100644 docs/sql/materialized.md diff --git a/docs/sidebars.js b/docs/sidebars.js index 257edde980e..d7f3b4f17ca 100644 --- a/docs/sidebars.js +++ b/docs/sidebars.js @@ -201,6 +201,7 @@ const sidebars = { 'sql/array', 'sql/map', 'sql/datetime', + 'sql/materialized', 'sql/streaming', 'sql/udf' ] diff --git a/docs/sql/grammar.md b/docs/sql/grammar.md index 4def39a0e98..6ff57492ad7 100644 --- a/docs/sql/grammar.md +++ b/docs/sql/grammar.md @@ -108,17 +108,16 @@ CREATE TABLE empsalary ( enroll_date date ) WITH ( 'source' = 'kafka', - 'url' = 'localhost:8080', - 'materialized' = 'false' + 'url' = 'localhost:8080' ); ``` -Unlike a database, Feldera does normally not maintain the contents of +Unlike a database, Feldera does not normally maintain the contents of tables; it will only store as much data as necessary to compute future outputs. By specifying the property `'materialized' = 'true'` a user -instructs Feldera to also maintain the complete contents of a table. -The contents of the table can be queried using the `http`-based API -described elsewhere. +instructs Feldera to also maintain the complete contents of the table. +Such materialized tables can be browsed and queried at runtime. +See [Materialized Tables and Views](materialized.md) for more details. ### LATENESS @@ -153,6 +152,8 @@ are used in the implementation of other views. The `MATERIALIZED` keyword instructs Feldera to maintain a full copy of the view's output in addition to producing the stream of changes. +Such materialized views can be browsed and queried at runtime. +See [Materialized Tables and Views](materialized.md) for more details. ``` createViewStatement diff --git a/docs/sql/intro.mdx b/docs/sql/intro.mdx index e23b087bd8c..a2b2a0dd5d6 100644 --- a/docs/sql/intro.mdx +++ b/docs/sql/intro.mdx @@ -22,9 +22,8 @@ Feldera is used in the following way: become *inputs* for DBSP. - users define a set of database views. The views become *outputs* for DBSP (unless they are declared as being `LOCAL`). - The views can be defined either in Rust, using the DBSP library, or can be - implemented in standard SQL, and compiled to Rust using the SQL to DBSP - compiler. + The views are implemented in standard SQL, and compiled to Rust using the + SQL to DBSP compiler. - the compiled DBSP program is started - DBSP assumes that the tables are initially empty - users inform DBSP of any *changes* of the input tables @@ -79,7 +78,7 @@ Differences between DBSP and a database: ## Supported SQL Constructs -Despite these limitations, DBSP offers a powerful set of features: +Feldera offers a powerful set of features: - A rich set of data types, including the standard SQL datatypes, dates, times, intervals, arrays, maps, user-defined types diff --git a/docs/sql/materialized.md b/docs/sql/materialized.md new file mode 100644 index 00000000000..c9b70a2d29d --- /dev/null +++ b/docs/sql/materialized.md @@ -0,0 +1,65 @@ +# Materialized Tables and Views + +By default, Feldera does not maintain the complete contents of tables and views; it only +stores the data necessary to compute future outputs. However, in some cases, users +may need to inspect or query the entire contents of a relation. This can +be useful in the following scenarios: + +* **Debugging**. The user may want to inspect the current contents of tables and views + to validate their SQL program. +* **Retrieve full state snapshot**. This is useful, for instance, to sync the output of Feldera with an external + database on demand. +* **Ad hoc queries**. In some applications, users may not want to store a complete copy of the data, + but instead query it on demand. + +Feldera supports such use cases by allowing users to label tables and views as **materialized**. +To declare a materialized table, use the materialized attribute: + +```sql +CREATE TABLE my_table (...) WITH ('materialized' = 'true'); +``` + +To declare a materialized view, use the `CREATE MATERIALIZED VIEW` syntax: + +```sql +CREATE MATERIALIZED VIEW my_view as SELECT * from my_table; +``` + +These declarations instruct Feldera to maintain a complete snapshot of the table or view. + +## Using materialized tables and views + +### Web Console + +You can browse materialized tables and views in the Feldera Web Console by clicking on the "eye" +icon next to the table or view of a running pipeline: + +![Browsing a materialized view in Web Console](../tutorials/basics/preferred-vendor1.png) + +### Ad hoc queries + +:::caution Under Construction + +We are implementing ad hoc querying support for materialized tables and views. + +::: + + +## Usage considerations + +Materialized relations can significantly increase the storage used by the program. +For example, Feldera can evaluate simple programs with no joins or aggregates without keeping +any state. However, Mmaterializing inputs or outputs of such programs will make them +**stateful**, requiring storage proportional to the size of the materialized +relations. + +Feldera takes advantage of [`LATENESS` annotations](streaming.md#lateness-expressions) +to garbage collect old records in time series tables and views (i.e., tables and views with +monotonically or near-monotonically growing timestamps). This allows evaluating complex queries +over unbounded streams using bounded storage. Materializing these tables forces Feldera to keep +their entire history, resulting in unbounded storage growth. + +Finally note that materialized tables and views are **not a performance optimization**. +Feldera automatically maintains all the state needed to incrementally evaluate user queries +efficiently. Materializing additional relations will not make it faster, but can actually +slow it down, as it needs to write more data to storage. From 137ad1a2248e38ded5c03434bed837a4782c664c Mon Sep 17 00:00:00 2001 From: Leonid Ryzhyk Date: Thu, 27 Jun 2024 16:17:31 -0700 Subject: [PATCH 6/7] integration_test: Output compilation status. Print compilation status to help debug compilation timeouts. Signed-off-by: Leonid Ryzhyk --- .../pipeline_manager/src/integration_test.rs | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/crates/pipeline_manager/src/integration_test.rs b/crates/pipeline_manager/src/integration_test.rs index 38d6c941b16..ee940fcc535 100644 --- a/crates/pipeline_manager/src/integration_test.rs +++ b/crates/pipeline_manager/src/integration_test.rs @@ -497,13 +497,6 @@ impl TestConfig { println!("Waiting for compilation"); let mut last_wait_println = Instant::now(); loop { - if last_wait_println.elapsed().as_secs() >= 60 { - println!( - "Waiting for compilation since {} seconds", - start.elapsed().as_secs() - ); - last_wait_println = Instant::now(); - } std::thread::sleep(time::Duration::from_secs(1)); if start.elapsed().as_secs() > 480 { panic!("Compilation timeout"); @@ -524,6 +517,14 @@ impl TestConfig { panic!("Compilation failed with status {}", status); } } + if last_wait_println.elapsed().as_secs() >= 60 { + println!( + "Waiting for compilation since {} seconds, status: {}", + start.elapsed().as_secs(), + status + ); + last_wait_println = Instant::now(); + } } } @@ -1835,11 +1836,12 @@ async fn pipeline_start_without_compiling() { let val: Value = resp.json().await.unwrap(); let status = val["status"].clone(); + println!("Program status is: {status:?}"); + if status == json!("None") || status == json!("Pending") || status == json!("CompilingSql") { continue; } - println!("Program status is: {status:?}"); break; } // Start the program From e55938791e76e40a8b5d2ca87e2abea9f9308585 Mon Sep 17 00:00:00 2001 From: Mihai Budiu Date: Fri, 28 Jun 2024 20:14:01 -0700 Subject: [PATCH 7/7] Rebase on main Signed-off-by: Mihai Budiu --- .../sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java | 1 - 1 file changed, 1 deletion(-) diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java index fe76f5a9183..6a58ba40c1d 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/frontend/CalciteToDBSPCompiler.java @@ -2133,7 +2133,6 @@ DBSPNode compileCreateView(CreateViewStatement view) { o = new DBSPSinkOperator( view.getCalciteObject(), view.relationName, view.statement, struct, meta, vo); - this.circuit.addOperator(o); } else { // We may already have a node for this output DBSPOperator previous = this.circuit.getView(view.relationName);