diff --git a/sqrl-planner/src/main/java/com/datasqrl/error/ErrorCode.java b/sqrl-planner/src/main/java/com/datasqrl/error/ErrorCode.java index 186cbc46c..d76e77f64 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/error/ErrorCode.java +++ b/sqrl-planner/src/main/java/com/datasqrl/error/ErrorCode.java @@ -53,7 +53,8 @@ public enum ErrorCode implements ErrorLabel { NOT_YET_IMPLEMENTED, NO_API_ENDPOINTS, MISSING_SORT_COLUMN, - ROWTIME_IS_NULLABLE; + ROWTIME_IS_NULLABLE, + UNBOUNDED_BATCH_SOURCE; @Override public String getLabel() { diff --git a/sqrl-planner/src/main/java/com/datasqrl/error/ValidationExceptionHandler.java b/sqrl-planner/src/main/java/com/datasqrl/error/ValidationExceptionHandler.java new file mode 100644 index 000000000..9a0c92dba --- /dev/null +++ b/sqrl-planner/src/main/java/com/datasqrl/error/ValidationExceptionHandler.java @@ -0,0 +1,40 @@ +/* + * Copyright © 2021 DataSQRL (contact@datasqrl.com) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datasqrl.error; + +import com.google.auto.service.AutoService; +import org.apache.flink.table.api.ValidationException; + +@AutoService(ErrorHandler.class) +public class ValidationExceptionHandler implements ErrorHandler { + + @Override + public ErrorMessage handle(ValidationException e, ErrorLocation baseLocation) { + var errCode = ErrorCode.GENERIC; + var msg = e.getMessage(); + + if (msg != null && msg.contains("table source is unbounded")) { + errCode = ErrorCode.UNBOUNDED_BATCH_SOURCE; + } + + return new ErrorMessage.Implementation(errCode, msg, baseLocation, ErrorMessage.Severity.FATAL); + } + + @Override + public Class getHandleClass() { + return ValidationException.class; + } +} diff --git a/sqrl-planner/src/main/java/com/datasqrl/planner/Sqrl2FlinkSQLTranslator.java b/sqrl-planner/src/main/java/com/datasqrl/planner/Sqrl2FlinkSQLTranslator.java index 477cd6e7b..33bdfb36d 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/planner/Sqrl2FlinkSQLTranslator.java +++ b/sqrl-planner/src/main/java/com/datasqrl/planner/Sqrl2FlinkSQLTranslator.java @@ -740,10 +740,16 @@ public Optional createTable( Optional mutationBuilder, SchemaLoader schemaLoader, HintsAndDoc hintsAndDoc) { + var result = addTable(Function.identity(), tableDefinition, schemaLoader, mutationBuilder); hintsAndDoc = updateDocumentationFromLike(result, hintsAndDoc); - if (result.isSourceTable()) return Optional.of(addSourceTable(result, hintsAndDoc)); - else return Optional.empty(); + + if (!result.isSourceTable() || hintsAndDoc.hints().isNoSource()) { + return Optional.empty(); + } + + var srcTable = addSourceTable(result, hintsAndDoc); + return Optional.of(srcTable); } private HintsAndDoc updateDocumentationFromLike( diff --git a/sqrl-planner/src/main/java/com/datasqrl/planner/hint/NoSourceHint.java b/sqrl-planner/src/main/java/com/datasqrl/planner/hint/NoSourceHint.java new file mode 100644 index 000000000..24e5dd3b3 --- /dev/null +++ b/sqrl-planner/src/main/java/com/datasqrl/planner/hint/NoSourceHint.java @@ -0,0 +1,44 @@ +/* + * Copyright © 2021 DataSQRL (contact@datasqrl.com) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package com.datasqrl.planner.hint; + +import com.datasqrl.planner.parser.ParsedObject; +import com.datasqrl.planner.parser.SqrlHint; +import com.google.auto.service.AutoService; + +/** Makes sure a table will not be registered as a source. */ +public class NoSourceHint extends PlannerHint { + + public static final String HINT_NAME = "no_source"; + + protected NoSourceHint(ParsedObject source) { + super(source, Type.DAG); + } + + @AutoService(Factory.class) + public static class NoSourceHintFactory implements Factory { + + @Override + public PlannerHint create(ParsedObject source) { + return new NoSourceHint(source); + } + + @Override + public String getName() { + return HINT_NAME; + } + } +} 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 1ac0027cd..4fe36d585 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 @@ -76,6 +76,10 @@ public boolean isWorkload() { return getHint(WorkloadHint.class).isPresent(); } + public boolean isNoSource() { + return getHint(NoSourceHint.class).isPresent(); + } + public Optional getHint(Class hintClass) { return getHints(hintClass).findFirst(); } diff --git a/sqrl-planner/src/main/resources/com/datasqrl/error/errorCodes/unbounded_batch_source.md b/sqrl-planner/src/main/resources/com/datasqrl/error/errorCodes/unbounded_batch_source.md new file mode 100644 index 000000000..412d00be0 --- /dev/null +++ b/sqrl-planner/src/main/resources/com/datasqrl/error/errorCodes/unbounded_batch_source.md @@ -0,0 +1 @@ +In case this table is solely used as a sink, apply the /*+ no_source */ hint to avoid marking it as a source. 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 8e1a897d5..f7040767a 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("mutationNestedTypeTest.sqrl"); + var script = SCRIPT_DIR.resolve("noSourceHint.sqrl"); scripts(script); } diff --git a/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/UseCaseCompileTest.java b/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/UseCaseCompileTest.java index 6f023d3f9..cda371961 100644 --- a/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/UseCaseCompileTest.java +++ b/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/UseCaseCompileTest.java @@ -66,7 +66,7 @@ void testUseCase(Path packageFile) { @Test @Disabled("Intended for manual usage") void runTestCaseByName() { - var pkg = USECASE_DIR.resolve("banking").resolve("package.json"); + var pkg = USECASE_DIR.resolve("batch-to-kafka-compile").resolve("package-fail.json"); UseCaseTestHelper.testUseCase( snapshotExtension, getClass(), diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/noSourceHint.sqrl b/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/noSourceHint.sqrl new file mode 100644 index 000000000..281c0bbbb --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/noSourceHint.sqrl @@ -0,0 +1,17 @@ +IMPORT ecommerceTs.orders; + +MaterializedOrders := SELECT * FROM Orders; + +/*+ no_source */ +CREATE TABLE KafkaSink ( + PRIMARY KEY (id, `time`) NOT ENFORCED +) WITH ( + 'connector' = 'upsert-kafka', + 'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}', + 'key.format' = 'json', + 'value.format' = 'json', + 'value.fields-include' = 'ALL', + 'topic' = 'customers' +) LIKE Orders__schema; + +EXPORT MaterializedOrders TO KafkaSink; diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/noSourceHint.txt b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/noSourceHint.txt new file mode 100644 index 000000000..3185f2b13 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/noSourceHint.txt @@ -0,0 +1,412 @@ +>>>pipeline_explain.txt +=== KafkaSink +ID: KafkaSink +Type: export +Stage: flink +Connector: upsert-kafka +--- +Inputs: + - default_catalog.default_database.MaterializedOrders + +=== MaterializedOrders +ID: default_catalog.default_database.MaterializedOrders +Type: stream +Stage: flink +Primary key: id, time +Timestamp: time +Row count: ~1e8 +--- +Schema: + - id: BIGINT NOT NULL + - customerid: BIGINT NOT NULL + - time: TIMESTAMP_LTZ(3) *ROWTIME* NOT NULL + - entries: RecordType:peek_no_expand(BIGINT NOT NULL productid, BIGINT NOT NULL quantity, DOUBLE NOT NULL unit_price, DOUBLE discount) NOT NULL ARRAY NOT NULL +Inputs: + - default_catalog.default_database.Orders +Annotations: + - stream-root: Orders +Plan: +LogicalProject(id=[$0], customerid=[$1], time=[$2], entries=[$3]) + LogicalTableScan(table=[[default_catalog, default_database, Orders]]) +SQL: +CREATE VIEW `MaterializedOrders` AS SELECT * FROM Orders; + +=== Orders +ID: default_catalog.default_database.Orders +Type: stream +Stage: flink +Primary key: id, time +Timestamp: time +Row count: ~1e8 +--- +Schema: + - id: BIGINT NOT NULL + - customerid: BIGINT NOT NULL + - time: TIMESTAMP_LTZ(3) *ROWTIME* NOT NULL + - entries: RecordType:peek_no_expand(BIGINT NOT NULL productid, BIGINT NOT NULL quantity, DOUBLE NOT NULL unit_price, DOUBLE discount) NOT NULL ARRAY NOT NULL +Inputs: + - default_catalog.default_database.Orders__base +Annotations: + - features: DENORMALIZE (feature) + - stream-root: Orders +Plan: +LogicalWatermarkAssigner(rowtime=[time], watermark=[-($2, 1:INTERVAL SECOND)]) + LogicalTableScan(table=[[default_catalog, default_database, Orders]]) +SQL: +CREATE TEMPORARY TABLE `Orders__schema` ( + `id` BIGINT NOT NULL, + `customerid` BIGINT NOT NULL, + `time` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL, + `entries` ROW(`productid` BIGINT NOT NULL, `quantity` BIGINT NOT NULL, `unit_price` DOUBLE NOT NULL, `discount` DOUBLE) NOT NULL ARRAY NOT NULL +) +WITH ( + 'connector' = 'datagen' +); +CREATE TABLE `Orders` ( + PRIMARY KEY (`id`, `time`) NOT ENFORCED, + WATERMARK FOR `time` AS `time` - INTERVAL '0.001' SECOND +) +WITH ( + 'connector' = 'filesystem', + 'format' = 'flexible-json', + 'path' = 'file:/mock', + 'source.monitor-interval' = '10 sec' +) +LIKE `Orders__schema` +>>>flink-sql-no-functions.sql +CREATE TEMPORARY TABLE `Orders__schema` ( + `id` BIGINT NOT NULL, + `customerid` BIGINT NOT NULL, + `time` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL, + `entries` ROW(`productid` BIGINT NOT NULL, `quantity` BIGINT NOT NULL, `unit_price` DOUBLE NOT NULL, `discount` DOUBLE) NOT NULL ARRAY NOT NULL +) +WITH ( + 'connector' = 'datagen' +); +CREATE TABLE `Orders` ( + PRIMARY KEY (`id`, `time`) NOT ENFORCED, + WATERMARK FOR `time` AS `time` - INTERVAL '0.001' SECOND +) +WITH ( + 'connector' = 'filesystem', + 'format' = 'flexible-json', + 'path' = 'file:/mock', + 'source.monitor-interval' = '10 sec' +) +LIKE `Orders__schema`; +CREATE VIEW `MaterializedOrders` +AS +SELECT * +FROM `Orders`; +CREATE TABLE `KafkaSink` ( + PRIMARY KEY (`id`, `time`) NOT ENFORCED +) +WITH ( + 'connector' = 'upsert-kafka', + 'key.format' = 'json', + 'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}', + 'topic' = 'customers', + 'value.fields-include' = 'ALL', + 'value.format' = 'json' +) +LIKE `Orders__schema`; +CREATE TABLE `MaterializedOrders_1` ( + `id` BIGINT NOT NULL, + `customerid` BIGINT NOT NULL, + `time` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL, + `entries` RAW('com.datasqrl.flinkrunner.stdlib.json.FlinkJsonType', 'AERjb20uZGF0YXNxcmwuZmxpbmtydW5uZXIuc3RkbGliLmpzb24uRmxpbmtKc29uVHlwZVNlcmlhbGl6ZXJTbmFwc2hvdAAAAAM='), + PRIMARY KEY (`id`, `time`) NOT ENFORCED +) +WITH ( + 'connector' = 'jdbc-sqrl', + 'driver' = 'org.postgresql.Driver', + 'password' = '${POSTGRES_PASSWORD}', + 'sink.on-conflict.action' = 'IGNORE', + 'table-name' = 'MaterializedOrders_1', + 'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}', + 'username' = '${POSTGRES_USERNAME}' +); +CREATE TABLE `Orders_2` ( + `id` BIGINT NOT NULL, + `customerid` BIGINT NOT NULL, + `time` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL, + `entries` RAW('com.datasqrl.flinkrunner.stdlib.json.FlinkJsonType', 'AERjb20uZGF0YXNxcmwuZmxpbmtydW5uZXIuc3RkbGliLmpzb24uRmxpbmtKc29uVHlwZVNlcmlhbGl6ZXJTbmFwc2hvdAAAAAM='), + PRIMARY KEY (`id`, `time`) NOT ENFORCED +) +WITH ( + 'connector' = 'jdbc-sqrl', + 'driver' = 'org.postgresql.Driver', + 'password' = '${POSTGRES_PASSWORD}', + 'sink.on-conflict.action' = 'IGNORE', + 'table-name' = 'Orders_2', + 'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}', + 'username' = '${POSTGRES_USERNAME}' +); +EXECUTE STATEMENT SET BEGIN +INSERT INTO `default_catalog`.`default_database`.`KafkaSink` +SELECT * +FROM `default_catalog`.`default_database`.`MaterializedOrders` +; +INSERT INTO `default_catalog`.`default_database`.`MaterializedOrders_1` +SELECT `id`, `customerid`, `time`, `to_jsonb`(`entries`) AS `entries` +FROM `default_catalog`.`default_database`.`MaterializedOrders` +; +INSERT INTO `default_catalog`.`default_database`.`Orders_2` +SELECT `id`, `customerid`, `time`, `to_jsonb`(`entries`) AS `entries` +FROM `default_catalog`.`default_database`.`Orders` +; +END +>>>kafka.json +{ + "topics" : [ ], + "testRunnerTopics" : [ ] +} +>>>postgres.json +{ + "statements" : [ + { + "name" : "MaterializedOrders_1", + "type" : "TABLE", + "sql" : "CREATE TABLE IF NOT EXISTS \"MaterializedOrders_1\" (\"id\" BIGINT NOT NULL, \"customerid\" BIGINT NOT NULL, \"time\" TIMESTAMP WITH TIME ZONE NOT NULL, \"entries\" JSONB, PRIMARY KEY (\"id\",\"time\"))", + "fields" : [ + { + "name" : "id", + "type" : "BIGINT", + "nullable" : false + }, + { + "name" : "customerid", + "type" : "BIGINT", + "nullable" : false + }, + { + "name" : "time", + "type" : "TIMESTAMP WITH TIME ZONE", + "nullable" : false + }, + { + "name" : "entries", + "type" : "JSONB", + "nullable" : true + } + ], + "primaryKey" : [ + "id", + "time" + ], + "partitionKey" : [ ], + "partitionType" : "NONE", + "numPartitions" : 0, + "ttl" : 0.0 + }, + { + "name" : "Orders_2", + "type" : "TABLE", + "sql" : "CREATE TABLE IF NOT EXISTS \"Orders_2\" (\"id\" BIGINT NOT NULL, \"customerid\" BIGINT NOT NULL, \"time\" TIMESTAMP WITH TIME ZONE NOT NULL, \"entries\" JSONB, PRIMARY KEY (\"id\",\"time\"))", + "fields" : [ + { + "name" : "id", + "type" : "BIGINT", + "nullable" : false + }, + { + "name" : "customerid", + "type" : "BIGINT", + "nullable" : false + }, + { + "name" : "time", + "type" : "TIMESTAMP WITH TIME ZONE", + "nullable" : false + }, + { + "name" : "entries", + "type" : "JSONB", + "nullable" : true + } + ], + "primaryKey" : [ + "id", + "time" + ], + "partitionKey" : [ ], + "partitionType" : "NONE", + "numPartitions" : 0, + "ttl" : 0.0 + }, + { + "name" : "MaterializedOrders", + "type" : "VIEW", + "sql" : "CREATE OR REPLACE VIEW \"MaterializedOrders\"(\"id\", \"customerid\", \"time\", \"entries\") AS SELECT *\nFROM \"MaterializedOrders_1\"", + "fields" : [ + { + "name" : "id", + "type" : "BIGINT", + "nullable" : false + }, + { + "name" : "customerid", + "type" : "BIGINT", + "nullable" : false + }, + { + "name" : "time", + "type" : "TIMESTAMP WITH TIME ZONE", + "nullable" : false + }, + { + "name" : "entries", + "type" : "JSONB", + "nullable" : true + } + ] + }, + { + "name" : "Orders", + "type" : "VIEW", + "sql" : "CREATE OR REPLACE VIEW \"Orders\"(\"id\", \"customerid\", \"time\", \"entries\") AS SELECT *\nFROM \"Orders_2\"", + "fields" : [ + { + "name" : "id", + "type" : "BIGINT", + "nullable" : false + }, + { + "name" : "customerid", + "type" : "BIGINT", + "nullable" : false + }, + { + "name" : "time", + "type" : "TIMESTAMP WITH TIME ZONE", + "nullable" : false + }, + { + "name" : "entries", + "type" : "JSONB", + "nullable" : true + } + ] + } + ], + "standaloneExtensionStatements" : [ ] +} +>>>vertx.json +{ + "models" : { + "v1" : { + "queries" : [ + { + "type" : "args", + "parentType" : "Query", + "fieldName" : "MaterializedOrders", + "exec" : { + "arguments" : [ + { + "type" : "variable", + "path" : "limit" + }, + { + "type" : "variable", + "path" : "offset" + } + ], + "query" : { + "type" : "SqlQuery", + "sql" : "SELECT *\nFROM \"MaterializedOrders_1\"", + "parameters" : [ ], + "pagination" : "LIMIT_AND_OFFSET", + "cacheDurationMs" : 0, + "database" : "POSTGRES" + } + } + }, + { + "type" : "args", + "parentType" : "Query", + "fieldName" : "Orders", + "exec" : { + "arguments" : [ + { + "type" : "variable", + "path" : "limit" + }, + { + "type" : "variable", + "path" : "offset" + } + ], + "query" : { + "type" : "SqlQuery", + "sql" : "SELECT *\nFROM \"Orders_2\"", + "parameters" : [ ], + "pagination" : "LIMIT_AND_OFFSET", + "cacheDurationMs" : 0, + "database" : "POSTGRES" + } + } + } + ], + "mutations" : [ ], + "subscriptions" : [ ], + "operations" : [ + { + "function" : { + "name" : "GetMaterializedOrders", + "parameters" : { + "type" : "object", + "properties" : { + "offset" : { + "type" : "integer" + }, + "limit" : { + "type" : "integer" + } + }, + "required" : [ ] + } + }, + "format" : "JSON", + "apiQuery" : { + "query" : "query MaterializedOrders($limit: Int = 10, $offset: Int = 0) {\nMaterializedOrders(limit: $limit, offset: $offset) {\nid\ncustomerid\ntime\nentries {\nproductid\nquantity\nunit_price\ndiscount\n}\n}\n\n}", + "queryName" : "MaterializedOrders", + "operationType" : "QUERY" + }, + "mcpMethod" : "TOOL", + "restMethod" : "GET", + "uriTemplate" : "queries/MaterializedOrders{?offset,limit}" + }, + { + "function" : { + "name" : "GetOrders", + "parameters" : { + "type" : "object", + "properties" : { + "offset" : { + "type" : "integer" + }, + "limit" : { + "type" : "integer" + } + }, + "required" : [ ] + } + }, + "format" : "JSON", + "apiQuery" : { + "query" : "query Orders($limit: Int = 10, $offset: Int = 0) {\nOrders(limit: $limit, offset: $offset) {\nid\ncustomerid\ntime\nentries {\nproductid\nquantity\nunit_price\ndiscount\n}\n}\n\n}", + "queryName" : "Orders", + "operationType" : "QUERY" + }, + "mcpMethod" : "TOOL", + "restMethod" : "GET", + "uriTemplate" : "queries/Orders{?offset,limit}" + } + ], + "schema" : { + "type" : "string", + "schema" : "\"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 Orders {\n id: Long!\n customerid: Long!\n time: DateTime!\n entries: [Orders_entriesOutput]!\n}\n\ntype Orders_entriesOutput {\n productid: Long!\n quantity: Long!\n unit_price: Float!\n discount: Float\n}\n\ntype Query {\n MaterializedOrders(limit: Int = 10, offset: Int = 0): [Orders!]\n Orders(limit: Int = 10, offset: Int = 0): [Orders!]\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" + } + } + } +} diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/batch-to-kafka-compile-package-fail.txt b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/batch-to-kafka-compile-package-fail.txt new file mode 100644 index 000000000..f9db69a08 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/batch-to-kafka-compile-package-fail.txt @@ -0,0 +1,6 @@ +[FATAL] Querying an unbounded table 'default_catalog.default_database.KafkaSink' in batch mode is not allowed. The table source is unbounded. +in script:sinks.sqrl [1:1]: +CREATE TABLE KafkaSink ( +^ +In case this table is solely used as a sink, apply the /*+ no_source */ hint to avoid marking it as a source. + diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/batch-to-kafka-compile-package.txt b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/batch-to-kafka-compile-package.txt new file mode 100644 index 000000000..fee52d106 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/batch-to-kafka-compile-package.txt @@ -0,0 +1,320 @@ +>>>inferred_schema.graphqls +type Customers { + id: Long! + first_name: String! + last_name: String! + email: String! + phone: String! + address: String! + date_of_birth: String! + updated_at: DateTime! +} + +"An RFC-3339 compliant Full Date Scalar" +scalar Date + +"A DateTime scalar that handles both full RFC3339 and shorter timestamp formats" +scalar DateTime + +"A JSON scalar" +scalar JSON + +"24-hour clock time value string in the format `hh:mm:ss` or `hh:mm:ss.sss`." +scalar LocalTime + +"A 64-bit signed integer" +scalar Long + +type Query { + Customers(limit: Int = 10, offset: Int = 0): [Customers!] + CustomersByBirth(limit: Int = 10, offset: Int = 0): [Customers!] +} + +enum _McpMethodType { + NONE + TOOL + RESOURCE +} + +enum _RestMethodType { + NONE + GET + POST +} + +directive @api(mcp: _McpMethodType, rest: _RestMethodType, uri: String) on QUERY | MUTATION | FIELD_DEFINITION + +>>>pipeline_explain.txt +=== KafkaSink +ID: KafkaSink +Type: export +Stage: flink +Connector: upsert-kafka +--- +Inputs: + - default_catalog.default_database.CustomersByBirth + +=== Customers +ID: default_catalog.default_database.Customers +Type: stream +Stage: flink +Primary key: id, updated_at +Timestamp: - +Row count: ~1e8 +--- +Schema: + - id: BIGINT NOT NULL + - first_name: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - last_name: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - email: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - phone: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - address: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - date_of_birth: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - updated_at: TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL +Inputs: + - default_catalog.default_database.Customers__base +Annotations: + - stream-root: Customers + +=== CustomersByBirth +ID: default_catalog.default_database.CustomersByBirth +Type: stream +Stage: flink +Primary key: id, updated_at +Timestamp: - +Row count: ~1e8 +--- +Schema: + - id: BIGINT NOT NULL + - first_name: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - last_name: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - email: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - phone: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - address: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - date_of_birth: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - updated_at: TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) NOT NULL +Inputs: + - default_catalog.default_database.Customers +Annotations: + - stream-root: Customers + - sort: [6 ASC-nulls-first] + +>>>flink-sql-no-functions.sql +CREATE TEMPORARY TABLE `Customers__schema` ( + `id` BIGINT NOT NULL, + `first_name` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `last_name` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `email` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `phone` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `address` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `date_of_birth` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `updated_at` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL +) +WITH ( + 'connector' = 'filesystem', + 'format' = 'flexible-json', + 'path' = '${DATA_PATH}/customers.jsonl' +); +CREATE TABLE `Customers` ( + PRIMARY KEY (`id`, `updated_at`) NOT ENFORCED +) +LIKE `Customers__schema`; +CREATE VIEW `CustomersByBirth` +AS +SELECT * +FROM `Customers`; +CREATE TABLE `KafkaSink` ( + `id` STRING NOT NULL, + `first_name` STRING, + `last_name` STRING, + `email` STRING, + `phone` STRING, + `address` STRING, + `date_of_birth` DATE, + `updated_at` TIMESTAMP_LTZ(3) NOT NULL, + PRIMARY KEY (`id`, `updated_at`) NOT ENFORCED +) +WITH ( + 'connector' = 'upsert-kafka', + 'key.format' = 'json', + 'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}', + 'topic' = 'customers', + 'value.fields-include' = 'ALL', + 'value.format' = 'json' +); +CREATE TABLE `Customers_1` ( + `id` BIGINT NOT NULL, + `first_name` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `last_name` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `email` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `phone` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `address` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `date_of_birth` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `updated_at` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL, + PRIMARY KEY (`id`, `updated_at`) NOT ENFORCED +) +WITH ( + 'connector' = 'jdbc-sqrl', + 'driver' = 'org.postgresql.Driver', + 'password' = '${POSTGRES_PASSWORD}', + 'sink.on-conflict.action' = 'IGNORE', + 'table-name' = 'Customers', + 'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}', + 'username' = '${POSTGRES_USERNAME}' +); +CREATE TABLE `CustomersByBirth_2` ( + `id` BIGINT NOT NULL, + `first_name` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `last_name` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `email` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `phone` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `address` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `date_of_birth` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `updated_at` TIMESTAMP(3) WITH LOCAL TIME ZONE NOT NULL, + PRIMARY KEY (`id`, `updated_at`) NOT ENFORCED +) +WITH ( + 'connector' = 'jdbc-sqrl', + 'driver' = 'org.postgresql.Driver', + 'password' = '${POSTGRES_PASSWORD}', + 'sink.on-conflict.action' = 'IGNORE', + 'table-name' = 'CustomersByBirth', + 'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}', + 'username' = '${POSTGRES_USERNAME}' +); +EXECUTE STATEMENT SET BEGIN +INSERT INTO `default_catalog`.`default_database`.`Customers_1` +SELECT * +FROM `default_catalog`.`default_database`.`Customers` +; +INSERT INTO `default_catalog`.`default_database`.`KafkaSink` +SELECT * +FROM `default_catalog`.`default_database`.`CustomersByBirth` +; +INSERT INTO `default_catalog`.`default_database`.`CustomersByBirth_2` +SELECT * +FROM `default_catalog`.`default_database`.`CustomersByBirth` +; +END +>>>postgres-schema.sql +CREATE TABLE IF NOT EXISTS "Customers" ("id" BIGINT NOT NULL, "first_name" TEXT NOT NULL, "last_name" TEXT NOT NULL, "email" TEXT NOT NULL, "phone" TEXT NOT NULL, "address" TEXT NOT NULL, "date_of_birth" TEXT NOT NULL, "updated_at" TIMESTAMP WITH TIME ZONE NOT NULL, PRIMARY KEY ("id","updated_at")); +CREATE TABLE IF NOT EXISTS "CustomersByBirth" ("id" BIGINT NOT NULL, "first_name" TEXT NOT NULL, "last_name" TEXT NOT NULL, "email" TEXT NOT NULL, "phone" TEXT NOT NULL, "address" TEXT NOT NULL, "date_of_birth" TEXT NOT NULL, "updated_at" TIMESTAMP WITH TIME ZONE NOT NULL, PRIMARY KEY ("id","updated_at")) +>>>vertx.json +{ + "models" : { + "v1" : { + "queries" : [ + { + "type" : "args", + "parentType" : "Query", + "fieldName" : "Customers", + "exec" : { + "arguments" : [ + { + "type" : "variable", + "path" : "limit" + }, + { + "type" : "variable", + "path" : "offset" + } + ], + "query" : { + "type" : "SqlQuery", + "sql" : "SELECT *\nFROM \"Customers\"", + "parameters" : [ ], + "pagination" : "LIMIT_AND_OFFSET", + "cacheDurationMs" : 0, + "database" : "POSTGRES" + } + } + }, + { + "type" : "args", + "parentType" : "Query", + "fieldName" : "CustomersByBirth", + "exec" : { + "arguments" : [ + { + "type" : "variable", + "path" : "limit" + }, + { + "type" : "variable", + "path" : "offset" + } + ], + "query" : { + "type" : "SqlQuery", + "sql" : "SELECT *\nFROM (SELECT \"id\", \"first_name\", \"last_name\", \"email\", \"phone\", \"address\", \"date_of_birth\", \"updated_at\"\n FROM \"CustomersByBirth\"\n ORDER BY \"date_of_birth\" NULLS FIRST) AS \"t\"", + "parameters" : [ ], + "pagination" : "LIMIT_AND_OFFSET", + "cacheDurationMs" : 0, + "database" : "POSTGRES" + } + } + } + ], + "mutations" : [ ], + "subscriptions" : [ ], + "operations" : [ + { + "function" : { + "name" : "GetCustomers", + "parameters" : { + "type" : "object", + "properties" : { + "offset" : { + "type" : "integer" + }, + "limit" : { + "type" : "integer" + } + }, + "required" : [ ] + } + }, + "format" : "JSON", + "apiQuery" : { + "query" : "query Customers($limit: Int = 10, $offset: Int = 0) {\nCustomers(limit: $limit, offset: $offset) {\nid\nfirst_name\nlast_name\nemail\nphone\naddress\ndate_of_birth\nupdated_at\n}\n\n}", + "queryName" : "Customers", + "operationType" : "QUERY" + }, + "mcpMethod" : "TOOL", + "restMethod" : "GET", + "uriTemplate" : "queries/Customers{?offset,limit}" + }, + { + "function" : { + "name" : "GetCustomersByBirth", + "parameters" : { + "type" : "object", + "properties" : { + "offset" : { + "type" : "integer" + }, + "limit" : { + "type" : "integer" + } + }, + "required" : [ ] + } + }, + "format" : "JSON", + "apiQuery" : { + "query" : "query CustomersByBirth($limit: Int = 10, $offset: Int = 0) {\nCustomersByBirth(limit: $limit, offset: $offset) {\nid\nfirst_name\nlast_name\nemail\nphone\naddress\ndate_of_birth\nupdated_at\n}\n\n}", + "queryName" : "CustomersByBirth", + "operationType" : "QUERY" + }, + "mcpMethod" : "TOOL", + "restMethod" : "GET", + "uriTemplate" : "queries/CustomersByBirth{?offset,limit}" + } + ], + "schema" : { + "type" : "string", + "schema" : "type Customers {\n id: Long!\n first_name: String!\n last_name: String!\n email: String!\n phone: String!\n address: String!\n date_of_birth: String!\n updated_at: 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 Query {\n Customers(limit: Int = 10, offset: Int = 0): [Customers!]\n CustomersByBirth(limit: Int = 10, offset: Int = 0): [Customers!]\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" + } + } + } +} diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/batch-to-kafka.sqrl b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/batch-to-kafka.sqrl new file mode 100644 index 000000000..0df59ce44 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/batch-to-kafka.sqrl @@ -0,0 +1,8 @@ +IMPORT connectors.sources.*; + +CustomersByBirth := SELECT * FROM Customers ORDER BY date_of_birth; + +{{#addNoSource}}/*+ no_source */{{/addNoSource}} +IMPORT connectors.sinks.*; + +EXPORT CustomersByBirth TO KafkaSink; diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/connectors/customers.jsonl b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/connectors/customers.jsonl new file mode 100644 index 000000000..fca84c900 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/connectors/customers.jsonl @@ -0,0 +1,10 @@ +{"id":"1","first_name":"John","last_name":"Doe","email":"johndoe@email.com","phone":"123-456-7890","address":"123 Main St","date_of_birth":"1990-01-01","updated_at":"2022-06-20T00:00:00Z"} +{"id":"2","first_name":"Jane","last_name":"Smith","email":"janesmith@email.com","phone":"234-567-8901","address":"456 Oak St","date_of_birth":"1985-02-14","updated_at":"2022-06-20T00:00:00Z"} +{"id":"3","first_name":"Michael","last_name":"Brown","email":"michaelbrown@email.com","phone":"345-678-9012","address":"789 Elm St","date_of_birth":"1978-09-30","updated_at":"2022-06-20T00:00:00Z"} +{"id":"4","first_name":"Emily","last_name":"Davis","email":"emilydavis@email.com","phone":"456-789-0123","address":"321 Pine St","date_of_birth":"1995-12-06","updated_at":"2022-06-20T00:00:00Z"} +{"id":"5","first_name":"David","last_name":"Johnson","email":"davidjohnson@email.com","phone":"567-890-1234","address":"654 Maple St","date_of_birth":"1982-07-22","updated_at":"2022-06-20T00:00:00Z"} +{"id":"6","first_name":"Sarah","last_name":"Williams","email":"sarahwilliams@email.com","phone":"678-901-2345","address":"987 Cedar St","date_of_birth":"1992-11-18","updated_at":"2022-06-20T00:00:00Z"} +{"id":"7","first_name":"James","last_name":"Jones","email":"jamesjones@email.com","phone":"789-012-3456","address":"147 Spruce St","date_of_birth":"1980-03-29","updated_at":"2022-06-20T00:00:00Z"} +{"id":"8","first_name":"Jennifer","last_name":"Taylor","email":"jennifertaylor@email.com","phone":"890-123-4567","address":"258 Birch St","date_of_birth":"1998-05-15","updated_at":"2022-06-20T00:00:00Z"} +{"id":"9","first_name":"Christopher","last_name":"Miller","email":"christophermiller@email.com","phone":"901-234-5678","address":"369 Willow St","date_of_birth":"1975-10-25","updated_at":"2022-06-20T00:00:00Z"} +{"id":"10","first_name":"Amy","last_name":"Wilson","email":"amywilson@email.com","phone":"012-345-6789","address":"480 Aspen St","date_of_birth":"1989-08-04","updated_at":"2022-06-20T00:00:00Z"} \ No newline at end of file diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/connectors/sinks.sqrl b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/connectors/sinks.sqrl new file mode 100644 index 000000000..f61943c1e --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/connectors/sinks.sqrl @@ -0,0 +1,18 @@ +CREATE TABLE KafkaSink ( + id STRING NOT NULL, + first_name STRING, + last_name STRING, + email STRING, + phone STRING, + address STRING, + date_of_birth DATE, + updated_at TIMESTAMP_LTZ(3) NOT NULL, + PRIMARY KEY(id, updated_at) NOT ENFORCED +) WITH ( + 'connector' = 'upsert-kafka', + 'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}', + 'key.format' = 'json', + 'value.format' = 'json', + 'value.fields-include' = 'ALL', + 'topic' = 'customers' +); diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/connectors/sources.sqrl b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/connectors/sources.sqrl new file mode 100644 index 000000000..7027c3faa --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/connectors/sources.sqrl @@ -0,0 +1,3 @@ +CREATE TABLE Customers ( + PRIMARY KEY (id, updated_at) NOT ENFORCED +) LIKE customers.jsonl; diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/package-fail.json b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/package-fail.json new file mode 100644 index 000000000..8bd5add12 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/package-fail.json @@ -0,0 +1,20 @@ +{ + "version": "1", + "enabled-engines": ["vertx", "postgres", "flink"], + "script": { + "main": "batch-to-kafka.sqrl", + "config": { + "addNoSource": false + } + }, + "engines": { + "flink" : { + "config": { + "execution.runtime-mode": "batch" + } + } + }, + "test-runner": { + "delay-sec": -1 + } +} diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/package.json b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/package.json new file mode 100644 index 000000000..eabc71e9d --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/batch-to-kafka-compile/package.json @@ -0,0 +1,20 @@ +{ + "version": "1", + "enabled-engines": ["vertx", "postgres", "flink"], + "script": { + "main": "batch-to-kafka.sqrl", + "config": { + "addNoSource": true + } + }, + "engines": { + "flink" : { + "config": { + "execution.runtime-mode": "batch" + } + } + }, + "test-runner": { + "delay-sec": -1 + } +}