IT fix

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

@@ -32,9 +32,10 @@ write zilla:data.ext ${pgsql:dataEx()
                               .query()
                                 .build()
                               .build()}
-write "CREATE FUNCTION series ( int ) RETURNS TABLE ( x int ) "
-      "AS series "
-      "LANGUAGE java "
+write "CREATE FUNCTION series(int)\n"
+      "RETURNS TABLE (x int)\n"
+      "AS series\n"
+      "LANGUAGE java\n"
       "USING LINK 'http://localhost:8815';"
       [0x00]
 

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

@@ -34,9 +34,10 @@ read zilla:data.ext ${pgsql:dataEx()
                               .query()
                                 .build()
                               .build()}
-read "CREATE FUNCTION series ( int ) RETURNS TABLE ( x int ) "
-     "AS series "
-     "LANGUAGE java "
+read "CREATE FUNCTION series(int)\n"
+     "RETURNS TABLE (x int)\n"
+     "AS series\n"
+     "LANGUAGE java\n"
      "USING LINK 'http://localhost:8815';"
      [0x00]
 

incubator/binding-risingwave/src/main/java/io/aklivity/zilla/runtime/binding/risingwave/internal/config/RisingwaveBindingConfig.java

@@ -22,6 +22,7 @@
 import org.agrona.DirectBuffer;
 
 import io.aklivity.zilla.runtime.binding.risingwave.config.RisingwaveOptionsConfig;
+import io.aklivity.zilla.runtime.binding.risingwave.config.RisingwaveUdfConfig;
 import io.aklivity.zilla.runtime.binding.risingwave.internal.RisingwaveConfiguration;
 import io.aklivity.zilla.runtime.binding.risingwave.internal.statement.RisingwaveCreateFunctionTemplate;
 import io.aklivity.zilla.runtime.binding.risingwave.internal.statement.RisingwaveCreateMaterializedViewTemplate;
@@ -61,20 +62,34 @@ public RisingwaveBindingConfig(
         this.kind = binding.kind;
         this.routes = binding.routes.stream().map(RisingwaveRouteConfig::new).collect(toList());
 
-        final CatalogedConfig cataloged = options.kafka.format.cataloged.get(0);
-        cataloged.id = binding.resolveId.applyAsLong(cataloged.name);
+        String bootstrapServer = null;
+        String location = null;
+        RisingwaveUdfConfig udf = null;
 
-        final CatalogHandler catalogHandler = supplyCatalog.apply(cataloged.id);
-        this.createTable = new RisingwaveCreateTableTemplate(options.kafka.properties.bootstrapServer,
-            catalogHandler.location(), config.kafkaScanStartupTimestampMillis());
-        this.createSource = new RisingwaveCreateSourceTemplate(options.kafka.properties.bootstrapServer,
-            catalogHandler.location(), config.kafkaScanStartupTimestampMillis());
-        this.createSink = new RisingwaveCreateSinkTemplate(
-            options.kafka.properties.bootstrapServer, catalogHandler.location());
+        if (options.kafka != null)
+        {
+            final CatalogedConfig cataloged = options.kafka.format.cataloged.get(0);
+            cataloged.id = binding.resolveId.applyAsLong(cataloged.name);
+
+            final CatalogHandler catalogHandler = supplyCatalog.apply(cataloged.id);
+            bootstrapServer = options.kafka.properties.bootstrapServer;
+            location = catalogHandler.location();
+        }
+
+        if (options.udfs != null)
+        {
+            udf = options.udfs.get(0);
+        }
+
+        this.createTable = new RisingwaveCreateTableTemplate(bootstrapServer,
+            location, config.kafkaScanStartupTimestampMillis());
+        this.createSource = new RisingwaveCreateSourceTemplate(bootstrapServer,
+            location, config.kafkaScanStartupTimestampMillis());
+        this.createSink = new RisingwaveCreateSinkTemplate(bootstrapServer, location);
         this.createTopic = new RisingwaveCreateTopicTemplate();
         this.createView = new RisingwaveCreateMaterializedViewTemplate();
         this.describeView = new RisingwaveDescribeMaterializedViewTemplate();
-        this.createFunction = new RisingwaveCreateFunctionTemplate(options.udfs.get(0));
+        this.createFunction = new RisingwaveCreateFunctionTemplate(udf);
     }
 
     public RisingwaveRouteConfig resolve(

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

@@ -22,6 +22,13 @@
 
 public class RisingwaveCreateFunctionTemplate extends RisingwaveCommandTemplate
 {
+    private final String sqlFormat = """
+        CREATE FUNCTION %s(%s)
+        RETURNS %s
+        AS %s
+        LANGUAGE %s
+        USING LINK '%s';\u0000""";
+
     private final RisingwaveUdfConfig udf;
 
     public RisingwaveCreateFunctionTemplate(
@@ -35,19 +42,45 @@ public String generate(
     {
         CreateFunction createFunction = (CreateFunction) statement;
         List<String> parts = createFunction.getFunctionDeclarationParts();
-        if (!parts.stream().anyMatch(item -> item.equalsIgnoreCase("LANGUAGE")))
-        {
-            createFunction.addFunctionDeclarationParts("LANGUAGE", udf.language);
-        }
 
-        if (!parts.stream().anyMatch(item -> item.equalsIgnoreCase("LINK")))
+        String functionName = parts.get(0);
+
+        int paramStartIndex = parts.indexOf("(");
+        int paramEndIndex = parts.indexOf(")");
+        String parameters = paramStartIndex >= 0 && paramEndIndex > paramStartIndex
+            ? String.join(" ", parts.subList(paramStartIndex + 1, paramEndIndex))
+            : "";
+
+        int returnsIndex = parts.indexOf("RETURNS");
+        String returnType = returnsIndex >= 0 ? parts.get(returnsIndex + 1) : "";
+
+        if ("TABLE".equalsIgnoreCase(returnType))
         {
-            createFunction.addFunctionDeclarationParts("USING", "LINK", "'%s'".formatted(udf.server));
+            int tableStartIndex = -1;
+            int tableEndIndex = -1;
+            for (int i = returnsIndex; i < parts.size(); i++)
+            {
+                if (parts.get(i).equals("("))
+                {
+                    tableStartIndex = i;
+                }
+                else if (parts.get(i).equals(")"))
+                {
+                    tableEndIndex = i;
+                    break;
+                }
+            }
+
+            if (tableStartIndex >= 0 && tableEndIndex > tableStartIndex)
+            {
+                String tableDefinition = String.join(" ", parts.subList(tableStartIndex + 1, tableEndIndex));
+                returnType += " (" + tableDefinition + ")";
+            }
         }
 
-        parts.removeIf(part -> part.equals(";"));
-        createFunction.addFunctionDeclarationParts(";");
+        int asIndex = parts.indexOf("AS");
+        String body = asIndex >= 0 ? parts.get(asIndex + 1) : "";
 
-        return createFunction.toString();
+        return sqlFormat.formatted(functionName, parameters, returnType, body, udf.language, udf.server);
     }
 }

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

@@ -1690,10 +1690,8 @@ else if (commandType.equals(RisingwaveCommandType.CREATE_FUNCTION_COMMAND))
                 {
                     length -= Byte.BYTES;
                 }
-                inputStream.wrap(buffer, offset, length);
-                statement = parserManager.parse(reader);
-                /*String sql = buffer.getStringWithoutLengthUtf8(offset, length);
-                statement = parserManager.parse(new StringReader(sql));*/
+                String sql = buffer.getStringWithoutLengthUtf8(offset, length);
+                statement = parserManager.parse(new StringReader(sql));
             }
             else
             {
@@ -1703,7 +1701,6 @@ else if (commandType.equals(RisingwaveCommandType.CREATE_FUNCTION_COMMAND))
         }
         catch (Exception ignored)
         {
-            ignored.printStackTrace();
         }
 
         return statement;