Ztable support and convert transformation logic to state machine

.gitignore

@@ -31,3 +31,6 @@ node_modules/
 *.pdf
 *.zshrc
 hs_err_pid*.log
+*.interp
+*.tokens
+gen/

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/config/proxy.generated.as.yaml

@@ -0,0 +1,36 @@
+#
+# Copyright 2021-2024 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+---
+name: test
+catalogs:
+  catalog0:
+    type: test
+    options:
+      url: http://localhost:8081
+      id: 1
+      schema: |-
+        {
+            "schema": "{\"fields\":[{\"name\":\"id\",\"type\":[\"null\",{\"type\":\"string\"}]},{\"name\":\"city\",\"type\":[\"null\",{\"type\":\"string\"}]},{\"name\":\"owner_id\",\"type\":[\"null\",{\"type\":\"string\"}],\"zillaType\":\"ZILLA_IDENTITY\"},{\"name\":\"created_at\",\"type\":[\"null\",{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"}],\"zillaType\":\"ZILLA_TIMESTAMP\"}],\"name\":\"cities\",\"namespace\":\"public\",\"type\":\"record\"}",
+            "schemaType": "AVRO"
+        }
+bindings:
+  app0:
+    type: pgsql-kafka
+    kind: proxy
+    catalog:
+      catalog0:
+        - strategy: topic
+    exit: app1

incubator/binding-risingwave.spec/src/main/scripts/io/aklivity/zilla/specs/binding/risingwave/streams/pgsql/create.table.with.primary.key.and.includes/client.rpt → incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/pgsql/create.topic.with.generated.as/client.rpt

@@ -17,6 +17,7 @@ connect "zilla://streams/app0"
     option zilla:window 8192
     option zilla:transmission "duplex"
 
+
 write zilla:begin.ext ${pgsql:beginEx()
                               .typeId(zilla:id("pgsql"))
                               .parameter("user", "app")
@@ -32,18 +33,19 @@ write zilla:data.ext ${pgsql:dataEx()
                               .query()
                                 .build()
                               .build()}
-write "CREATE TABLE IF NOT EXISTS cities "
-      "(id VARCHAR, name VARCHAR, description VARCHAR,"
-      " zilla_correlation_id VARCHAR, zilla_identity VARCHAR, zilla_timestamp TIMESTAMP,"
-      " PRIMARY KEY (id));"
+write "CREATE TOPIC IF NOT EXISTS cities ("
+      "id VARCHAR, "
+      "city VARCHAR, "
+      "owner_id VARCHAR GENERATED ALWAYS AS IDENTITY, "
+      "created_at TIMESTAMP GENERATED ALWAYS AS NOW, "
+      "PRIMARY KEY (id, city));"
       [0x00]
-
 write flush
 
 read advised zilla:flush ${pgsql:flushEx()
                               .typeId(zilla:id("pgsql"))
                               .completion()
-                                .tag("CREATE_TABLE")
+                                .tag("CREATE_TOPIC")
                                 .build()
                               .build()}
 
@@ -54,3 +56,5 @@ read advised zilla:flush ${pgsql:flushEx()
                                 .build()
                               .build()}
 
+write close
+read closed

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/pgsql/create.topic.with.generated.as/server.rpt

@@ -0,0 +1,61 @@
+#
+# Copyright 2021-2024 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+accept "zilla://streams/app0"
+    option zilla:window 8192
+    option zilla:transmission "duplex"
+
+accepted
+
+read zilla:begin.ext ${pgsql:beginEx()
+                              .typeId(zilla:id("pgsql"))
+                              .parameter("user", "root")
+                              .parameter("database", "dev")
+                              .parameter("application_name", "psql")
+                              .parameter("client_encoding", "UTF8")
+                              .build()}
+
+connected
+
+read zilla:data.ext ${pgsql:dataEx()
+                              .typeId(zilla:id("pgsql"))
+                              .query()
+                                .build()
+                              .build()}
+read "CREATE TOPIC IF NOT EXISTS cities ("
+     "id VARCHAR, "
+     "city VARCHAR, "
+     "owner_id VARCHAR GENERATED ALWAYS AS IDENTITY, "
+     "created_at TIMESTAMP GENERATED ALWAYS AS NOW, "
+     "PRIMARY KEY (id, city));"
+     [0x00]
+
+write advise zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .completion()
+                                .tag("CREATE_TOPIC")
+                                .build()
+                              .build()}
+
+write notify CREATE_TOPIC_COMPLETED
+
+write advise zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .ready()
+                                .status("IDLE")
+                                .build()
+                              .build()}
+read closed
+write close

incubator/binding-pgsql-kafka/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/schema/AvroField.java

@@ -16,6 +16,7 @@
 
 public record AvroField(
     String name,
-    Object type)
+    Object type,
+    Object zillaType)
 {
 }

incubator/binding-pgsql-kafka/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/schema/PgsqlKafkaAvroSchemaTemplate.java

@@ -15,6 +15,7 @@
 package io.aklivity.zilla.runtime.binding.pgsql.kafka.internal.schema;
 
 import java.util.List;
+import java.util.Map;
 
 import jakarta.json.Json;
 import jakarta.json.JsonArray;
@@ -25,13 +26,25 @@
 import jakarta.json.bind.JsonbBuilder;
 import jakarta.json.bind.JsonbConfig;
 
+import org.agrona.collections.Object2ObjectHashMap;
+
 import io.aklivity.zilla.runtime.binding.pgsql.parser.model.AlterExpression;
 import io.aklivity.zilla.runtime.binding.pgsql.parser.model.Operation;
 
 public abstract class PgsqlKafkaAvroSchemaTemplate
 {
     protected static final String DATABASE_PLACEHOLDER = "{database}";
 
+    protected static final String ZILLA_IDENTITY = "GENERATED ALWAYS AS IDENTITY";
+    protected static final String ZILLA_TIMESTAMP = "GENERATED ALWAYS AS NOW";
+
+    protected static final Map<String, String> ZILLA_MAPPINGS = new Object2ObjectHashMap<>();
+    static
+    {
+        ZILLA_MAPPINGS.put(ZILLA_IDENTITY, "ZILLA_IDENTITY");
+        ZILLA_MAPPINGS.put(ZILLA_TIMESTAMP, "ZILLA_TIMESTAMP");
+    }
+
     protected Jsonb jsonbFormatted = JsonbBuilder.create(new JsonbConfig().withFormatting(true));
     protected Jsonb jsonb = JsonbBuilder.create();
 

incubator/binding-pgsql-kafka/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/schema/PgsqlKafkaKeyAvroSchemaTemplate.java

@@ -36,7 +36,7 @@ public String generate(
         final String newNamespace = namespace.replace(DATABASE_PLACEHOLDER, database);
 
         List<AvroField> fields = command.columns().stream()
-            .map(column -> new AvroField(column.name(), mapSqlTypeToAvroType(column.type())))
+            .map(column -> new AvroField(column.name(), mapSqlTypeToAvroType(column.type()), null))
             .collect(Collectors.toList());
 
         AvroSchema schema = new AvroSchema("record", command.name(), newNamespace, fields);

incubator/binding-pgsql-kafka/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/schema/PgsqlKafkaValueAvroSchemaTemplate.java

@@ -49,12 +49,19 @@ public String generate(
                 String columnName = column.name();
                 String sqlType = column.type();
                 Object avroType = mapSqlTypeToAvroType(sqlType);
+                List<String> constraints = column.constraints();
 
-                boolean isNullable = !column.constraints().contains("NOT NULL");
+                boolean isNullable = !constraints.contains("NOT NULL");
+
+                String zillaType = ZILLA_MAPPINGS.entrySet().stream()
+                        .filter(e -> constraints.contains(e.getKey()))
+                        .map(Map.Entry::getValue)
+                        .findFirst()
+                        .orElse(null);
 
                 return isNullable
-                    ? new AvroField(columnName, new Object[]{"null", avroType})
-                    : new AvroField(columnName, avroType);
+                    ? new AvroField(columnName, new Object[]{"null", avroType}, zillaType)
+                    : new AvroField(columnName, avroType, zillaType);
             })
             .collect(Collectors.toList());
 

incubator/binding-pgsql-kafka/src/test/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/stream/ProxyIT.java

@@ -59,6 +59,17 @@ public void shouldCreateTopic() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("proxy.generated.as.yaml")
+    @Specification({
+        "${pgsql}/create.topic.with.generated.as/client",
+        "${kafka}/create.topic/server"
+    })
+    public void shouldCreateTopicWithGeneratedAs() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("proxy.alter.yaml")
     @Specification({

incubator/binding-pgsql.spec/src/main/java/io/aklivity/zilla/specs/binding/pgsql/PgsqlFunctions.java

@@ -28,6 +28,7 @@
 import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlFlushExFW;
 import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlFormat;
 import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlMessageType;
+import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlNoticeFlushExFW;
 import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlQueryDataExFW;
 import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlReadyFlushExFW;
 import io.aklivity.zilla.specs.binding.pgsql.internal.types.stream.PgsqlRowDataExFW;
@@ -217,6 +218,13 @@ public PgsqlCompletedFlushExBuilder completion()
             return new PgsqlCompletedFlushExBuilder();
         }
 
+        public PgsqlNoticeFlushExBuilder notice()
+        {
+            flushExRW.kind(PgsqlMessageType.NOTICE.value());
+
+            return new PgsqlNoticeFlushExBuilder();
+        }
+
         public PgsqlErrorFlushExBuilder error()
         {
             flushExRW.kind(PgsqlMessageType.ERROR.value());
@@ -405,6 +413,44 @@ public PgsqlFlushExBuilder build()
             }
         }
 
+        public final class PgsqlNoticeFlushExBuilder
+        {
+            private final PgsqlNoticeFlushExFW.Builder pgsqlNoticeFlushExRW = new PgsqlNoticeFlushExFW.Builder();
+
+            private PgsqlNoticeFlushExBuilder()
+            {
+                pgsqlNoticeFlushExRW.wrap(writeBuffer, PgsqlFlushExFW.FIELD_OFFSET_TYPE, writeBuffer.capacity());
+            }
+
+            public PgsqlNoticeFlushExBuilder severity(
+                String severity)
+            {
+                pgsqlNoticeFlushExRW.severity(String.format("S%s\u0000", severity));
+                return this;
+            }
+
+            public PgsqlNoticeFlushExBuilder code(
+                String code)
+            {
+                pgsqlNoticeFlushExRW.code(String.format("C%s\u0000", code));
+                return this;
+            }
+
+            public PgsqlNoticeFlushExBuilder message(
+                String message)
+            {
+                pgsqlNoticeFlushExRW.message(String.format("M%s\u0000", message));
+                return this;
+            }
+
+            public PgsqlFlushExBuilder build()
+            {
+                final PgsqlNoticeFlushExFW pgsqlNoticeFlushEx = pgsqlNoticeFlushExRW.build();
+                flushExRO.wrap(writeBuffer, 0, pgsqlNoticeFlushEx.limit());
+                return PgsqlFlushExBuilder.this;
+            }
+        }
+
         public final class PgsqlCancelRequestFlushExBuilder
         {
             private final PgsqlCancelRequestFlushExFW.Builder pgsqlCancelREquestFlushExRW =

incubator/binding-pgsql.spec/src/main/resources/META-INF/zilla/pgsql.idl

@@ -33,6 +33,7 @@ scope pgsql
             ROW (68),
             TYPE (84),
             COMPLETION (67),
+            NOTICE (78),
             ERROR (69),
             CANCEL_REQUEST (255),
             READY (90)
@@ -60,6 +61,7 @@ scope pgsql
             case 84: pgsql::stream::PgsqlTypeFlushEx type;
             case 67: pgsql::stream::PgsqlCompletedFlushEx completion;
             case 69: pgsql::stream::PgsqlErrorFlushEx error;
+            case 78: pgsql::stream::PgsqlNoticeFlushEx notice;
             case 90: pgsql::stream::PgsqlReadyFlushEx ready;
         }
 
@@ -103,6 +105,13 @@ scope pgsql
             string16 message;
         }
 
+        struct PgsqlNoticeFlushEx
+        {
+            string16 severity;
+            string16 code;
+            string16 message;
+        }
+
         enum PgsqlStatus (uint8)
         {
             IDLE(73),

incubator/binding-pgsql.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/streams/application/create.view.with.notice/client.rpt

@@ -0,0 +1,63 @@
+#
+# Copyright 2021-2024 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+connect "zilla://streams/app0"
+    option zilla:window 8192
+    option zilla:transmission "duplex"
+
+write zilla:begin.ext ${pgsql:beginEx()
+                              .typeId(zilla:id("pgsql"))
+                              .parameter("user", "root")
+                              .parameter("database", "dev")
+                              .parameter("application_name", "psql")
+                              .parameter("client_encoding", "UTF8")
+                              .build()}
+
+connected
+
+write zilla:data.ext ${pgsql:dataEx()
+                              .typeId(zilla:id("pgsql"))
+                              .query()
+                                .build()
+                              .build()}
+write "CREATE VIEW IF NOT EXISTS cities_view AS SELECT name, created_at::timestamp FROM cities;"
+      [0x00]
+write flush
+
+read advised zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .notice()
+                                .severity("NOTICE")
+                                .code("00000")
+                                .message("Your session timezone is UTC.")
+                                .build()
+                              .build()}
+
+read advised zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .completion()
+                                .tag("CREATE_VIEW")
+                                .build()
+                              .build()}
+
+read advised zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .ready()
+                                .status("IDLE")
+                                .build()
+                              .build()}
+
+read closed
+write close