Risingwave demo bug fixes

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

@@ -23,23 +23,8 @@ catalogs:
       id: 1
       schema: |-
         {
-        "type": "record",
-        "name": "cities",
-        "namespace": "dev",
-        "fields": [
-          {
-            "name": "description",
-            "type": string
-          },
-          {
-            "name": "id",
-            "type": string
-          },
-          {
-            "name": "name",
-            "type": string
-          }
-        ]
+        "schemaType": "AVRO",
+        "schema": "{\"type\": \"record\", \"name\": \"cities\", \"namespace\": \"dev\", \"fields\": [ {\"name\": \"id\", \"type\": \"string\"}, {\"name\": \"city\", \"type\": \"string\"}]}"
         }
 bindings:
   app0:

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/kafka/create.topic/client.rpt

@@ -13,40 +13,6 @@
 # specific language governing permissions and limitations under the License.
 #
 
-connect "zilla://streams/app1"
-    option zilla:window 8192
-    option zilla:transmission "half-duplex"
-
-write zilla:begin.ext ${kafka:beginEx()
-                               .typeId(zilla:id("kafka"))
-                               .request()
-                                 .describeCluster()
-                                    .includeAuthorizedOperations("false")
-                                 .build()
-                               .build()}
-
-connected
-
-read zilla:begin.ext ${kafka:matchBeginEx()
-                               .typeId(zilla:id("kafka"))
-                               .response()
-                                 .describeCluster()
-                                    .throttle(0)
-                                    .error(0)
-                                    .clusterId("cluster-0")
-                                    .controllerId(0)
-                                    .broker()
-                                       .brokerId(1)
-                                       .host("broker1.example.com")
-                                       .port(9092)
-                                       .build()
-                                   .authorizedOperations(0)
-                                   .build()
-                                .build()}
-
-write close
-read closed
-
 connect "zilla://streams/app1"
     option zilla:window 8192
     option zilla:transmission "half-duplex"
@@ -59,7 +25,6 @@ write zilla:begin.ext ${kafka:beginEx()
                                        .name("dev.cities")
                                        .partitionCount(1)
                                        .replicas(1)
-                                       .assignment(0, 1)
                                        .config("cleanup.policy", "compact")
                                        .build()
                                     .timeout(30000)

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

@@ -21,39 +21,6 @@ accept ${serverAddress}
 
 accepted
 
-read zilla:begin.ext ${kafka:matchBeginEx()
-                               .typeId(zilla:id("kafka"))
-                               .request()
-                                 .describeCluster()
-                                    .includeAuthorizedOperations("false")
-                                 .build()
-                               .build()}
-
-connected
-
-write zilla:begin.ext ${kafka:beginEx()
-                               .typeId(zilla:id("kafka"))
-                               .response()
-                                 .describeCluster()
-                                    .throttle(0)
-                                    .error(0)
-                                    .clusterId("cluster-0")
-                                    .controllerId(0)
-                                    .broker()
-                                       .brokerId(1)
-                                       .host("broker1.example.com")
-                                       .port(9092)
-                                       .build()
-                                   .authorizedOperations(0)
-                                   .build()
-                                .build()}
-write flush
-
-read closed
-write close
-
-accepted
-
 read zilla:begin.ext ${kafka:matchBeginEx()
                                .typeId(zilla:id("kafka"))
                                .request()
@@ -62,7 +29,6 @@ read zilla:begin.ext ${kafka:matchBeginEx()
                                        .name("dev.cities")
                                        .partitionCount(1)
                                        .replicas(1)
-                                       .assignment(0, 1)
                                        .config("cleanup.policy", "compact")
                                        .build()
                                    .timeout(30000)

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/pgsql/create.topic/client.rpt

@@ -34,7 +34,7 @@ write zilla:data.ext ${pgsql:dataEx()
                                 .build()
                               .build()}
 write "CREATE TOPIC IF NOT EXISTS cities "
-      "(description VARCHAR, id VARCHAR, name VARCHAR, PRIMARY KEY (id));"
+      "(id VARCHAR, city VARCHAR, PRIMARY KEY (id, city));"
       [0x00]
 write flush
 

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

@@ -34,8 +34,8 @@ read zilla:data.ext ${pgsql:dataEx()
                               .query()
                                 .build()
                               .build()}
-read "CREATE TOPIC IF NOT EXISTS cities "
-     "(description VARCHAR, id VARCHAR, name VARCHAR, PRIMARY KEY (id));"
+read  "CREATE TOPIC IF NOT EXISTS cities "
+      "(id VARCHAR, city VARCHAR, PRIMARY KEY (id, city));"
      [0x00]
 
 write advise zilla:flush ${pgsql:flushEx()

incubator/binding-pgsql-kafka/pom.xml

@@ -24,7 +24,7 @@
   </licenses>
 
   <properties>
-    <jacoco.coverage.ratio>0.83</jacoco.coverage.ratio>
+    <jacoco.coverage.ratio>0.82</jacoco.coverage.ratio>
     <jacoco.missed.count>0</jacoco.missed.count>
   </properties>
 

incubator/binding-pgsql-kafka/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/kafka/internal/config/PgsqlKafkaBindingConfig.java

@@ -20,6 +20,7 @@
 import java.util.function.LongFunction;
 
 import io.aklivity.zilla.runtime.binding.pgsql.kafka.internal.PgsqlKafkaConfiguration;
+import io.aklivity.zilla.runtime.binding.pgsql.kafka.internal.schema.PgsqlKafkaKeyAvroSchemaTemplate;
 import io.aklivity.zilla.runtime.binding.pgsql.kafka.internal.schema.PgsqlKafkaValueAvroSchemaTemplate;
 import io.aklivity.zilla.runtime.engine.catalog.CatalogHandler;
 import io.aklivity.zilla.runtime.engine.config.BindingConfig;
@@ -33,6 +34,7 @@ public final class PgsqlKafkaBindingConfig
     public final List<PgsqlKafkaRouteConfig> routes;
     public final CatalogHandler catalog;
     public final PgsqlKafkaValueAvroSchemaTemplate avroValueSchema;
+    public final PgsqlKafkaKeyAvroSchemaTemplate avroKeySchema;
 
     public PgsqlKafkaBindingConfig(
         PgsqlKafkaConfiguration config,
@@ -46,6 +48,7 @@ public PgsqlKafkaBindingConfig(
 
         this.catalog = supplyCatalog.apply(binding.catalogs.get(0).id);
         this.avroValueSchema = new PgsqlKafkaValueAvroSchemaTemplate(config.kafkaAvroSchemaNamespace());
+        this.avroKeySchema = new PgsqlKafkaKeyAvroSchemaTemplate(config.kafkaAvroSchemaNamespace());
     }
 
     public PgsqlKafkaRouteConfig resolve(

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

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2021-2023 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.
+ */
+package io.aklivity.zilla.runtime.binding.pgsql.kafka.internal.schema;
+
+import java.util.List;
+
+import net.sf.jsqlparser.statement.create.table.ColumnDefinition;
+import net.sf.jsqlparser.statement.create.table.CreateTable;
+import net.sf.jsqlparser.statement.create.table.Index;
+
+public class PgsqlKafkaKeyAvroSchemaTemplate extends PgsqlKafkaAvroSchemaTemplate
+{
+    private static final String DATABASE_PLACEHOLDER = "{database}";
+
+    private final StringBuilder schemaBuilder = new StringBuilder();
+    private final String namespace;
+
+    public PgsqlKafkaKeyAvroSchemaTemplate(
+        String namespace)
+    {
+        this.namespace = namespace;
+    }
+
+    public String generateSchema(
+        String database,
+        CreateTable createTable)
+    {
+        schemaBuilder.setLength(0);
+
+        final String newNamespace = namespace.replace(DATABASE_PLACEHOLDER, database);
+        final String recordName = String.format("%s_key", createTable.getTable().getName());
+
+        schemaBuilder.append("{\n");
+        schemaBuilder.append("\"schemaType\": \"AVRO\",\n");
+        schemaBuilder.append("\"schema\": \""); // Begin the schema field
+
+        // Building the actual Avro schema
+        schemaBuilder.append("{\\\"type\\\": \\\"record\\\",");
+        schemaBuilder.append(" \\\"name\\\": \\\"").append(recordName).append("\\\",");
+        schemaBuilder.append(" \\\"namespace\\\": \\\"").append(newNamespace).append("\\\",");
+        schemaBuilder.append(" \\\"fields\\\": [");
+
+        for (ColumnDefinition column : createTable.getColumnDefinitions())
+        {
+            String fieldName = column.getColumnName();
+            String pgsqlType = column.getColDataType().getDataType();
+
+            String avroType = convertPgsqlTypeToAvro(pgsqlType);
+
+            schemaBuilder.append(" {\\\"name\\\": \\\"").append(fieldName).append("\\\",");
+            schemaBuilder.append(" \\\"type\\\": [\\\"").append(avroType).append("\\\", \\\"null\\\"] },");
+        }
+
+        // Remove the last comma and close the fields array
+        schemaBuilder.setLength(schemaBuilder.length() - 1);
+        schemaBuilder.append("]");
+
+        // Closing the Avro schema
+        schemaBuilder.append("}\"\n}");
+
+        return schemaBuilder.toString();
+    }
+
+    public String primaryKey(
+        CreateTable statement)
+    {
+        String primaryKey = null;
+
+        final List<Index> indexes = statement.getIndexes();
+
+        if (indexes != null && !indexes.isEmpty())
+        {
+            match:
+            for (Index index : indexes)
+            {
+                if ("PRIMARY KEY".equalsIgnoreCase(index.getType()))
+                {
+                    final List<Index.ColumnParams> primaryKeyColumns = index.getColumns();
+                    primaryKey = primaryKeyColumns.get(0).columnName;
+                    break match;
+                }
+            }
+        }
+
+        return primaryKey;
+    }
+}

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

@@ -43,10 +43,14 @@ public String generateSchema(
         final String recordName = createTable.getTable().getName();
 
         schemaBuilder.append("{\n");
-        schemaBuilder.append("\"type\": \"record\",\n");
-        schemaBuilder.append("\"name\": \"").append(recordName).append("\",\n");
-        schemaBuilder.append("\"namespace\": \"").append(newNamespace).append("\",\n");
-        schemaBuilder.append("\"fields\": [\n");
+        schemaBuilder.append("\"schemaType\": \"AVRO\",\n");
+        schemaBuilder.append("\"schema\": \""); // Begin the schema field
+
+        // Building the actual Avro schema
+        schemaBuilder.append("{\\\"type\\\": \\\"record\\\",");
+        schemaBuilder.append(" \\\"name\\\": \\\"").append(recordName).append("\\\",");
+        schemaBuilder.append(" \\\"namespace\\\": \\\"").append(newNamespace).append("\\\",");
+        schemaBuilder.append(" \\\"fields\\\": [");
 
         for (ColumnDefinition column : createTable.getColumnDefinitions())
         {
@@ -55,15 +59,16 @@ public String generateSchema(
 
             String avroType = convertPgsqlTypeToAvro(pgsqlType);
 
-            schemaBuilder.append("  {\n");
-            schemaBuilder.append("    \"name\": \"").append(fieldName).append("\",\n");
-            schemaBuilder.append("    \"type\": ").append(avroType).append("\n");
-            schemaBuilder.append("  },\n");
+            schemaBuilder.append(" {\\\"name\\\": \\\"").append(fieldName).append("\\\",");
+            schemaBuilder.append(" \\\"type\\\": \\\"").append(avroType).append("\\\"},");
         }
 
-        // Remove the last comma after the last field and close the JSON array
-        schemaBuilder.setLength(schemaBuilder.length() - 2); // Remove last comma
-        schemaBuilder.append("\n]\n}");
+        // Remove the last comma and close the fields array
+        schemaBuilder.setLength(schemaBuilder.length() - 1);
+        schemaBuilder.append("]");
+
+        // Closing the Avro schema
+        schemaBuilder.append("}\"\n}");
 
         return schemaBuilder.toString();
     }
@@ -91,4 +96,27 @@ public String primaryKey(
 
         return primaryKey;
     }
+
+    public int primaryKeyCount(
+        CreateTable statement)
+    {
+        int primaryKeyCount = 0;
+
+        final List<Index> indexes = statement.getIndexes();
+
+        if (indexes != null && !indexes.isEmpty())
+        {
+            match:
+            for (Index index : indexes)
+            {
+                if ("PRIMARY KEY".equalsIgnoreCase(index.getType()))
+                {
+                    primaryKeyCount = index.getColumns().size();
+                    break match;
+                }
+            }
+        }
+
+        return primaryKeyCount;
+    }
 }