pgsql ALTER TOPIC command to register new schema (#1309)

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

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

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

@@ -0,0 +1,55 @@
+#
+# 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 "ALTER TOPIC cities ADD COLUMN zip INT;"
+      [0x00]
+write flush
+
+read advised zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .completion()
+                                .tag("ALTER_TOPIC")
+                                .build()
+                              .build()}
+
+read advised zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .ready()
+                                .status("IDLE")
+                                .build()
+                              .build()}
+
+write close
+read closed

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

@@ -0,0 +1,56 @@
+#
+# 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  "ALTER TOPIC cities ADD COLUMN zip INT;"
+      [0x00]
+
+write advise zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .completion()
+                                .tag("ALTER_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.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/pgsql/alter.topic.modify.column.rejected/client.rpt

@@ -0,0 +1,57 @@
+#
+# 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 "ALTER TOPIC cities ALTER COLUMN name TYPE VARCHAR(100);"
+      [0x00]
+write flush
+
+read advised zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .error()
+                                .severity("ERROR")
+                                .code("XX000")
+                                .message("Unable to alter topic cities")
+                                .build()
+                              .build()}
+
+read advised zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .ready()
+                                .status("IDLE")
+                                .build()
+                              .build()}
+
+write close
+read closed

incubator/binding-pgsql-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/pgsql/alter.topic.modify.column.rejected/server.rpt

@@ -0,0 +1,56 @@
+#
+# 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  "ALTER TOPIC cities ALTER COLUMN name TYPE VARCHAR(100);"
+      [0x00]
+
+write advise zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .error()
+                                .severity("ERROR")
+                                .code("XX000")
+                                .message("Unable to alter topic cities")
+                                .build()
+                              .build()}
+
+write advise zilla:flush ${pgsql:flushEx()
+                              .typeId(zilla:id("pgsql"))
+                              .ready()
+                                .status("IDLE")
+                                .build()
+                              .build()}
+read closed
+write close

incubator/binding-pgsql-kafka.spec/src/test/java/io/aklivity/zilla/specs/binding/pgsql/kafka/streams/PgsqlIT.java

@@ -46,6 +46,26 @@ public void shouldCreateTopic() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Specification({
+        "${app}/alter.topic.add.column/client",
+        "${app}/alter.topic.add.column/server"
+    })
+    public void shouldAlterTopic() throws Exception
+    {
+        k3po.finish();
+    }
+
+    @Test
+    @Specification({
+        "${app}/alter.topic.modify.column.rejected/client",
+        "${app}/alter.topic.modify.column.rejected/server"
+    })
+    public void shouldNotAlterTopicModifyColumn() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Specification({
         "${app}/drop.topic/client",

incubator/binding-pgsql-kafka/pom.xml

@@ -24,7 +24,7 @@
   </licenses>
 
   <properties>
-    <jacoco.coverage.ratio>0.86</jacoco.coverage.ratio>
+    <jacoco.coverage.ratio>0.85</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/schema/PgsqlKafkaAvroSchemaTemplate.java

@@ -14,23 +14,99 @@
  */
 package io.aklivity.zilla.runtime.binding.pgsql.kafka.internal.schema;
 
+import java.util.List;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+
+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 String convertPgsqlTypeToAvro(
-        String pgsqlType)
+    protected static final String DATABASE_PLACEHOLDER = "{database}";
+
+    protected final ObjectMapper mapper = new ObjectMapper();
+
+    protected Object mapSqlTypeToAvroType(
+        String sqlType)
+    {
+        sqlType = sqlType.toUpperCase();
+
+        Object result = null;
+
+        switch (sqlType)
+        {
+        case "INT":
+        case "INTEGER":
+            result = "int";
+            break;
+        case "BIGINT":
+            result = "long";
+            break;
+        case "BOOLEAN":
+            result = "boolean";
+            break;
+        case "FLOAT":
+            result = "float";
+            break;
+        case "DOUBLE":
+            result = "double";
+            break;
+        case "DECIMAL":
+            ObjectNode decimalNode = mapper.createObjectNode();
+            decimalNode.put("type", "bytes");
+            decimalNode.put("logicalType", "decimal");
+            decimalNode.put("precision", 10);
+            decimalNode.put("scale", 2);
+            result = decimalNode;
+            break;
+        case "DATE":
+            ObjectNode dateNode = mapper.createObjectNode();
+            dateNode.put("type", "int");
+            dateNode.put("logicalType", "date");
+            result = dateNode;
+            break;
+        case "TIMESTAMP":
+            ObjectNode timestampNode = mapper.createObjectNode();
+            timestampNode.put("type", "long");
+            timestampNode.put("logicalType", "timestamp-millis");
+            result = timestampNode;
+            break;
+        case "VARCHAR":
+        case "CHAR":
+        case "TEXT":
+        default:
+            result = "string";
+        }
+
+        return result;
+    }
+
+    protected boolean applyAlterations(
+        ArrayNode fields,
+        List<AlterExpression> alterExpressions)
     {
-        return switch (pgsqlType.toLowerCase())
+        boolean applied = true;
+
+        apply:
+        for (AlterExpression alterExpr : alterExpressions)
         {
-        case "varchar", "text", "char", "bpchar" -> "\\\"string\\\"";
-        case "int", "integer", "serial" -> "\\\"int\\\"";
-        case "numeric" -> "\\\"double\\\"";
-        case "bigint", "bigserial" -> "\\\"long\\\"";
-        case "boolean", "bool" -> "\\\"boolean\\\"";
-        case "real", "float4" -> "\\\"float\\\"";
-        case "double", "double precision", "float8" -> "\\\"double\\\"";
-        case "timestamp", "timestampz", "date", "time" ->
-            "{ \\\"type\\\": \\\"long\\\", \\\"logicalTyp\\\": \\\"timestamp-millis\\\" }";
-        default -> null;
-        };
+            if (alterExpr.operation() == Operation.ADD)
+            {
+                ObjectNode newField = mapper.createObjectNode();
+                newField.put("name", alterExpr.columnName());
+                newField.set("type", mapper.valueToTree(mapSqlTypeToAvroType(alterExpr.columnType())));
+                fields.add(newField);
+            }
+            else
+            {
+                applied = false;
+                break apply;
+            }
+        }
+
+        return applied;
     }
 }