From fcf07ae2b16e6c4af138a72f2d05f707b8898240 Mon Sep 17 00:00:00 2001 From: Ferenc Csaky Date: Wed, 28 Oct 2026 15:55:45 +0100 Subject: [PATCH] Build mutation message keys after filling in the uuid and auth columns https://github.com/DataSQRL/sqrl/issues/2511 Signed-off-by: Ferenc Csaky --- agent/skills/implement-sqrl/SKILL.md | 1 + documentation/docs/sqrl-language.md | 2 +- .../planner/Sqrl2FlinkSQLTranslator.java | 26 ++ .../server/MutationConfigurationImpl.java | 36 +- .../server/MutationConfigurationImplTest.java | 78 ++++ ...itionedKafkaMutationTimestampKey-fail.sqrl | 6 + ...titionedKafkaMutationTimestampKey-fail.txt | 6 + .../kafka-message-keys-package.txt | 418 +++++++++++++++++- .../kafka-message-keys.sqrl | 41 ++ .../snapshots/mutation4.snapshot | 16 + .../snapshots/query_alerts.snapshot | 17 + .../tests/mutation4.graphql | 14 + .../tests/query_alerts.graphql | 6 + 13 files changed, 644 insertions(+), 23 deletions(-) create mode 100644 sqrl-server/sqrl-server-vertx-base/src/test/java/com/datasqrl/server/MutationConfigurationImplTest.java create mode 100644 sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/partitionedKafkaMutationTimestampKey-fail.sqrl create mode 100644 sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/partitionedKafkaMutationTimestampKey-fail.txt create mode 100644 sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/snapshots/mutation4.snapshot create mode 100644 sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/snapshots/query_alerts.snapshot create mode 100644 sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/tests/mutation4.graphql create mode 100644 sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/tests/query_alerts.graphql diff --git a/agent/skills/implement-sqrl/SKILL.md b/agent/skills/implement-sqrl/SKILL.md index 9999ca8cff..14c9c77de0 100644 --- a/agent/skills/implement-sqrl/SKILL.md +++ b/agent/skills/implement-sqrl/SKILL.md @@ -141,6 +141,7 @@ The internal table does not need `WATERMARK` declaration. SQRL generates the wat Declare a `WATERMARK` explicitely on an internal table only when the user asks for one. With `engine(kafka)`, `PARTITIONED BY (col, ...)` on an internal table without a `PRIMARY KEY` keys each record by those columns, both the records the server writes for mutations and the records the pipeline exports into the table, so records with the same key stay on one Kafka partition in order. +The key columns can be `uuid` or `auth` metadata columns, but not the `timestamp` metadata column. The Flink source ignores that key and reads every column from the message value, so adding `PARTITIONED BY` to an existing internal table is safe: records written before the table had a key are still read, also when the topic is read again from the earliest offset. This relies on the default `kafka-safe` connector. Do not switch internal Kafka tables to the plain `kafka` connector, which drops records that have no key. Use `PARTITIONED BY` when records only need a Kafka key for partitioning and ordering. A `PRIMARY KEY` also keys the records, but it makes the table an upsert table (`VERSIONED_STATE`), and its upsert source drops records that were written without a key. diff --git a/documentation/docs/sqrl-language.md b/documentation/docs/sqrl-language.md index f93875f8c8..cf29965850 100644 --- a/documentation/docs/sqrl-language.md +++ b/documentation/docs/sqrl-language.md @@ -113,7 +113,7 @@ Internal tables support `kafka` and `iceberg` engines. | Watermark spec | **generated** | **required** | | Primary key | *Unenforced* upsert semantics | Same as Flink | -With the `kafka` engine, `PARTITIONED BY (col, ...)` on an internal table without a primary key sets the Kafka message key of each record, both for the records that the server writes for a mutation and for the records that the pipeline exports into the table. Records with the same key go to the same Kafka partition, so their order is kept. The clause only declares the key; the table is not partitioned in Flink. +With the `kafka` engine, `PARTITIONED BY (col, ...)` on an internal table without a primary key sets the Kafka message key of each record, both for the records that the server writes for a mutation and for the records that the pipeline exports into the table. Records with the same key go to the same Kafka partition, so their order is kept. The clause only declares the key; the table is not partitioned in Flink. The key can contain `METADATA FROM 'uuid'` and `METADATA FROM 'auth.'` columns, which the server fills in before it writes the record, but not the `METADATA FROM 'timestamp'` column, which is only known after the record is written. The Flink source of the table reads every column from the message value and ignores the key. Adding `PARTITIONED BY` to an existing table therefore keeps the records written before the table had a key, also when the pipeline reads the topic again from the earliest offset. This relies on the default `kafka-safe` connector; the plain `kafka` connector drops records that have no key. A table with a primary key is keyed by its primary key instead, and its upsert source drops records that were written without a key. 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 aaa37cf773..222a0ced8a 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/planner/Sqrl2FlinkSQLTranslator.java +++ b/sqrl-planner/src/main/java/com/datasqrl/planner/Sqrl2FlinkSQLTranslator.java @@ -121,6 +121,7 @@ import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl; import org.apache.flink.table.api.internal.TableResultInternal; import org.apache.flink.table.catalog.CatalogManager; +import org.apache.flink.table.catalog.Column; import org.apache.flink.table.catalog.Column.ComputedColumn; import org.apache.flink.table.catalog.Column.MetadataColumn; import org.apache.flink.table.catalog.Column.PhysicalColumn; @@ -953,6 +954,7 @@ Note, that this requires we replace the table with the actual table (and the cor .orElse(PrimaryKeyMap.UNDEFINED); // Finish building mutation query by building input and output types from table schema if (mutationBld != null) { + validateMutationPartitionKey(flinkSchema, mutationBld.build().getPartitionKey()); var fields = convertSchema2RelDataType(flinkSchema); var inputType = CalciteUtil.getRelTypeBuilder(typeFactory); var outputType = CalciteUtil.getRelTypeBuilder(typeFactory); @@ -1175,4 +1177,28 @@ private static void validateMutationHints( + " Supported engines: \"kafka\", \"iceberg\"."); } } + + /** + * Partition columns must be physical, which includes the 'uuid' and 'auth' metadata columns the + * server fills in. Flink does not check this for engines that remove the partition from the + * table, e.g. Kafka uses it as the message key, which the server builds before the record is + * written. + */ + private static void validateMutationPartitionKey( + ResolvedSchema schema, List partitionKey) { + + for (var colName : partitionKey) { + if (!schema.getColumn(colName).map(Column::isPhysical).orElse(false)) { + var availableCols = + schema.getColumns().stream() + .filter(Column::isPhysical) + .map(Column::getName) + .collect(Collectors.joining(", ")); + + throw new NoLocationStatementParserException( + "Invalid PARTITIONED BY column [%s]: partition columns must be regular columns or 'uuid' or 'auth' metadata columns. Available columns: [%s]" + .formatted(colName, availableCols)); + } + } + } } diff --git a/sqrl-server/sqrl-server-vertx-base/src/main/java/com/datasqrl/server/MutationConfigurationImpl.java b/sqrl-server/sqrl-server-vertx-base/src/main/java/com/datasqrl/server/MutationConfigurationImpl.java index 46af81fd78..25028af9ac 100644 --- a/sqrl-server/sqrl-server-vertx-base/src/main/java/com/datasqrl/server/MutationConfigurationImpl.java +++ b/sqrl-server/sqrl-server-vertx-base/src/main/java/com/datasqrl/server/MutationConfigurationImpl.java @@ -23,6 +23,7 @@ import com.datasqrl.server.io.SinkProducer; import com.datasqrl.server.kafka.KafkaSinkProducer; import com.datasqrl.server.kafka.SaslRetryingSinkProducer; +import com.google.common.annotations.VisibleForTesting; import graphql.schema.DataFetcher; import graphql.schema.DataFetchingEnvironment; import io.vertx.core.CompositeFuture; @@ -122,7 +123,8 @@ interface ComputeInputColumns { Object compute(DataFetchingEnvironment env); } - private List getRecords( + @VisibleForTesting + static List getRecords( DataFetchingEnvironment env, List keyColumns, Map computedColumns) { @@ -140,20 +142,6 @@ private List getRecords( entries = List.of((Map) argument); } - // Construct keys if necessary (null if no key cols) - var keys = new Map[entries.size()]; - if (!keyColumns.isEmpty()) { - for (int i = 0; i < entries.size(); i++) { - var filtered = new LinkedHashMap(); - for (var key : keyColumns) { - if (entries.get(i).containsKey(key)) { - filtered.put(key, entries.get(i).get(key)); - } - } - keys[i] = filtered; - } - } - if (!computedColumns.isEmpty()) { // Add UUID for event for the computed uuid columns entries.forEach( @@ -166,14 +154,28 @@ private List getRecords( }); } + // Keys are built after the computed columns are added, since those can be key columns too var records = new ArrayList(); - for (int i = 0; i < entries.size(); i++) { - records.add(new SinkProducer.Record(keys[i], entries.get(i))); + for (var entry : entries) { + records.add(new SinkProducer.Record(createKey(entry, keyColumns), entry)); } return records; } + /** + * Keeps every key column, also when the input omits it, so that the key matches the one Flink + * writes for the same record (null if no key cols) + */ + private static Map createKey(Map entry, List keyColumns) { + if (keyColumns.isEmpty()) { + return null; + } + var key = new LinkedHashMap(); + keyColumns.forEach(col -> key.put(col, entry.get(col))); + return key; + } + private void completeWithResults( CompletableFuture cf, List results, boolean returnList) { if (returnList) { diff --git a/sqrl-server/sqrl-server-vertx-base/src/test/java/com/datasqrl/server/MutationConfigurationImplTest.java b/sqrl-server/sqrl-server-vertx-base/src/test/java/com/datasqrl/server/MutationConfigurationImplTest.java new file mode 100644 index 0000000000..1bb71806e6 --- /dev/null +++ b/sqrl-server/sqrl-server-vertx-base/src/test/java/com/datasqrl/server/MutationConfigurationImplTest.java @@ -0,0 +1,78 @@ +/* + * 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.server; + +import static java.nio.charset.StandardCharsets.UTF_8; +import static org.assertj.core.api.Assertions.assertThat; + +import com.datasqrl.server.MutationConfigurationImpl.ComputeInputColumns; +import com.datasqrl.server.io.SinkProducer; +import com.datasqrl.server.kafka.JsonSerializer; +import graphql.schema.DataFetchingEnvironment; +import graphql.schema.DataFetchingEnvironmentImpl; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.junit.jupiter.api.Test; + +class MutationConfigurationImplTest { + + private static final JsonSerializer KEY_SERIALIZER = new JsonSerializer(); + + @Test + void givenComputedKeyColumn_whenGettingRecords_thenKeyContainsComputedValue() { + var env = mutationOf(new HashMap<>(Map.of("sensorid", 1))); + Map computedColumns = Map.of("tenantId", e -> "tenant-1"); + + var records = + MutationConfigurationImpl.getRecords(env, List.of("sensorid", "tenantId"), computedColumns); + + assertThat(serializedKeys(records)) + .containsExactly("{\"sensorid\":1,\"tenantId\":\"tenant-1\"}"); + } + + @Test + void givenOmittedAndNullKeyColumn_whenGettingRecords_thenBothKeysContainNull() { + var explicitNull = new HashMap(); + explicitNull.put("zone", null); + var env = mutationOf(List.of(new HashMap<>(), explicitNull)); + + var records = MutationConfigurationImpl.getRecords(env, List.of("zone"), Map.of()); + + assertThat(serializedKeys(records)).containsExactly("{\"zone\":null}", "{\"zone\":null}"); + } + + @Test + void givenNoKeyColumns_whenGettingRecords_thenRecordHasNoKey() { + var env = mutationOf(new HashMap<>(Map.of("sensorid", 1))); + + var records = MutationConfigurationImpl.getRecords(env, List.of(), Map.of()); + + assertThat(records).singleElement().extracting(SinkProducer.Record::key).isNull(); + } + + private static DataFetchingEnvironment mutationOf(Object input) { + return DataFetchingEnvironmentImpl.newDataFetchingEnvironment() + .arguments(Map.of("event", input)) + .build(); + } + + private static List serializedKeys(List records) { + return records.stream() + .map(rec -> new String(KEY_SERIALIZER.serialize("topic", rec.key()), UTF_8)) + .toList(); + } +} diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/partitionedKafkaMutationTimestampKey-fail.sqrl b/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/partitionedKafkaMutationTimestampKey-fail.sqrl new file mode 100644 index 0000000000..12e180ae28 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/dagplanner/partitionedKafkaMutationTimestampKey-fail.sqrl @@ -0,0 +1,6 @@ +/*+engine(kafka) */ +CREATE TABLE Events ( + tenantId STRING NOT NULL METADATA FROM 'auth.tenant_id', + payload STRING NOT NULL, + event_time TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp' +) PARTITIONED BY (tenantId, event_time); diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/partitionedKafkaMutationTimestampKey-fail.txt b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/partitionedKafkaMutationTimestampKey-fail.txt new file mode 100644 index 0000000000..0d6fc95efa --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/partitionedKafkaMutationTimestampKey-fail.txt @@ -0,0 +1,6 @@ +[FATAL] Invalid PARTITIONED BY column [event_time]: partition columns must be regular columns or 'uuid' or 'auth' metadata columns. Available columns: [tenantId, payload] +in script:partitionedKafkaMutationTimestampKey-fail.sqrl [2:1]: +/*+engine(kafka) */ +CREATE TABLE Events ( +^ + diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/kafka-message-keys-package.txt b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/kafka-message-keys-package.txt index ac48c3fd31..cabb56f55e 100644 --- a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/kafka-message-keys-package.txt +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/kafka-message-keys-package.txt @@ -1,4 +1,21 @@ >>>inferred_schema.graphqls +input AlertInput { + region: String + writer: String! +} + +type AlertMessage { + writer: String! + message_key: String +} + +type AlertResultOutput { + region: String + alert_id: String! + writer: String! + event_time: DateTime! +} + "An RFC-3339 compliant Full Date Scalar" scalar Date @@ -16,6 +33,18 @@ type ExternalReadingResultOutput { event_time: DateTime! } +input ForwardedAlertInput { + region: String + writer: String! +} + +type ForwardedAlertResultOutput { + region: String + alert_id: String! + writer: String! + event_time: DateTime! +} + input ForwardedReadingInput { sensorid: Int! writer: String! @@ -37,12 +66,15 @@ scalar LocalTime scalar Long type Mutation { + Alert(event: AlertInput!): AlertResultOutput! ExternalReading(event: ExternalReadingInput!): ExternalReadingResultOutput! + ForwardedAlert(event: ForwardedAlertInput!): ForwardedAlertResultOutput! ForwardedReading(event: ForwardedReadingInput!): ForwardedReadingResultOutput! Reading(event: ReadingInput!): ReadingResultOutput! } type Query { + AlertMessage(limit: Int = 10, offset: Int = 0): [AlertMessage!] ReadingMessage(limit: Int = 10, offset: Int = 0): [ReadingMessage!] ReadingRecord(limit: Int = 10, offset: Int = 0): [ReadingRecord!] } @@ -84,6 +116,15 @@ enum _RestMethodType { directive @api(mcp: _McpMethodType, rest: _RestMethodType, uri: String) on QUERY | MUTATION | FIELD_DEFINITION >>>pipeline_explain.txt +=== Alert +ID: Alert +Type: export +Stage: flink +Connector: kafka-safe +--- +Inputs: + - default_catalog.default_database.ForwardedAlert + === Reading ID: Reading Type: export @@ -93,6 +134,22 @@ Connector: kafka-safe Inputs: - default_catalog.default_database.ForwardedReading +=== AlertMessage +ID: default_catalog.default_database.AlertMessage +Type: stream +Stage: flink +Primary key: - +Timestamp: - +Row count: ~1e8 +--- +Schema: + - writer: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - message_key: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" +Inputs: + - default_catalog.default_database._AlertMessage +Annotations: + - sort: [0 ASC-nulls-first] + === ExternalReading ID: default_catalog.default_database.ExternalReading Type: stream @@ -108,6 +165,24 @@ Schema: Inputs: - default_catalog.default_database.ExternalReading__base +=== ForwardedAlert +ID: default_catalog.default_database.ForwardedAlert +Type: stream +Stage: flink +Primary key: alert_id +Timestamp: event_time +Row count: ~1e8 +--- +Schema: + - region: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" + - alert_id: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - writer: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - event_time: TIMESTAMP_LTZ(3) *ROWTIME* NOT NULL +Inputs: + - default_catalog.default_database.ForwardedAlert__base +Annotations: + - stream-root: ForwardedAlert + === ForwardedReading ID: default_catalog.default_database.ForwardedReading Type: stream @@ -171,6 +246,23 @@ Inputs: Annotations: - sort: [0 ASC-nulls-first, 1 ASC-nulls-first] +=== _AlertMessage +ID: default_catalog.default_database._AlertMessage +Type: stream +Stage: flink +Primary key: - +Timestamp: event_time +Row count: ~1e8 +--- +Schema: + - message_key: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" + - region: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" + - alert_id: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - writer: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" NOT NULL + - event_time: TIMESTAMP_LTZ(3) *ROWTIME* NOT NULL +Inputs: + - default_catalog.default_database._AlertMessage__base + === _ReadingMessage ID: default_catalog.default_database._ReadingMessage Type: stream @@ -287,7 +379,84 @@ CREATE VIEW `ReadingMessage` AS SELECT `writer`, `sensorid`, `message_key` FROM `_ReadingMessage`; -CREATE TABLE `ReadingMessage_1` ( +CREATE TABLE `Alert` ( + `region` STRING, + `alert_id` STRING NOT NULL, + `writer` STRING NOT NULL, + `event_time` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp', + WATERMARK FOR `event_time` AS `event_time` - INTERVAL '0.0' SECOND +) +WITH ( + 'connector' = 'kafka-safe', + 'key.fields' = 'region;alert_id', + 'key.flexible-json.timestamp-format.standard' = 'ISO-8601', + 'key.format' = 'flexible-json', + 'properties.auto.offset.reset' = 'earliest', + 'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}', + 'properties.compression.type' = 'zstd', + 'properties.group.id' = '${KAFKA_GROUP_ID}', + 'scan.key-fields.read-from-value' = 'true', + 'topic' = 'Alert', + 'value.fields-include' = 'ALL', + 'value.flexible-json.timestamp-format.standard' = 'ISO-8601', + 'value.format' = 'flexible-json' +); +CREATE TABLE `ForwardedAlert` ( + `region` STRING, + `alert_id` STRING NOT NULL, + `writer` STRING NOT NULL, + `event_time` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp', + WATERMARK FOR `event_time` AS `event_time` - INTERVAL '0.0' SECOND +) +WITH ( + 'connector' = 'kafka-safe', + 'flexible-json.timestamp-format.standard' = 'ISO-8601', + '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' = 'ForwardedAlert' +); +CREATE TABLE `_AlertMessage` ( + `message_key` STRING, + `region` STRING, + `alert_id` STRING NOT NULL, + `writer` STRING NOT NULL, + `event_time` TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp', + WATERMARK FOR `event_time` AS `event_time` - INTERVAL '1' SECOND +) +WITH ( + 'connector' = 'kafka-safe', + 'key.fields' = 'message_key', + 'key.format' = 'raw', + 'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}', + 'properties.group.id' = 'alert-message', + 'scan.startup.mode' = 'earliest-offset', + 'topic' = 'Alert', + 'value.fields-include' = 'EXCEPT_KEY', + 'value.format' = 'flexible-json' +); +CREATE VIEW `AlertMessage` +AS +SELECT `writer`, REPLACE(`message_key`, `alert_id`, '') AS `message_key` +FROM `_AlertMessage`; +CREATE TABLE `AlertMessage_1` ( + `writer` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, + `message_key` VARCHAR(2147483647) CHARACTER SET `UTF-16LE`, + `__pk_hash` CHAR(32) CHARACTER SET `UTF-16LE`, + PRIMARY KEY (`__pk_hash`) NOT ENFORCED +) +WITH ( + 'connector' = 'jdbc-sqrl', + 'driver' = 'org.postgresql.Driver', + 'password' = '${POSTGRES_PASSWORD}', + 'sink.on-conflict.action' = 'IGNORE', + 'table-name' = 'AlertMessage', + 'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}', + 'username' = '${POSTGRES_USERNAME}' +); +CREATE TABLE `ReadingMessage_2` ( `writer` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, `sensorid` INTEGER NOT NULL, `message_key` VARCHAR(2147483647) CHARACTER SET `UTF-16LE`, @@ -303,7 +472,7 @@ WITH ( 'url' = 'jdbc:postgresql://${POSTGRES_AUTHORITY}', 'username' = '${POSTGRES_USERNAME}' ); -CREATE TABLE `ReadingRecord_2` ( +CREATE TABLE `ReadingRecord_3` ( `writer` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` NOT NULL, `sensorid` INTEGER NOT NULL, `__pk_hash` CHAR(32) CHARACTER SET `UTF-16LE`, @@ -319,20 +488,29 @@ WITH ( 'username' = '${POSTGRES_USERNAME}' ); EXECUTE STATEMENT SET BEGIN +INSERT INTO `default_catalog`.`default_database`.`AlertMessage_1` +SELECT `writer`, `message_key`, `hash_key`(`writer`, `message_key`) AS `__pk_hash` +FROM `default_catalog`.`default_database`.`AlertMessage` +ON CONFLICT DO DEDUPLICATE +; INSERT INTO `default_catalog`.`default_database`.`ReadingWithoutKey_ex1` SELECT * FROM `default_catalog`.`default_database`.`ExternalReading` ; +INSERT INTO `default_catalog`.`default_database`.`Alert` +SELECT * +FROM `default_catalog`.`default_database`.`ForwardedAlert` +; INSERT INTO `default_catalog`.`default_database`.`Reading` SELECT * FROM `default_catalog`.`default_database`.`ForwardedReading` ; -INSERT INTO `default_catalog`.`default_database`.`ReadingMessage_1` +INSERT INTO `default_catalog`.`default_database`.`ReadingMessage_2` SELECT `writer`, `sensorid`, `message_key`, `hash_key`(`writer`, `sensorid`, `message_key`) AS `__pk_hash` FROM `default_catalog`.`default_database`.`ReadingMessage` ON CONFLICT DO DEDUPLICATE ; -INSERT INTO `default_catalog`.`default_database`.`ReadingRecord_2` +INSERT INTO `default_catalog`.`default_database`.`ReadingRecord_3` SELECT `writer`, `sensorid`, `hash_key`(`writer`, `sensorid`) AS `__pk_hash` FROM `default_catalog`.`default_database`.`ReadingRecord` ON CONFLICT DO DEDUPLICATE @@ -341,6 +519,21 @@ END >>>kafka.json { "topics" : [ + { + "topicName" : "Alert", + "tableName" : "Alert", + "logicalName" : "Alert", + "format" : "flexible-json", + "numPartitions" : 1, + "replicationFactor" : 3, + "type" : "MUTATION", + "messageKeys" : [ + "region", + "alert_id" + ], + "messageSchema" : "", + "config" : { } + }, { "topicName" : "ExternalReading", "tableName" : "ExternalReading", @@ -353,6 +546,18 @@ END "messageSchema" : "", "config" : { } }, + { + "topicName" : "ForwardedAlert", + "tableName" : "ForwardedAlert", + "logicalName" : "ForwardedAlert", + "format" : "flexible-json", + "numPartitions" : 1, + "replicationFactor" : 3, + "type" : "MUTATION", + "messageKeys" : [ ], + "messageSchema" : "", + "config" : { } + }, { "topicName" : "ForwardedReading", "tableName" : "ForwardedReading", @@ -383,6 +588,7 @@ END "testRunnerTopics" : [ ] } >>>postgres-schema.sql +CREATE TABLE IF NOT EXISTS "AlertMessage" ("writer" TEXT NOT NULL, "message_key" TEXT, "__pk_hash" TEXT, PRIMARY KEY ("__pk_hash")); CREATE TABLE IF NOT EXISTS "ReadingMessage" ("writer" TEXT NOT NULL, "sensorid" INTEGER NOT NULL, "message_key" TEXT, "__pk_hash" TEXT, PRIMARY KEY ("__pk_hash")); CREATE TABLE IF NOT EXISTS "ReadingRecord" ("writer" TEXT NOT NULL, "sensorid" INTEGER NOT NULL, "__pk_hash" TEXT, PRIMARY KEY ("__pk_hash")) >>>vertx.json @@ -390,6 +596,31 @@ CREATE TABLE IF NOT EXISTS "ReadingRecord" ("writer" TEXT NOT NULL, "sensorid" I "models" : { "v1" : { "queries" : [ + { + "type" : "args", + "parentType" : "Query", + "fieldName" : "AlertMessage", + "exec" : { + "arguments" : [ + { + "type" : "variable", + "path" : "limit" + }, + { + "type" : "variable", + "path" : "offset" + } + ], + "query" : { + "type" : "SqlQuery", + "sql" : "SELECT *\nFROM (SELECT \"writer\", \"message_key\"\n FROM \"AlertMessage\"\n ORDER BY \"writer\" NULLS FIRST) AS \"t0\"", + "parameters" : [ ], + "pagination" : "LIMIT_AND_OFFSET", + "cacheDurationMs" : 0, + "database" : "POSTGRES" + } + } + }, { "type" : "args", "parentType" : "Query", @@ -442,6 +673,30 @@ CREATE TABLE IF NOT EXISTS "ReadingRecord" ("writer" TEXT NOT NULL, "sensorid" I } ], "mutations" : [ + { + "type" : "kafka", + "fieldName" : "Alert", + "returnList" : false, + "topic" : "Alert", + "keyColumns" : [ + "region", + "alert_id" + ], + "computedColumns" : { + "alert_id" : { + "metadataType" : "UUID", + "name" : "", + "required" : true + }, + "event_time" : { + "metadataType" : "TIMESTAMP", + "name" : "", + "required" : true + } + }, + "transactional" : false, + "sinkConfig" : { } + }, { "type" : "kafka", "fieldName" : "ExternalReading", @@ -458,6 +713,27 @@ CREATE TABLE IF NOT EXISTS "ReadingRecord" ("writer" TEXT NOT NULL, "sensorid" I "transactional" : false, "sinkConfig" : { } }, + { + "type" : "kafka", + "fieldName" : "ForwardedAlert", + "returnList" : false, + "topic" : "ForwardedAlert", + "keyColumns" : [ ], + "computedColumns" : { + "alert_id" : { + "metadataType" : "UUID", + "name" : "", + "required" : true + }, + "event_time" : { + "metadataType" : "TIMESTAMP", + "name" : "", + "required" : true + } + }, + "transactional" : false, + "sinkConfig" : { } + }, { "type" : "kafka", "fieldName" : "ForwardedReading", @@ -495,6 +771,46 @@ CREATE TABLE IF NOT EXISTS "ReadingRecord" ("writer" TEXT NOT NULL, "sensorid" I ], "subscriptions" : [ ], "operations" : [ + { + "function" : { + "name" : "GetAlertMessage", + "parameters" : { + "type" : "object", + "properties" : { + "offset" : { + "type" : "integer" + }, + "limit" : { + "type" : "integer" + } + }, + "required" : [ ] + } + }, + "apiQuery" : { + "query" : "query AlertMessage($limit: Int = 10, $offset: Int = 0) {\nAlertMessage(limit: $limit, offset: $offset) {\nwriter\nmessage_key\n}\n\n}", + "queryName" : "AlertMessage", + "operationType" : "QUERY" + }, + "mcpMethod" : "TOOL", + "restMethod" : "GET", + "uriTemplate" : "queries/AlertMessage{?offset,limit}", + "format" : "JSON", + "resultDefinition" : { + "type" : "array", + "items" : { + "type" : "object", + "properties" : { + "writer" : { + "type" : "string" + }, + "message_key" : { + "type" : "string" + } + } + } + } + }, { "function" : { "name" : "GetReadingMessage", @@ -578,6 +894,52 @@ CREATE TABLE IF NOT EXISTS "ReadingRecord" ("writer" TEXT NOT NULL, "sensorid" I } } }, + { + "function" : { + "name" : "AddAlert", + "parameters" : { + "type" : "object", + "properties" : { + "writer" : { + "type" : "string" + }, + "region" : { + "type" : "string" + } + }, + "required" : [ + "writer" + ] + } + }, + "apiQuery" : { + "query" : "mutation Alert($region: String, $writer: String!) {\nAlert(event: { region: $region, writer: $writer }) {\nregion\nalert_id\nwriter\nevent_time\n}\n\n}", + "queryName" : "Alert", + "operationType" : "MUTATION" + }, + "mcpMethod" : "TOOL", + "restMethod" : "POST", + "uriTemplate" : "mutations/Alert", + "format" : "JSON", + "resultDefinition" : { + "type" : "object", + "properties" : { + "region" : { + "type" : "string" + }, + "alert_id" : { + "type" : "string" + }, + "writer" : { + "type" : "string" + }, + "event_time" : { + "type" : "string", + "format" : "date-time" + } + } + } + }, { "function" : { "name" : "AddExternalReading", @@ -622,6 +984,52 @@ CREATE TABLE IF NOT EXISTS "ReadingRecord" ("writer" TEXT NOT NULL, "sensorid" I } } }, + { + "function" : { + "name" : "AddForwardedAlert", + "parameters" : { + "type" : "object", + "properties" : { + "writer" : { + "type" : "string" + }, + "region" : { + "type" : "string" + } + }, + "required" : [ + "writer" + ] + } + }, + "apiQuery" : { + "query" : "mutation ForwardedAlert($region: String, $writer: String!) {\nForwardedAlert(event: { region: $region, writer: $writer }) {\nregion\nalert_id\nwriter\nevent_time\n}\n\n}", + "queryName" : "ForwardedAlert", + "operationType" : "MUTATION" + }, + "mcpMethod" : "TOOL", + "restMethod" : "POST", + "uriTemplate" : "mutations/ForwardedAlert", + "format" : "JSON", + "resultDefinition" : { + "type" : "object", + "properties" : { + "region" : { + "type" : "string" + }, + "alert_id" : { + "type" : "string" + }, + "writer" : { + "type" : "string" + }, + "event_time" : { + "type" : "string", + "format" : "date-time" + } + } + } + }, { "function" : { "name" : "AddForwardedReading", @@ -713,7 +1121,7 @@ CREATE TABLE IF NOT EXISTS "ReadingRecord" ("writer" TEXT NOT NULL, "sensorid" I ], "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\ninput ExternalReadingInput {\n sensorid: Int!\n writer: String!\n}\n\ntype ExternalReadingResultOutput {\n sensorid: Int!\n writer: String!\n event_time: DateTime!\n}\n\ninput ForwardedReadingInput {\n sensorid: Int!\n writer: String!\n}\n\ntype ForwardedReadingResultOutput {\n sensorid: Int!\n writer: String!\n event_time: DateTime!\n}\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 ExternalReading(event: ExternalReadingInput!): ExternalReadingResultOutput!\n ForwardedReading(event: ForwardedReadingInput!): ForwardedReadingResultOutput!\n Reading(event: ReadingInput!): ReadingResultOutput!\n}\n\ntype Query {\n ReadingMessage(limit: Int = 10, offset: Int = 0): [ReadingMessage!]\n ReadingRecord(limit: Int = 10, offset: Int = 0): [ReadingRecord!]\n}\n\ninput ReadingInput {\n sensorid: Int!\n writer: String!\n}\n\ntype ReadingMessage {\n writer: String!\n sensorid: Int!\n message_key: String\n}\n\ntype ReadingRecord {\n writer: String!\n sensorid: Int!\n}\n\ntype ReadingResultOutput {\n sensorid: Int!\n writer: String!\n event_time: DateTime!\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" : "input AlertInput {\n region: String\n writer: String!\n}\n\ntype AlertMessage {\n writer: String!\n message_key: String\n}\n\ntype AlertResultOutput {\n region: String\n alert_id: String!\n writer: String!\n event_time: 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\ninput ExternalReadingInput {\n sensorid: Int!\n writer: String!\n}\n\ntype ExternalReadingResultOutput {\n sensorid: Int!\n writer: String!\n event_time: DateTime!\n}\n\ninput ForwardedAlertInput {\n region: String\n writer: String!\n}\n\ntype ForwardedAlertResultOutput {\n region: String\n alert_id: String!\n writer: String!\n event_time: DateTime!\n}\n\ninput ForwardedReadingInput {\n sensorid: Int!\n writer: String!\n}\n\ntype ForwardedReadingResultOutput {\n sensorid: Int!\n writer: String!\n event_time: DateTime!\n}\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 Alert(event: AlertInput!): AlertResultOutput!\n ExternalReading(event: ExternalReadingInput!): ExternalReadingResultOutput!\n ForwardedAlert(event: ForwardedAlertInput!): ForwardedAlertResultOutput!\n ForwardedReading(event: ForwardedReadingInput!): ForwardedReadingResultOutput!\n Reading(event: ReadingInput!): ReadingResultOutput!\n}\n\ntype Query {\n AlertMessage(limit: Int = 10, offset: Int = 0): [AlertMessage!]\n ReadingMessage(limit: Int = 10, offset: Int = 0): [ReadingMessage!]\n ReadingRecord(limit: Int = 10, offset: Int = 0): [ReadingRecord!]\n}\n\ninput ReadingInput {\n sensorid: Int!\n writer: String!\n}\n\ntype ReadingMessage {\n writer: String!\n sensorid: Int!\n message_key: String\n}\n\ntype ReadingRecord {\n writer: String!\n sensorid: Int!\n}\n\ntype ReadingResultOutput {\n sensorid: Int!\n writer: String!\n event_time: DateTime!\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/kafka-message-keys/kafka-message-keys.sqrl b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/kafka-message-keys.sqrl index b409f8544b..dcbc859ae9 100644 --- a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/kafka-message-keys.sqrl +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/kafka-message-keys.sqrl @@ -49,3 +49,44 @@ CREATE TABLE _ReadingMessage ( ); ReadingMessage := SELECT writer, sensorid, message_key FROM _ReadingMessage ORDER BY writer ASC, sensorid ASC; + +-- The Alert key contains a column the client can omit and a column the server computes +/*+engine(kafka), no_query */ +CREATE TABLE Alert ( + region STRING, + alert_id STRING NOT NULL METADATA FROM 'uuid', + writer STRING NOT NULL, + event_time TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp' +) PARTITIONED BY (region, alert_id); + +/*+engine(kafka), no_query */ +CREATE TABLE ForwardedAlert ( + region STRING, + alert_id STRING NOT NULL METADATA FROM 'uuid', + writer STRING NOT NULL, + event_time TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp' +); + +EXPORT ForwardedAlert TO Alert; + +CREATE TABLE _AlertMessage ( + message_key STRING, + region STRING, + alert_id STRING NOT NULL, + writer STRING NOT NULL, + event_time TIMESTAMP_LTZ(3) NOT NULL METADATA FROM 'timestamp', + WATERMARK FOR event_time AS event_time - INTERVAL '1' SECOND +) WITH ( + 'connector' = 'kafka-safe', + 'topic' = 'Alert', + 'properties.bootstrap.servers' = '${KAFKA_BOOTSTRAP_SERVERS}', + 'properties.group.id' = 'alert-message', + 'scan.startup.mode' = 'earliest-offset', + 'key.format' = 'raw', + 'key.fields' = 'message_key', + 'value.format' = 'flexible-json', + 'value.fields-include' = 'EXCEPT_KEY' +); + +-- Masks the random alert_id so that the snapshot shows the exact message key +AlertMessage := SELECT writer, REPLACE(message_key, alert_id, '') AS message_key FROM _AlertMessage ORDER BY writer ASC; diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/snapshots/mutation4.snapshot b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/snapshots/mutation4.snapshot new file mode 100644 index 0000000000..1e626b9c31 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/snapshots/mutation4.snapshot @@ -0,0 +1,16 @@ +{ + "data" : { + "add1" : { + "writer" : "server-region" + }, + "add2" : { + "writer" : "server-omitted-region" + }, + "add3" : { + "writer" : "server-null-region" + }, + "add4" : { + "writer" : "flink-omitted-region" + } + } +} \ No newline at end of file diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/snapshots/query_alerts.snapshot b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/snapshots/query_alerts.snapshot new file mode 100644 index 0000000000..699f326161 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/snapshots/query_alerts.snapshot @@ -0,0 +1,17 @@ +{ + "data" : { + "AlertMessage" : [ { + "writer" : "flink-omitted-region", + "message_key" : "{\"region\":null,\"alert_id\":\"\"}" + }, { + "writer" : "server-null-region", + "message_key" : "{\"region\":null,\"alert_id\":\"\"}" + }, { + "writer" : "server-omitted-region", + "message_key" : "{\"region\":null,\"alert_id\":\"\"}" + }, { + "writer" : "server-region", + "message_key" : "{\"region\":\"north\",\"alert_id\":\"\"}" + } ] + } +} \ No newline at end of file diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/tests/mutation4.graphql b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/tests/mutation4.graphql new file mode 100644 index 0000000000..a4baa544ee --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/tests/mutation4.graphql @@ -0,0 +1,14 @@ +mutation Alerts { + add1: Alert(event: { region: "north", writer: "server-region" }) { + writer + } + add2: Alert(event: { writer: "server-omitted-region" }) { + writer + } + add3: Alert(event: { region: null, writer: "server-null-region" }) { + writer + } + add4: ForwardedAlert(event: { writer: "flink-omitted-region" }) { + writer + } +} diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/tests/query_alerts.graphql b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/tests/query_alerts.graphql new file mode 100644 index 0000000000..b118bdbcfa --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/kafka-message-keys/tests/query_alerts.graphql @@ -0,0 +1,6 @@ +query GetAlertMessages { + AlertMessage(limit: 10, offset: 0) { + writer + message_key + } +}