Test system function service discovery

pom.xml

@@ -278,10 +278,19 @@
               </artifactSet>
               <filters>
                 <filter>
-                  <!-- Do not copy the signatures in the META-INF folder.
-                  Otherwise, this might cause SecurityExceptions when using the JAR. -->
                   <artifact>*:*</artifact>
                   <excludes>
+                    <!--
+  We exclude AutoRegisterSystemFunction from the shaded JAR because this class
+  is loaded via ServiceLoader from the Flink classpath (/opt/flink/lib), not the job JAR.
+  Including it here would result in a NoClassDefFoundError due to Flink's classloader isolation.
+
+  A separate 'service' JAR is built (see maven-jar-plugin) that contains only this interface,
+  and is deployed to Flink's classpath to ensure it is available at runtime for discovery.
+-->
+                    <exclude>com/datasqrl/function/AutoRegisterSystemFunction.class</exclude>
+                    <!-- Do not copy the signatures in the META-INF folder.
+                  Otherwise, this might cause SecurityExceptions when using the JAR. -->
                     <exclude>META-INF/*.SF</exclude>
                     <exclude>META-INF/*.DSA</exclude>
                     <exclude>META-INF/*.RSA</exclude>
@@ -299,6 +308,26 @@
         </executions>
       </plugin>
 
+      <plugin>
+        <artifactId>maven-jar-plugin</artifactId>
+        <version>3.3.0</version>
+        <executions>
+          <execution>
+            <id>service-jar</id>
+            <goals>
+              <goal>jar</goal>
+            </goals>
+            <phase>package</phase>
+            <configuration>
+              <classifier>service</classifier>
+              <includes>
+                <include>com/datasqrl/function/AutoRegisterSystemFunction.class</include>
+              </includes>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+
       <!-- linter for java code, will be applied automatically when files are committed to git -->
       <!-- alternatively, can run with -Pdev to force code format on the whole project -->
       <plugin>
@@ -392,7 +421,7 @@
         <version>5.0.0</version>
         <configuration>
           <skip>${skipTests}</skip>
-          <composeFile>${project.basedir}/target/test-classes/docker-compose.yml</composeFile>
+          <composeFile>${project.basedir}/target/docker-compose.yml</composeFile>
         </configuration>
         <executions>
           <execution>

src/main/java/com/datasqrl/SqlExecutor.java

@@ -57,20 +57,29 @@ public SqlExecutor(Configuration configuration, String udfPath) {
   }
 
   public void setupSystemFunctions() {
+    System.out.println("Setting up automatically registered system functions");
 
-    ServiceLoader<AutoRegisterSystemFunction> standardLibraryFunctions =
-        ServiceLoader.load(AutoRegisterSystemFunction.class);
-
-    standardLibraryFunctions.forEach(
-        function -> {
-          String sql =
-              String.format(
-                  "CREATE TEMPORARY FUNCTION IF NOT EXISTS `%s` AS '%s' LANGUAGE JAVA;",
-                  getFunctionNameFromClass(function.getClass()), function.getClass().getName());
+    try {
+      ServiceLoader<AutoRegisterSystemFunction> standardLibraryFunctions =
+          ServiceLoader.load(AutoRegisterSystemFunction.class);
+
+      standardLibraryFunctions.forEach(
+          function -> {
+            String sql =
+                String.format(
+                    "CREATE TEMPORARY FUNCTION IF NOT EXISTS `%s` AS '%s' LANGUAGE JAVA;",
+                    getFunctionNameFromClass(function.getClass()), function.getClass().getName());
+
+            System.out.println(sql);
+
+            tableEnv.executeSql(sql);
+          });
+    } catch (Throwable e) {
+      e.printStackTrace(System.out);
+      throw new RuntimeException(e);
+    }
 
-          System.out.println(sql);
-          tableEnv.executeSql(sql);
-        });
+    System.out.println("Completed auto function registered system functions");
   }
 
   static String getFunctionNameFromClass(Class clazz) {

src/test/docker/docker-compose.yml

@@ -30,7 +30,7 @@ services:
     ports:
       - "5432:5432"
     volumes:
-      - ./sqrl/postgres-schema.sql:/docker-entrypoint-initdb.d/database-schema.sql:ro
+      - ./test-classes/sqrl/postgres-schema.sql:/docker-entrypoint-initdb.d/database-schema.sql:ro
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U postgres"]
       interval: 1s
@@ -61,9 +61,11 @@ services:
         condition: service_healthy
     volumes:
       - flink_conf:/opt/flink/conf
-      - ./:/opt/flink/usrlib/
-      - ./datasources/:/datasources/
+      - ./test-classes/:/opt/flink/usrlib/
+      - ./test-classes/datasources/:/datasources/
       - /tmp:/tmp
+      - ./flink-sql-runner-1.0.0-SNAPSHOT-service.jar:/opt/flink/lib/flink-sql-runner-service.jar
+      - ./test-classes/systemfunction/target/systemfunction-sample-1.0.0-SNAPSHOT.jar:/opt/flink/lib/systemfunction.jar
     healthcheck:
       test: ["CMD-SHELL", "curl -f http://localhost:8081/ || exit 1"]
       interval: 1s
@@ -95,9 +97,11 @@ services:
         condition: service_healthy
     volumes:
       - flink_conf:/opt/flink/conf
-      - ./:/opt/flink/usrlib/
-      - ./datasources/:/datasources/
+      - ./test-classes/:/opt/flink/usrlib/
+      - ./test-classes/datasources/:/datasources/
       - /tmp:/tmp
+      - ./flink-sql-runner-1.0.0-SNAPSHOT-service.jar:/opt/flink/lib/flink-sql-runner-service.jar
+      - ./test-classes/systemfunction/target/systemfunction-sample-1.0.0-SNAPSHOT.jar:/opt/flink/lib/systemfunction.jar
     restart: always
     deploy:
       resources:

src/test/resources/sql/flink.sql

@@ -114,7 +114,7 @@ WHERE `_rownum` = 1;
 CREATE VIEW `table$2`
 AS
 SELECT *
-FROM (SELECT `customerId`, `cardNo`, `timestamp`, `cardType`, ROW_NUMBER() OVER (PARTITION BY `cardNo` ORDER BY `timestamp` DESC) AS `_rownum`
+FROM (SELECT `customerId`, `cardNo`, `timestamp`, upper(`cardType`) as cardType, ROW_NUMBER() OVER (PARTITION BY `cardNo` ORDER BY `timestamp` DESC) AS `_rownum`
   FROM `cardassignment_1`) AS `t1`
 WHERE `_rownum` = 1;
 
@@ -142,7 +142,7 @@ WHERE `_rownum` = 1;
 CREATE VIEW `table$6`
 AS
 SELECT *
-FROM (SELECT `customerId`, `cardNo`, `timestamp`, `cardType`, ROW_NUMBER() OVER (PARTITION BY `cardNo` ORDER BY `timestamp` DESC) AS `_rownum`
+FROM (SELECT `customerId`, `cardNo`, `timestamp`, upper(`cardType`) as cardType, ROW_NUMBER() OVER (PARTITION BY `cardNo` ORDER BY `timestamp` DESC) AS `_rownum`
   FROM `cardassignment_1`) AS `t1`
 WHERE `_rownum` = 1;
 

src/test/resources/systemfunction/pom.xml

@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    Copyright © 2024 DataSQRL (contact@datasqrl.com)
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+        http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+
+  <groupId>com.datasqrl</groupId>
+  <artifactId>systemfunction-sample</artifactId>
+  <version>1.0.0-SNAPSHOT</version>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.flink</groupId>
+      <artifactId>flink-table-common</artifactId>
+      <version>1.19.2</version>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.datasqrl</groupId>
+      <artifactId>flink-sql-runner</artifactId>
+      <version>1.0.0-SNAPSHOT</version>
+      <classifier>service</classifier>
+      <scope>provided</scope>
+    </dependency>
+    <dependency>
+      <groupId>com.google.auto.service</groupId>
+      <artifactId>auto-service</artifactId>
+      <version>1.0.1</version>
+    </dependency>
+  </dependencies>
+
+</project>

src/test/resources/systemfunction/src/main/java/sample/Upper.java

@@ -0,0 +1,26 @@
+package sample;
+
+import org.apache.flink.table.functions.ScalarFunction;
+
+import com.datasqrl.function.AutoRegisterSystemFunction;
+import com.google.auto.service.AutoService;
+
+/**
+ * Returns an array of substrings by splitting the input string based on the given delimiter.
+ * If the delimiter is not found in the string, the original string is returned as the only element
+ * in the array. If the delimiter is empty, every character in the string is split. If the string or
+ * delimiter is null, a null value is returned. If the delimiter is found at the beginning or end of
+ * the string, or there are contiguous delimiters, then an empty string is added to the array.
+ */
+@AutoService(AutoRegisterSystemFunction.class)
+public class Upper extends ScalarFunction implements AutoRegisterSystemFunction {
+
+  public String eval(String text) {
+    if (text == null) {
+      return null;
+    }
+
+    return text.toUpperCase();
+  }
+
+}