WIP

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

@@ -14,14 +14,19 @@
  */
 package io.aklivity.zilla.runtime.binding.pgsql.parser;
 
+import java.util.Set;
+
 import org.antlr.v4.runtime.BailErrorStrategy;
 import org.antlr.v4.runtime.CharStream;
 import org.antlr.v4.runtime.CharStreams;
 import org.antlr.v4.runtime.CommonTokenStream;
 import org.antlr.v4.runtime.tree.ParseTreeWalker;
 
-import io.aklivity.zilla.runtime.binding.pgsql.parser.listener.SqlTableCommandListener;
+import io.aklivity.zilla.runtime.binding.pgsql.parser.listener.SqlCreateMaterializedViewListener;
+import io.aklivity.zilla.runtime.binding.pgsql.parser.listener.SqlCreateTableListener;
+import io.aklivity.zilla.runtime.binding.pgsql.parser.listener.SqlDropListener;
 import io.aklivity.zilla.runtime.binding.pgsql.parser.module.TableInfo;
+import io.aklivity.zilla.runtime.binding.pgsql.parser.module.ViewInfo;
 
 public final class PgsqlParser
 {
@@ -30,7 +35,9 @@ public final class PgsqlParser
     private final PostgreSqlLexer lexer;
     private final CommonTokenStream tokens;
     private final PostgreSqlParser parser;
-    private final SqlTableCommandListener tableCommand;
+    private final SqlCreateTableListener createTableListener;
+    private final SqlCreateMaterializedViewListener createMaterializedViewListener;
+    private final SqlDropListener dropListener;
 
     public PgsqlParser()
     {
@@ -39,12 +46,36 @@ public PgsqlParser()
         this.lexer = new PostgreSqlLexer(null);
         this.parser = new PostgreSqlParser(null);
         this.tokens = new CommonTokenStream(lexer);
-        this.tableCommand = new SqlTableCommandListener();
+        this.createTableListener = new SqlCreateTableListener();
+        this.createMaterializedViewListener = new SqlCreateMaterializedViewListener();
+        this.dropListener = new SqlDropListener();
         parser.setErrorHandler(errorStrategy);
     }
 
-    public TableInfo parseTable(
+    public TableInfo parseCreateTable(
         String sql)
+    {
+        parser(sql, createTableListener);
+        return createTableListener.tableInfo();
+    }
+
+    public ViewInfo parseCreateMaterializedView(
+        String sql)
+    {
+        parser(sql, createMaterializedViewListener);
+        return createMaterializedViewListener.viewInfo();
+    }
+
+    public Set<String> parseDrop(
+        String sql)
+    {
+        parser(sql, dropListener);
+        return dropListener.drops();
+    }
+
+    private void parser(
+        String sql,
+        PostgreSqlParserBaseListener listener)
     {
         CharStream input = CharStreams.fromString(sql);
         lexer.reset();
@@ -53,8 +84,6 @@ public TableInfo parseTable(
         tokens.setTokenSource(lexer);
         parser.setTokenStream(tokens);
 
-        walker.walk(tableCommand, parser.root());
-
-        return tableCommand.tableInfo();
+        walker.walk(listener, parser.root());
     }
 }

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

@@ -0,0 +1,44 @@
+/*
+ * 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.parser.listener;
+
+import io.aklivity.zilla.runtime.binding.pgsql.parser.PostgreSqlParser;
+import io.aklivity.zilla.runtime.binding.pgsql.parser.PostgreSqlParserBaseListener;
+import io.aklivity.zilla.runtime.binding.pgsql.parser.module.ViewInfo;
+
+public class SqlCreateMaterializedViewListener extends PostgreSqlParserBaseListener
+{
+    private String name;
+    private String select;
+
+    public ViewInfo viewInfo()
+    {
+        return new ViewInfo(name, select);
+    }
+
+    @Override
+    public void enterCreatematviewstmt(
+        PostgreSqlParser.CreatematviewstmtContext ctx)
+    {
+        name = ctx.create_mv_target().qualified_name().getText();
+    }
+
+    @Override
+    public void enterSelectstmt(
+        PostgreSqlParser.SelectstmtContext ctx)
+    {
+        select = ctx.getText();
+    }
+}

incubator/binding-pgsql/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/parser/listener/SqlTableCommandListener.java → incubator/binding-pgsql/src/main/java/io/aklivity/zilla/runtime/binding/pgsql/parser/listener/SqlCreateTableListener.java

@@ -24,22 +24,22 @@
 import io.aklivity.zilla.runtime.binding.pgsql.parser.PostgreSqlParserBaseListener;
 import io.aklivity.zilla.runtime.binding.pgsql.parser.module.TableInfo;
 
-public class SqlTableCommandListener extends PostgreSqlParserBaseListener
+public class SqlCreateTableListener extends PostgreSqlParserBaseListener
 {
-    private String tableName;
+    private String name;
     private final Map<String, String> columns = new Object2ObjectHashMap<>();
     private final Set<String> primaryKeys = new ObjectHashSet<>();
 
     public TableInfo tableInfo()
     {
-        return new TableInfo(tableName, columns, primaryKeys);
+        return new TableInfo(name, columns, primaryKeys);
     }
 
     @Override
     public void enterQualified_name(
         PostgreSqlParser.Qualified_nameContext ctx)
     {
-        tableName = ctx.getText();
+        name = ctx.getText();
     }
 
     @Override
@@ -49,30 +49,34 @@ public void enterCreatestmt(
         columns.clear();
         primaryKeys.clear();
 
-        for (PostgreSqlParser.TableelementContext tableElement : ctx.opttableelementlist().tableelementlist().tableelement())
+        if (ctx.opttableelementlist().tableelementlist() != null)
         {
-            if (tableElement.columnDef() != null)
+            for (PostgreSqlParser.TableelementContext tableElement : ctx.opttableelementlist().tableelementlist().tableelement())
             {
-                String columnName = tableElement.columnDef().colid().getText();
-                String dataType = tableElement.columnDef().typename().getText();
-                columns.put(columnName, dataType);
-
-                for (PostgreSqlParser.ColconstraintContext constraint : tableElement.columnDef().colquallist().colconstraint())
+                if (tableElement.columnDef() != null)
                 {
-                    if (constraint.colconstraintelem().PRIMARY() != null &&
-                        constraint.colconstraintelem().KEY() != null)
+                    String columnName = tableElement.columnDef().colid().getText();
+                    String dataType = tableElement.columnDef().typename().getText();
+                    columns.put(columnName, dataType);
+
+                    for (PostgreSqlParser.ColconstraintContext constraint :
+                        tableElement.columnDef().colquallist().colconstraint())
                     {
-                        primaryKeys.add(columnName);
+                        if (constraint.colconstraintelem().PRIMARY() != null &&
+                            constraint.colconstraintelem().KEY() != null)
+                        {
+                            primaryKeys.add(columnName);
+                        }
                     }
                 }
-            }
-            else if (tableElement.tableconstraint() != null)
-            {
-                if (tableElement.tableconstraint().constraintelem().PRIMARY() != null &&
-                    tableElement.tableconstraint().constraintelem().KEY() != null)
+                else if (tableElement.tableconstraint() != null)
                 {
-                    tableElement.tableconstraint().constraintelem().columnlist().columnElem().forEach(
-                        column -> primaryKeys.add(column.getText()));
+                    if (tableElement.tableconstraint().constraintelem().PRIMARY() != null &&
+                        tableElement.tableconstraint().constraintelem().KEY() != null)
+                    {
+                        tableElement.tableconstraint().constraintelem().columnlist().columnElem().forEach(
+                            column -> primaryKeys.add(column.getText()));
+                    }
                 }
             }
         }

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

@@ -0,0 +1,41 @@
+/*
+ * 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.parser.listener;
+
+import java.util.Set;
+
+import org.agrona.collections.ObjectHashSet;
+
+import io.aklivity.zilla.runtime.binding.pgsql.parser.PostgreSqlParser;
+import io.aklivity.zilla.runtime.binding.pgsql.parser.PostgreSqlParserBaseListener;
+
+public class SqlDropListener extends PostgreSqlParserBaseListener
+{
+    private final Set<String> drops = new ObjectHashSet<>();
+
+    public Set<String> drops()
+    {
+        return drops;
+    }
+
+    @Override
+    public void enterDropstmt(
+        PostgreSqlParser.DropstmtContext ctx)
+    {
+        drops.clear();
+
+        ctx.any_name_list().any_name().forEach(name -> drops.add(name.getText()));
+    }
+}

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

@@ -4,7 +4,7 @@
 import java.util.Set;
 
 public record TableInfo(
-        String tableName,
+        String name,
         Map<String, String> columns,
         Set<String> primaryKeys)
 {

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

@@ -0,0 +1,7 @@
+package io.aklivity.zilla.runtime.binding.pgsql.parser.module;
+
+public record ViewInfo(
+        String name,
+        String select)
+{
+}