diff --git a/sqrl-cli/src/main/java/com/datasqrl/compile/CompilationProcess.java b/sqrl-cli/src/main/java/com/datasqrl/compile/CompilationProcess.java index 0efc64a49..b448d0e3f 100644 --- a/sqrl-cli/src/main/java/com/datasqrl/compile/CompilationProcess.java +++ b/sqrl-cli/src/main/java/com/datasqrl/compile/CompilationProcess.java @@ -66,7 +66,7 @@ public Pair executeCompilation(Optional testsPath) buildPath, (FlinkStreamEngine) planner.getStreamStage().engine(), config.getCompilerConfig()); - planner.planMain(mainScript, environment); + planner.planMain(mainScript, Optional.empty(), environment); var dagBuilder = planner.getDagBuilder(); var dag = dagPlanner.optimize(dagBuilder.getDag()); var physicalPlan = dagPlanner.assemble(dag, environment); diff --git a/sqrl-planner/src/main/java/com/datasqrl/planner/SqlScriptPlanner.java b/sqrl-planner/src/main/java/com/datasqrl/planner/SqlScriptPlanner.java index f99c9e8bd..5c916d233 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/planner/SqlScriptPlanner.java +++ b/sqrl-planner/src/main/java/com/datasqrl/planner/SqlScriptPlanner.java @@ -211,14 +211,19 @@ public SqlScriptPlanner( } /** - * Main entry method for parsing a SQRL script. The bulk of this method ensure that exceptions and - * errors are correctly mapped to the source so that users can easily understand what the issue is - * and what's causing it. + * Main entry method for parsing a SQRL script. The bulk of this method ensures that exceptions + * and errors are correctly mapped to the source so that users can easily understand what the + * issue is and what's causing it. * - * @param mainScript - * @param sqrlEnv + * @param mainScript SQRL script to plan + * @param inheritedHints inherited hints if the planned script is coming from an IMPORT + * @param sqrlEnv the SQRL compilation environment */ - public void planMain(MainScript mainScript, Sqrl2FlinkSQLTranslator sqrlEnv) { + public void planMain( + MainScript mainScript, + Optional inheritedHints, + Sqrl2FlinkSQLTranslator sqrlEnv) { + var scriptErrors = errorCollector.withScript(mainScript.getPath(), mainScript.getContent()); var statements = sqrlParser.parseScript(mainScript.getContent(), scriptErrors); var statementStack = new ArrayList(); @@ -229,7 +234,7 @@ public void planMain(MainScript mainScript, Sqrl2FlinkSQLTranslator sqrlEnv) { var sqlStatement = statement.get(); try { - planStatement(sqlStatement, statementStack, sqrlEnv, lineErrors); + planStatement(sqlStatement, statementStack, inheritedHints, sqrlEnv, lineErrors); } catch (CollectedException e) { throw e; @@ -302,6 +307,7 @@ public void planMain(MainScript mainScript, Sqrl2FlinkSQLTranslator sqrlEnv) { private void planStatement( SQLStatement stmt, List statementStack, + Optional inheritedHints, Sqrl2FlinkSQLTranslator sqrlEnv, ErrorCollector errors) { // Process hints & doc @@ -309,7 +315,7 @@ private void planStatement( Optional documentation = Optional.empty(); if (stmt instanceof SqrlStatement statement) { var comments = statement.getComments(); - hints = PlannerHints.from(comments, errors); + hints = PlannerHints.from(comments, inheritedHints, errors); if (!comments.documentation().isEmpty()) { documentation = Optional.of( @@ -842,7 +848,7 @@ private void addImport( NamePath aliasPath = null; if (importStmt.getAlias().isPresent()) { aliasPath = importStmt.getAlias().get(); - ; + checkFatal( aliasPath.size() == 1, ErrorCode.INVALID_IMPORT, @@ -937,7 +943,11 @@ private void addImport( completeScript.append(stmts); } - planMain(scriptObject.getScript(), sqrlEnv); + var hints = + hintsAndDoc.hints().isEmpty() + ? Optional.empty() + : Optional.of(hintsAndDoc.hints()); + planMain(scriptObject.getScript(), hints, sqrlEnv); if (priorContext.hasDifferentDatabase(scriptContext)) { // Switch it back diff --git a/sqrl-planner/src/main/java/com/datasqrl/planner/hint/PlannerHints.java b/sqrl-planner/src/main/java/com/datasqrl/planner/hint/PlannerHints.java index 16982430f..1ac0027cd 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/planner/hint/PlannerHints.java +++ b/sqrl-planner/src/main/java/com/datasqrl/planner/hint/PlannerHints.java @@ -19,6 +19,7 @@ import com.datasqrl.error.ErrorLabel; import com.datasqrl.planner.parser.SqrlComments; import com.datasqrl.planner.parser.StatementParserException; +import java.util.ArrayList; import java.util.List; import java.util.Optional; import java.util.function.Function; @@ -34,12 +35,15 @@ public class PlannerHints { private final List hints; - public static PlannerHints from(SqrlComments comments, ErrorCollector errors) { - var hints = - comments.hints().stream() - .map(c -> PlannerHint.from(c, errors)) - .flatMap(Optional::stream) - .toList(); + public static PlannerHints from( + SqrlComments comments, Optional inheritedHints, ErrorCollector errors) { + var hints = new ArrayList(); + inheritedHints.ifPresent(inherited -> hints.addAll(inherited.hints)); + + comments.hints().stream() + .map(c -> PlannerHint.from(c, errors)) + .flatMap(Optional::stream) + .forEach(hints::add); return new PlannerHints(hints); } @@ -56,6 +60,10 @@ public Optional getQueryByHint() { return queryBy.stream().map(ColumnNamesHint.class::cast).findFirst(); } + public boolean isEmpty() { + return hints.isEmpty(); + } + public boolean isTest() { return getHint(TestHint.class).isPresent(); } diff --git a/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/DAGPlannerTest.java b/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/DAGPlannerTest.java index 4e5f9e6da..1acb05e91 100644 --- a/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/DAGPlannerTest.java +++ b/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/DAGPlannerTest.java @@ -78,7 +78,7 @@ void scripts(Path script) { @Disabled @Test void specificScript() { - var script = SCRIPT_DIR.resolve("unknownHint-warn.sqrl"); + var script = SCRIPT_DIR.resolve("imports.sqrl"); scripts(script); } diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/imports.sqrl b/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/imports.sqrl index e9378cb99..234ab6b39 100644 --- a/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/imports.sqrl +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/imports.sqrl @@ -1,4 +1,10 @@ IMPORT ecommerceTs.customer; -/*+query_by_all(productid) */ + +/*+ query_by_all(productid) */ IMPORT ecommerceTs.product; +/*+ ttl(14 days) */ +IMPORT mutation.events.*; + +-- Export mutation table explicitly for test snapshot +EXPORT MyEvents TO logger.myevents; diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/mutation/events.sqrl b/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/mutation/events.sqrl new file mode 100644 index 000000000..20ec3734e --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/mutation/events.sqrl @@ -0,0 +1,7 @@ +/*+ no_query, engine(kafka) */ +CREATE TABLE MyEvents ( + `id` STRING NOT NULL METADATA FROM 'uuid', + `payload` STRING, + `timestamp` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp', + WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '0.001' SECOND +); diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/imports.txt b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/imports.txt index a072a7abb..62aecb41c 100644 --- a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/imports.txt +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/imports.txt @@ -43,6 +43,42 @@ WITH ( 'connector' = 'filesystem' ) LIKE `Customer__schema` +=== MyEvents +ID: default_catalog.default_database.MyEvents +Type: stream +Stage: flink +Primary key: id +Timestamp: timestamp +Row count: ~1e8 +--- +Schema: + - id: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - payload: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" + - timestamp: TIMESTAMP_LTZ(3) *ROWTIME* NOT NULL +Inputs: + - default_catalog.default_database.MyEvents__base +Annotations: + - stream-root: MyEvents +Plan: +LogicalWatermarkAssigner(rowtime=[timestamp], watermark=[-($2, 1:INTERVAL SECOND)]) + LogicalProject(id=[$0], payload=[$1], timestamp=[CAST($2):TIMESTAMP_LTZ(3) *ROWTIME* NOT NULL]) + LogicalTableScan(table=[[default_catalog, default_database, MyEvents, metadata=[timestamp]]]) +SQL: +CREATE TABLE `MyEvents` ( + `id` STRING NOT NULL, + `payload` STRING, + `timestamp` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp', + WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '0.001' SECOND +) +WITH ( + 'connector' = 'kafka', + 'format' = 'flexible-json', + 'properties.auto.offset.reset' = 'earliest', + 'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}', + 'properties.compression.type' = 'zstd', + 'properties.group.id' = '${KAFKA_GROUP_ID}', + 'topic' = 'kafka-mutation-MyEvents' +) === Product ID: default_catalog.default_database.Product Type: stream @@ -86,6 +122,14 @@ WITH ( 'connector' = 'filesystem' ) LIKE `Product__schema` +=== myevents +ID: logger.myevents +Type: export +Stage: flink +--- +Inputs: + - default_catalog.default_database.MyEvents + >>>flink-sql-no-functions.sql CREATE TEMPORARY TABLE `Customer__schema` ( `customerid` BIGINT NOT NULL, @@ -129,6 +173,21 @@ WITH ( 'connector' = 'filesystem' ) LIKE `Product__schema`; +CREATE TABLE `MyEvents` ( + `id` STRING NOT NULL, + `payload` STRING, + `timestamp` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp', + WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '0.001' SECOND +) +WITH ( + 'connector' = 'kafka', + 'format' = 'flexible-json', + 'properties.auto.offset.reset' = 'earliest', + 'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}', + 'properties.compression.type' = 'zstd', + 'properties.group.id' = '${KAFKA_GROUP_ID}', + 'topic' = 'kafka-mutation-MyEvents' +); CREATE TABLE `Customer_1` ( `customerid` BIGINT NOT NULL, `email` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, @@ -146,7 +205,17 @@ WITH ( 'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}', 'username' = '${POSTGRES_USERNAME}' ); -CREATE TABLE `Product_2` ( +CREATE TABLE `myevents_2` ( + `id` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `payload` VARCHAR(2147483647) CHARACTER SET `UTF-16LE`, + `timestamp` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL, + PRIMARY KEY (`id`) NOT ENFORCED +) +WITH ( + 'connector' = 'print', + 'print-identifier' = 'myevents' +); +CREATE TABLE `Product_3` ( `productid` BIGINT NOT NULL, `name` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, `description` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, @@ -159,7 +228,7 @@ WITH ( 'driver' = 'org.postgresql.Driver', 'password' = '${POSTGRES_PASSWORD}', 'sink.on-conflict.action' = 'IGNORE', - 'table-name' = 'Product_2', + 'table-name' = 'Product_3', 'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}', 'username' = '${POSTGRES_USERNAME}' ); @@ -168,14 +237,32 @@ INSERT INTO `default_catalog`.`default_database`.`Customer_1` SELECT * FROM `default_catalog`.`default_database`.`Customer` ; -INSERT INTO `default_catalog`.`default_database`.`Product_2` +INSERT INTO `default_catalog`.`default_database`.`myevents_2` SELECT * - FROM `default_catalog`.`default_database`.`Product` + FROM `default_catalog`.`default_database`.`MyEvents` ; - END + INSERT INTO `default_catalog`.`default_database`.`Product_3` + SELECT * + FROM `default_catalog`.`default_database`.`Product` + ; + END >>>kafka.json { - "topics" : [ ], + "topics" : [ + { + "topicName" : "kafka-mutation-MyEvents", + "tableName" : "MyEvents", + "format" : "flexible-json", + "numPartitions" : 1, + "replicationFactor" : 3, + "type" : "MUTATION", + "messageKeys" : [ ], + "messageSchema" : "", + "config" : { + "retention.ms" : "1209600000" + } + } + ], "testRunnerTopics" : [ ] } >>>postgres.json @@ -222,9 +309,9 @@ INSERT INTO `default_catalog`.`default_database`.`Product_2` "ttl" : 0.0 }, { - "name" : "Product_2", + "name" : "Product_3", "type" : "TABLE", - "sql" : "CREATE TABLE IF NOT EXISTS \"Product_2\" (\"productid\" BIGINT NOT NULL, \"name\" TEXT NOT NULL, \"description\" TEXT NOT NULL, \"category\" TEXT NOT NULL, \"_ingest_time\" TIMESTAMP WITH TIME ZONE NOT NULL, PRIMARY KEY (\"productid\",\"name\",\"description\",\"category\"))", + "sql" : "CREATE TABLE IF NOT EXISTS \"Product_3\" (\"productid\" BIGINT NOT NULL, \"name\" TEXT NOT NULL, \"description\" TEXT NOT NULL, \"category\" TEXT NOT NULL, \"_ingest_time\" TIMESTAMP WITH TIME ZONE NOT NULL, PRIMARY KEY (\"productid\",\"name\",\"description\",\"category\"))", "fields" : [ { "name" : "productid", @@ -349,7 +436,7 @@ INSERT INTO `default_catalog`.`default_database`.`Product_2` ], "query" : { "type" : "SqlQuery", - "sql" : "SELECT *\nFROM \"Product_2\"\nWHERE \"productid\" = $1", + "sql" : "SELECT *\nFROM \"Product_3\"\nWHERE \"productid\" = $1", "parameters" : [ { "type" : "arg", @@ -364,7 +451,29 @@ INSERT INTO `default_catalog`.`default_database`.`Product_2` } } ], - "mutations" : [ ], + "mutations" : [ + { + "type" : "kafka", + "fieldName" : "MyEvents", + "returnList" : false, + "topic" : "kafka-mutation-MyEvents", + "keyColumns" : [ ], + "computedColumns" : { + "id" : { + "metadataType" : "UUID", + "name" : "", + "required" : true + }, + "timestamp" : { + "metadataType" : "TIMESTAMP", + "name" : "", + "required" : true + } + }, + "transactional" : false, + "sinkConfig" : { } + } + ], "subscriptions" : [ ], "operations" : [ { @@ -423,11 +532,34 @@ INSERT INTO `default_catalog`.`default_database`.`Product_2` "mcpMethod" : "TOOL", "restMethod" : "GET", "uriTemplate" : "queries/Product{?productid,offset,limit}" + }, + { + "function" : { + "name" : "AddMyEvents", + "parameters" : { + "type" : "object", + "properties" : { + "payload" : { + "type" : "string" + } + }, + "required" : [ ] + } + }, + "format" : "JSON", + "apiQuery" : { + "query" : "mutation MyEvents($payload: String) {\nMyEvents(event: { payload: $payload }) {\nid\npayload\ntimestamp\n}\n\n}", + "queryName" : "MyEvents", + "operationType" : "MUTATION" + }, + "mcpMethod" : "TOOL", + "restMethod" : "POST", + "uriTemplate" : "mutations/MyEvents" } ], "schema" : { "type" : "string", - "schema" : "type Customer {\n customerid: Long!\n email: String!\n name: String!\n lastUpdated: Long!\n timestamp: DateTime!\n}\n\n\"An RFC-3339 compliant Full Date Scalar\"\nscalar Date\n\n\"A DateTime scalar that handles both full RFC3339 and shorter timestamp formats\"\nscalar DateTime\n\n\"A JSON scalar\"\nscalar JSON\n\n\"24-hour clock time value string in the format `hh:mm:ss` or `hh:mm:ss.sss`.\"\nscalar LocalTime\n\n\"A 64-bit signed integer\"\nscalar Long\n\ntype Product {\n productid: Long!\n name: String!\n description: String!\n category: String!\n}\n\ntype Query {\n Customer(limit: Int = 10, offset: Int = 0): [Customer!]\n Product(productid: Long!, limit: Int = 10, offset: Int = 0): [Product!]\n}\n\nenum _McpMethodType {\n NONE\n TOOL\n RESOURCE\n}\n\nenum _RestMethodType {\n NONE\n GET\n POST\n}\n\ndirective @api(mcp: _McpMethodType, rest: _RestMethodType, uri: String) on QUERY | MUTATION | FIELD_DEFINITION\n" + "schema" : "type Customer {\n customerid: Long!\n email: String!\n name: String!\n lastUpdated: Long!\n timestamp: DateTime!\n}\n\n\"An RFC-3339 compliant Full Date Scalar\"\nscalar Date\n\n\"A DateTime scalar that handles both full RFC3339 and shorter timestamp formats\"\nscalar DateTime\n\n\"A JSON scalar\"\nscalar JSON\n\n\"24-hour clock time value string in the format `hh:mm:ss` or `hh:mm:ss.sss`.\"\nscalar LocalTime\n\n\"A 64-bit signed integer\"\nscalar Long\n\ntype Mutation {\n MyEvents(event: MyEventsInput!): MyEventsResultOutput!\n}\n\ninput MyEventsInput {\n payload: String\n}\n\ntype MyEventsResultOutput {\n id: String!\n payload: String\n timestamp: DateTime!\n}\n\ntype Product {\n productid: Long!\n name: String!\n description: String!\n category: String!\n}\n\ntype Query {\n Customer(limit: Int = 10, offset: Int = 0): [Customer!]\n Product(productid: Long!, limit: Int = 10, offset: Int = 0): [Product!]\n}\n\nenum _McpMethodType {\n NONE\n TOOL\n RESOURCE\n}\n\nenum _RestMethodType {\n NONE\n GET\n POST\n}\n\ndirective @api(mcp: _McpMethodType, rest: _RestMethodType, uri: String) on QUERY | MUTATION | FIELD_DEFINITION\n" } } }