WIP

incubator/binding-pgsql/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/parser/PgsqlParser.java

@@ -66,6 +66,7 @@ public PgsqlParser()
     public String parseCommand(
         String sql)
     {
+        commandListener.resetCommand();
         parser(sql, commandListener);
         return commandListener.command();
     }

incubator/binding-pgsql/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/parser/listener/SqlCommandListener.java

@@ -19,13 +19,18 @@
 
 public class SqlCommandListener extends PostgreSqlParserBaseListener
 {
-    private String command;
+    private String command = "";
 
     public String command()
     {
         return command;
     }
 
+    public void resetCommand()
+    {
+        command = "";
+    }
+
     @Override
     public void enterCreatestmt(
         PostgreSqlParser.CreatestmtContext ctx)

incubator/binding-risingwave.spec/src/main/scripts/io/aklivity/zilla/specs/binding/risingwave/streams/effective/create.table.with.primary.key.and.includes/client.rpt

@@ -67,11 +67,10 @@ write zilla:data.ext ${pgsql:dataEx()
                               .query()
                                 .build()
                               .build()}
-write "CREATE MATERIALIZED VIEW IF NOT EXISTS cities_view AS"
-      " SELECT id, name, description,"
-      " COALESCE(correlation_id, zilla_correlation_id_header::varchar) as correlation_id,"
-      " COALESCE(identity, zilla_identity_header::varchar) as identity,"
-      " timestamp"
+write "CREATE MATERIALIZED VIEW IF NOT EXISTS cities_view"
+      " AS SELECT id, name, description,"
+      " COALESCE(zilla_correlation_id, zilla_correlation_id_header::varchar) as zilla_correlation_id,"
+      " COALESCE(zilla_identity, zilla_identity_header::varchar) as zilla_identity, timestamp"
       " FROM cities_source;"
       [0x00]
 write flush
@@ -97,7 +96,7 @@ write zilla:data.ext ${pgsql:dataEx()
                               .build()}
 write "CREATE TABLE IF NOT EXISTS cities"
       " (id VARCHAR, name VARCHAR, description VARCHAR,"
-      " correlation_id VARCHAR, identity VARCHAR, timestamp TIMESTAMP,"
+      " zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
       " PRIMARY KEY (id));"
       [0x00]
 write flush
@@ -192,7 +191,7 @@ write zilla:data.ext ${pgsql:dataEx()
                               .build()}
 write "CREATE TOPIC IF NOT EXISTS cities "
       "(id VARCHAR, name VARCHAR, description VARCHAR,"
-      " correlation_id VARCHAR, identity VARCHAR, timestamp TIMESTAMP,"
+      " zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
       " PRIMARY KEY (id));"
       [0x00]
 write flush

incubator/binding-risingwave.spec/src/main/scripts/io/aklivity/zilla/specs/binding/risingwave/streams/effective/create.table.with.primary.key.and.includes/server.rpt

@@ -70,11 +70,10 @@ read zilla:data.ext ${pgsql:dataEx()
                               .query()
                                 .build()
                               .build()}
-read "CREATE MATERIALIZED VIEW IF NOT EXISTS cities_view AS"
-     " SELECT id, name, description,"
-     " COALESCE(correlation_id, zilla_correlation_id_header::varchar) as correlation_id,"
-     " COALESCE(identity, zilla_identity_header::varchar) as identity,"
-     " timestamp"
+read "CREATE MATERIALIZED VIEW IF NOT EXISTS cities_view"
+     " AS SELECT id, name, description,"
+     " COALESCE(zilla_correlation_id, zilla_correlation_id_header::varchar) as zilla_correlation_id,"
+     " COALESCE(zilla_identity, zilla_identity_header::varchar) as zilla_identity, timestamp"
      " FROM cities_source;"
      [0x00]
 
@@ -99,7 +98,7 @@ read zilla:data.ext ${pgsql:dataEx()
                               .build()}
 read "CREATE TABLE IF NOT EXISTS cities"
      " (id VARCHAR, name VARCHAR, description VARCHAR,"
-     " correlation_id VARCHAR, identity VARCHAR, timestamp TIMESTAMP,"
+     " zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
      " PRIMARY KEY (id));"
      [0x00]
 
@@ -193,7 +192,7 @@ read zilla:data.ext ${pgsql:dataEx()
                               .build()}
 read "CREATE TOPIC IF NOT EXISTS cities "
      "(id VARCHAR, name VARCHAR, description VARCHAR,"
-     " correlation_id VARCHAR, identity VARCHAR, timestamp TIMESTAMP,"
+     " zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
      " PRIMARY KEY (id));"
      [0x00]
 

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

@@ -33,10 +33,9 @@ write zilla:data.ext ${pgsql:dataEx()
                                 .build()
                               .build()}
 write "CREATE TABLE IF NOT EXISTS cities "
-      "(id VARCHAR, name VARCHAR, description VARCHAR, PRIMARY KEY (id))\n"
-      "INCLUDE zilla_correlation_id AS correlation_id\n"
-      "INCLUDE zilla_identity AS identity\n"
-      "INCLUDE timestamp as timestamp;"
+      "(id VARCHAR, name VARCHAR, description VARCHAR,"
+      " zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
+      " PRIMARY KEY (id));"
       [0x00]
 
 write flush

incubator/binding-risingwave.spec/src/main/scripts/io/aklivity/zilla/specs/binding/risingwave/streams/pgsql/create.table.with.primary.key.and.includes/server.rpt

@@ -37,10 +37,9 @@ read zilla:data.ext ${pgsql:dataEx()
                                 .build()
                               .build()}
 read  "CREATE TABLE IF NOT EXISTS cities "
-      "(id VARCHAR, name VARCHAR, description VARCHAR, PRIMARY KEY (id))\n"
-      "INCLUDE zilla_correlation_id AS correlation_id\n"
-      "INCLUDE zilla_identity AS identity\n"
-      "INCLUDE timestamp as timestamp;"
+      "(id VARCHAR, name VARCHAR, description VARCHAR,"
+      " zilla_correlation_id VARCHAR, zilla_identity VARCHAR, timestamp TIMESTAMP,"
+      " PRIMARY KEY (id));"
       [0x00]
 
 write advise zilla:flush ${pgsql:flushEx()

incubator/binding-risingwave/src/main/java/io/aklivity/zilla/runtime/binding/risingwave/internal/statement/RisingwaveCreateMaterializedViewTemplate.java

@@ -14,6 +14,7 @@
  */
 package io.aklivity.zilla.runtime.binding.risingwave.internal.statement;
 
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.stream.Collectors;
 
@@ -26,7 +27,7 @@ public class RisingwaveCreateMaterializedViewTemplate extends RisingwaveCommandT
     private final String sqlFormat = """
         CREATE MATERIALIZED VIEW IF NOT EXISTS %s AS %s;\u0000""";
     private final String fieldFormat = "%s, ";
-    private final String includeFormat = "COALESCE(%s, zilla_%s_header::varchar) as %s, ";
+    private final String includeFormat = "COALESCE(%s, %s_header::varchar) as %s, ";
 
     public RisingwaveCreateMaterializedViewTemplate()
     {
@@ -50,25 +51,26 @@ public String generate(
 
         Map<String, String> includes = tableInfo.columns().entrySet().stream()
             .filter(e -> ZILLA_MAPPINGS.containsKey(e.getKey()))
-            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+            .collect(LinkedHashMap::new, (m, e) -> m.put(e.getKey(), e.getValue()), Map::putAll);
 
         if (!includes.isEmpty())
         {
             fieldBuilder.setLength(0);
 
-            tableInfo.columns().values().stream()
+            tableInfo.columns().keySet()
+                .stream()
                 .filter(c -> !ZILLA_MAPPINGS.containsKey(c))
                 .forEach(c -> fieldBuilder.append(String.format(fieldFormat, c)));
 
             includes.forEach((k, v) ->
             {
                 if ("timestamp".equals(k))
                 {
-                    fieldBuilder.append(String.format(fieldFormat, v));
+                    fieldBuilder.append(String.format(fieldFormat, k));
                 }
                 else
                 {
-                    fieldBuilder.append(String.format(includeFormat, v, v, v));
+                    fieldBuilder.append(String.format(includeFormat, k, k, k));
                 }
             });
 

incubator/binding-risingwave/src/main/java/io/aklivity/zilla/runtime/binding/risingwave/internal/statement/RisingwaveCreateSourceTemplate.java

@@ -14,6 +14,7 @@
  */
 package io.aklivity.zilla.runtime.binding.risingwave.internal.statement;
 
+import java.util.LinkedHashMap;
 import java.util.Map;
 import java.util.stream.Collectors;
 
@@ -78,7 +79,7 @@ public String generateTableSource(
         includeBuilder.setLength(0);
         Map<String, String> includes = tableInfo.columns().entrySet().stream()
             .filter(e -> ZILLA_MAPPINGS.containsKey(e.getKey()))
-            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+            .collect(LinkedHashMap::new, (m, e) -> m.put(e.getKey(), e.getValue()), Map::putAll);
 
         if (!includes.isEmpty())
         {
@@ -87,11 +88,11 @@ public String generateTableSource(
             {
                 if ("timestamp".equals(k))
                 {
-                    includeBuilder.append(String.format(ZILLA_MAPPINGS.get(k), v));
+                    includeBuilder.append(String.format(ZILLA_MAPPINGS.get(k), k));
                 }
                 else
                 {
-                    includeBuilder.append(String.format(ZILLA_MAPPINGS.get(k), "zilla_%s_header".formatted(v)));
+                    includeBuilder.append(String.format(ZILLA_MAPPINGS.get(k), "%s_header".formatted(k)));
                 }
             });
             includeBuilder.delete(includeBuilder.length() - 1, includeBuilder.length());

incubator/binding-risingwave/src/main/java/io/aklivity/zilla/runtime/binding/risingwave/internal/statement/RisingwaveCreateTableTemplate.java

@@ -14,9 +14,6 @@
  */
 package io.aklivity.zilla.runtime.binding.risingwave.internal.statement;
 
-import java.util.Map;
-import java.util.stream.Collectors;
-
 import io.aklivity.zilla.runtime.binding.pgsql.parser.module.TableInfo;
 
 public class RisingwaveCreateTableTemplate extends RisingwaveCommandTemplate
@@ -44,13 +41,6 @@ public String generate(
             .forEach((k, v) -> fieldBuilder.append(
                 String.format(fieldFormat, k, v)));
 
-        Map<String, String> includes = tableInfo.columns().entrySet().stream()
-            .filter(e -> ZILLA_MAPPINGS.containsKey(e.getKey()))
-            .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
-
-        includes.forEach((k, v) -> fieldBuilder.append(
-            String.format(fieldFormat, v, ZILLA_INCLUDE_TYPE_MAPPINGS.get(k))));
-
         fieldBuilder.delete(fieldBuilder.length() - 2, fieldBuilder.length());
 
         return String.format(sqlFormat, topic, fieldBuilder, primaryKey);

incubator/binding-risingwave/src/main/java/io/aklivity/zilla/runtime/binding/risingwave/internal/stream/RisingwaveProxyFactory.java

@@ -67,7 +67,7 @@
 
 public final class RisingwaveProxyFactory implements RisingwaveStreamFactory
 {
-    private static final String SPLIT_STATEMENTS = "\"(?<=;)(?!\\x00)\"";
+    private static final String SPLIT_STATEMENTS = "(?<=;)(?!\\x00)";
     private static final int END_OF_FIELD = 0x00;
 
     private static final int FLAGS_INIT = 0x02;

incubator/binding-risingwave/src/test/java/io/aklivity/zilla/runtime/binding/risingwave/internal/statement/RisingwaveCreateMaterializedViewTemplateTest.java

@@ -0,0 +1,78 @@
+package io.aklivity.zilla.runtime.binding.risingwave.internal.statement;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Set;
+
+import org.junit.Ignore;
+import org.junit.Test;
+
+import io.aklivity.zilla.runtime.binding.pgsql.parser.module.TableInfo;
+import io.aklivity.zilla.runtime.binding.pgsql.parser.module.ViewInfo;
+
+public class RisingwaveCreateMaterializedViewTemplateTest
+{
+    private final RisingwaveCreateMaterializedViewTemplate template = new RisingwaveCreateMaterializedViewTemplate();
+
+    @Test
+    public void shouldGenerateMaterializedViewWithValidViewInfo()
+    {
+        ViewInfo viewInfo = new ViewInfo("test_view", "SELECT * FROM test_table");
+        String expectedSQL = """
+            CREATE MATERIALIZED VIEW IF NOT EXISTS test_view AS SELECT * FROM test_table;\u0000""";
+
+        String actualSQL = template.generate(viewInfo);
+
+        assertEquals(expectedSQL, actualSQL);
+    }
+
+    @Ignore("TODO")
+    @Test
+    public void shouldGenerateMaterializedViewWithValidTableInfo()
+    {
+        TableInfo tableInfo = new TableInfo(
+            "test_table",
+                  Map.of("id", "INT", "name", "STRING"),
+                  Set.of("id"));
+        String expectedSQL = """
+            CREATE MATERIALIZED VIEW IF NOT EXISTS test_table_view AS SELECT id, name FROM test_table_source;\u0000""";
+
+        String actualSQL = template.generate(tableInfo);
+
+        assertEquals(expectedSQL, actualSQL);
+    }
+
+    @Test
+    public void shouldGenerateMaterializedViewWithEmptyColumns()
+    {
+        TableInfo tableInfo = new TableInfo("empty_table", Map.of(), Set.of());
+        String expectedSQL = """
+            CREATE MATERIALIZED VIEW IF NOT EXISTS empty_table_view AS SELECT * FROM empty_table_source;\u0000""";
+
+        String actualSQL = template.generate(tableInfo);
+
+        assertEquals(expectedSQL, actualSQL);
+    }
+
+    @Test
+    public void shouldGenerateMaterializedViewWithIncludes()
+    {
+        Map<String, String> columns = new LinkedHashMap<>();
+        columns.put("id", "INT");
+        columns.put("zilla_correlation_id", "VARCHAR");
+        columns.put("zilla_identity", "VARCHAR");
+        columns.put("timestamp", "TIMESTAMP");
+
+        TableInfo tableInfo = new TableInfo("test_table", columns, Set.of("id"));
+        String expectedSQL = "CREATE MATERIALIZED VIEW IF NOT EXISTS test_table_view AS SELECT id," +
+            " COALESCE(zilla_correlation_id, zilla_correlation_id_header::varchar) as zilla_correlation_id," +
+            " COALESCE(zilla_identity, zilla_identity_header::varchar) as zilla_identity, timestamp" +
+            " FROM test_table_source;\u0000";
+
+        String actualSQL = template.generate(tableInfo);
+
+        assertEquals(expectedSQL, actualSQL);
+    }
+}