WIP

incubator/binding-pgsql/src/main/antlr4/io/aklivity/zilla/runtime/binding/pgsql/parser/PostgreSqlLexer.g4

@@ -40,6 +40,8 @@ options {
 
 Dollar: '$';
 
+ZFUNCTION_S: '$$';
+
 OPEN_PAREN: '(';
 
 CLOSE_PAREN: ')';

incubator/binding-pgsql/src/main/antlr4/io/aklivity/zilla/runtime/binding/pgsql/parser/PostgreSqlParser.g4

@@ -2043,9 +2043,8 @@ opt_nulls_order
     ;
 
 createzfunctionstmt
-    : CREATE opt_or_replace ZFUNCTION func_name func_args_with_defaults (
-        RETURNS (func_return | TABLE OPEN_PAREN table_func_column_list CLOSE_PAREN)
-    )? createfunc_opt_list
+    : CREATE ZFUNCTION func_name func_args_with_defaults
+      RETURNS TABLE OPEN_PAREN table_func_column_list CLOSE_PAREN LANGUAGE SQL_P AS ZFUNCTION_S selectstmt SEMI ZFUNCTION_S
     ;
 
 createfunctionstmt

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

@@ -23,6 +23,7 @@
 import io.aklivity.zilla.runtime.binding.pgsql.parser.PostgreSqlParserBaseListener;
 import io.aklivity.zilla.runtime.binding.pgsql.parser.model.CreateZfunction;
 import io.aklivity.zilla.runtime.binding.pgsql.parser.model.FunctionArgument;
+import io.aklivity.zilla.runtime.binding.pgsql.parser.model.Select;
 
 public class SqlCreateZfunctionListener extends PostgreSqlParserBaseListener
 {
@@ -33,22 +34,46 @@ public class SqlCreateZfunctionListener extends PostgreSqlParserBaseListener
     private final List<FunctionArgument> arguments;
     private final TokenStream tokens;
 
+    private final List<String> columns = new ArrayList<>();
+
+    private String table;
+    private String whereClause;
+    private String groupByClause;
+    private String orderByClause;
+    private String limitClause;
+    private String offsetClause;
+
     private String schema;
     private String name;
-    private String asFunction;
-    private String language;
 
     public SqlCreateZfunctionListener(
         TokenStream tokens)
     {
         this.returnTypes = new ArrayList<>();
         this.arguments = new ArrayList<>();
+
+        this.columns.clear();
+
         this.tokens = tokens;
     }
 
     public CreateZfunction zfunction()
     {
-        return new CreateZfunction(schema, name, arguments, returnTypes, asFunction, language);
+        final Select select = new Select(
+            columns,
+            table,
+            whereClause,
+            groupByClause,
+            orderByClause,
+            limitClause,
+            offsetClause);
+
+        return new CreateZfunction(
+            schema,
+            name,
+            arguments,
+            returnTypes,
+            select);
     }
 
     @Override
@@ -57,10 +82,16 @@ public void enterRoot(
     {
         schema = null;
         name = null;
-        asFunction = null;
-        language = null;
         arguments.clear();
         returnTypes.clear();
+
+        columns.clear();
+        table = null;
+        whereClause = null;
+        groupByClause = null;
+        orderByClause = null;
+        limitClause = null;
+        offsetClause = null;
     }
 
     @Override
@@ -92,16 +123,54 @@ public void enterTable_func_column(
     }
 
     @Override
-    public void enterFunc_as(
-        PostgreSqlParser.Func_asContext ctx)
+    public void enterTarget_list(
+        PostgreSqlParser.Target_listContext ctx)
+    {
+        ctx.target_el().forEach(c -> columns.add(c.getText()));
+    }
+
+    @Override
+    public void enterFrom_clause(
+        PostgreSqlParser.From_clauseContext ctx)
+    {
+        if (ctx.from_list() != null)
+        {
+            table = ctx.from_list().getText();
+        }
+    }
+
+    @Override
+    public void enterWhere_clause(
+        PostgreSqlParser.Where_clauseContext ctx)
+    {
+        whereClause = ctx.getText();
+    }
+
+    @Override
+    public void enterGroup_clause(
+        PostgreSqlParser.Group_clauseContext ctx)
+    {
+        groupByClause = ctx.getText();
+    }
+
+    @Override
+    public void enterSort_clause(
+        PostgreSqlParser.Sort_clauseContext ctx)
+    {
+        orderByClause = ctx.getText();
+    }
+
+    @Override
+    public void enterLimit_clause(
+        PostgreSqlParser.Limit_clauseContext ctx)
     {
-        asFunction = ctx.getText();
+        limitClause = ctx.getText();
     }
 
     @Override
-    public void enterNonreservedword_or_sconst(
-        PostgreSqlParser.Nonreservedword_or_sconstContext ctx)
+    public void enterOffset_clause(
+        PostgreSqlParser.Offset_clauseContext ctx)
     {
-        language = ctx.getText();
+        offsetClause = ctx.getText();
     }
 }

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

@@ -21,7 +21,6 @@ public record CreateZfunction(
     String name,
     List<FunctionArgument> arguments,
     List<FunctionArgument> returnTypes,
-    String asFunction,
-    String language)
+    Select select)
 {
 }

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

@@ -0,0 +1,28 @@
+/*
+ * 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.
+ */
+package io.aklivity.zilla.runtime.binding.pgsql.parser.model;
+
+import java.util.List;
+
+public record Select(
+    List<String> columns,
+    String table,
+    String whereClause,
+    String groupByClause,
+    String orderByClause,
+    String limitClause,
+    String offsetClause)
+{
+}

incubator/binding-pgsql/src/test/java/io/aklivity/zilla/runtime/binding/pgsql/parser/PgsqlParserTest.java

@@ -511,59 +511,41 @@ public void shouldParseShowZtables()
     public void shouldParseCreateZfunctionWithTableReturnType()
     {
         String sql = """
-            CREATE ZFUNCTION send_payment_handler(
-                type VARCHAR,
-                user_id VARCHAR,
-                random VARCHAR,
-                amount DOUBLE PRECISION,
-                notes VARCHAR)
-              RETURNS TABLE(
-                type VARCHAR,
-                user_id VARCHAR,
-                request_id VARCHAR,
-                amount DOUBLE PRECISION,
-                notes VARCHAR)
-              LANGUAGE SQL AS $$
-                SELECT
-                    CASE
-                        WHEN balance >= amount THEN "PaymentSent"
-                        ELSE "PaymentDeclined"
-                    END AS type,
-                    user_id,
-                    request_id,
-                    amount,
-                    balance,
-                    notes
-                FROM balance as b WHERE b.user_id = user_id;
-                $$
-            """;
+           CREATE ZFUNCTION send_payment_handler(
+             user_id VARCHAR,
+             amount DOUBLE PRECISION)
+            RETURNS TABLE(
+             event VARCHAR,
+             user_id VARCHAR,
+             amount DOUBLE PRECISION)
+            LANGUAGE SQL AS $$
+             SELECT
+                 CASE
+                     WHEN balance >= args.amount THEN "PaymentSent"
+                     ELSE "PaymentDeclined"
+                 END AS event,
+                 args.user_id,
+                 args.amount
+             FROM balance WHERE user_id = args.user_id;
+            $$
+           """;
         CreateZfunction function = parser.parseCreateZfunction(sql);
         assertNotNull(function);
 
         assertEquals("send_payment_handler", function.name());
-        assertEquals(5, function.arguments().size());
-        assertEquals("type", function.arguments().get(0).name());
+        assertEquals(2, function.arguments().size());
+        assertEquals("user_id", function.arguments().get(0).name());
         assertEquals("VARCHAR", function.arguments().get(0).type());
-        assertEquals("user_id", function.arguments().get(1).name());
-        assertEquals("VARCHAR", function.arguments().get(1).type());
-        assertEquals("random", function.arguments().get(2).name());
-        assertEquals("VARCHAR", function.arguments().get(2).type());
-        assertEquals("amount", function.arguments().get(3).name());
-        assertEquals("DOUBLE PRECISION", function.arguments().get(3).type());
-        assertEquals("notes", function.arguments().get(4).name());
-        assertEquals("VARCHAR", function.arguments().get(4).type());
-
-        assertEquals(5, function.returnTypes().size());
-        assertEquals("type", function.returnTypes().get(0).name());
+        assertEquals("amount", function.arguments().get(1).name());
+        assertEquals("DOUBLE PRECISION", function.arguments().get(1).type());
+
+        assertEquals(3, function.returnTypes().size());
+        assertEquals("event", function.returnTypes().get(0).name());
         assertEquals("VARCHAR", function.returnTypes().get(0).type());
         assertEquals("user_id", function.returnTypes().get(1).name());
         assertEquals("VARCHAR", function.returnTypes().get(1).type());
-        assertEquals("request_id", function.returnTypes().get(2).name());
-        assertEquals("VARCHAR", function.returnTypes().get(2).type());
-        assertEquals("amount", function.returnTypes().get(3).name());
-        assertEquals("DOUBLE PRECISION", function.returnTypes().get(3).type());
-        assertEquals("notes", function.returnTypes().get(4).name());
-        assertEquals("VARCHAR", function.returnTypes().get(4).type());
+        assertEquals("amount", function.returnTypes().get(2).name());
+        assertEquals("DOUBLE PRECISION", function.returnTypes().get(2).type());
     }
 
     @Test

incubator/binding-risingwave.spec/src/main/scripts/io/aklivity/zilla/specs/binding/risingwave/streams/effective/create.zstream/server.rpt

@@ -69,7 +69,7 @@ write zilla:data.ext ${pgsql:dataEx()
 write [0x00 0x02]                                                 # Field Count
       [0x00 0x00 0x00 0x06]                                       # Length
       [0x63 0x69 0x74 0x69 0x65 0x73]                             # Data
-      [0x00 0x00 0x01 0x73]                                       # Length
+      [0x00 0x00 0x01 0x80]                                       # Length
       "CREATE ZFUNCTION send_payment_handler(\n"                  # Data
       " user_id VARCHAR,\n"
       " amount DOUBLE PRECISION\n"
@@ -80,12 +80,12 @@ write [0x00 0x02]                                                 # Field Count
       "LANGUAGE SQL AS $$\n"
       " SELECT\n"
       "     CASE\n"
-      "         WHEN balance >= amount THEN \"PaymentSent\"\n"
+      "         WHEN balance >= args.amount THEN \"PaymentSent\"\n"
       "         ELSE \"PaymentDeclined\"\n"
       "     END AS event,\n"
-      "     user_id,\n"
-      "     amount,\n"
-      " FROM balance as b WHERE b.user_id = user_id;\n"
+      "     args.user_id,\n"
+      "     args.amount,\n"
+      " FROM balance WHERE user_id = args.user_id;\n"
       "$$"
       [0x00]