feat: Add sql-runner wrapper script to simplify flink run (#321)

Author: ferenc-csaky

README.md

@@ -78,14 +78,17 @@ docker run -d --rm -it \
 In a separate terminal, run:
 
 ```bash
-docker exec -it runner flink run flink-sql-runner.jar --sqlfile /flink/sql/flink.sql
+docker exec -it runner sql-runner --sqlfile /flink/sql/flink.sql
 ```
 
 The job will be submitted to the embedded JobManager and executed using the local TaskManager.
 
 > [!NOTE]  
-> The `flink-sql-runner.jar` is a symlink placed in the Flink root directory (`/opt/flink`) for easier access, but the actual file resides in its own plugin directory: `/opt/flink/plugins/flink-sql-runner`.
-> It is possible to add any Flink arguments or run any accessible JAR, just like with a vanilla `flink run` command.
+> The [`sql-runner`](https://github.com/DataSQRL/flink-sql-runner/blob/main/flink-sql-runner/src/main/docker/sql-runner) wrapper submits the runner through `flink run` so you do not need to specify the main class or placeholder JAR manually.
+> The placeholder JAR is necessary because `sql-runner.jar` is shipped under Flink's `lib/` fodler, while the Flink CLI still requires a job JAR argument for `flink run`.
+> By default, all arguments are passed to the SQL runner, for example `sql-runner --sqlfile /flink/sql/flink.sql`.
+> To pass options to `flink run`, put them before `--`; arguments after `--` are passed to the SQL runner, for example `sql-runner -s /path/to/savepoint -- --planfile /flink/plan.json`.
+> You can also omit using `sql-runner` and directly use `flink run`, just make sure to pass the main class and use the `noop.jar`.
 
 4\. Inspect output
 If your SQL uses the print connector as a sink, you can check logs via:

flink-sql-runner/src/main/docker/Dockerfile

@@ -25,6 +25,7 @@ COPY iceberg-flink-runtime-*.jar /opt/flink/lib
 COPY iceberg-aws-bundle-*.jar /opt/flink/lib
 COPY stdlib-utils-*.jar /opt/flink/lib
 COPY flink-sql-runner.uber.jar /opt/flink/lib/sql-runner.uber.jar
+COPY --chmod=755 sql-runner /opt/flink/bin/sql-runner
 COPY --chmod=755 entrypoint.sh /entrypoint.sh
 
 RUN rm -rf /opt/flink/lib/flink-table-planner-loader-*.jar \

flink-sql-runner/src/main/docker/sql-runner

@@ -0,0 +1,40 @@
+#!/usr/bin/env bash
+#
+# Copyright © 2026 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.
+#
+
+flink_args=()
+runner_args=()
+split_seen=false
+
+for arg in "$@"; do
+  if [[ "$split_seen" == false && "$arg" == "--" ]]; then
+    split_seen=true
+    continue
+  fi
+
+  if [[ "$split_seen" == true ]]; then
+    runner_args+=("$arg")
+  else
+    flink_args+=("$arg")
+  fi
+done
+
+if [[ "$split_seen" == false ]]; then
+  runner_args=("${flink_args[@]}")
+  flink_args=()
+fi
+
+exec flink run "${flink_args[@]}" -c com.datasqrl.flinkrunner.CliRunner /opt/flink/noop.jar "${runner_args[@]}"

flink-sql-runner/src/test/java/com/datasqrl/flinkrunner/AbstractITSupport.java

@@ -158,16 +158,14 @@ protected String flinkRun(List<String> sqlRunnerArgs) throws Exception {
 
   protected String flinkRun(List<String> sqlRunnerArgs, @Nullable String savepointPath)
       throws Exception {
-    var execCmd = new ArrayList<>(List.of("flink", "run"));
+    var execCmd = new ArrayList<>(List.of("sql-runner"));
 
     if (savepointPath != null) {
       execCmd.add("-s");
       execCmd.add(savepointPath);
+      execCmd.add("--");
     }
 
-    execCmd.add("-c");
-    execCmd.add("com.datasqrl.flinkrunner.CliRunner");
-    execCmd.add("noop.jar");
     execCmd.addAll(sqlRunnerArgs);
 
     var execRes = flinkContainer.execInContainer(execCmd.toArray(new String[0]));