From 4021930770a56a95aac2c207c218cc85402ae864 Mon Sep 17 00:00:00 2001 From: Ferenc Csaky Date: Thu, 30 Apr 2026 17:43:24 +0200 Subject: [PATCH] feat: Add `sql-runner` wrapper script to simplify `flink run` --- README.md | 9 +++-- flink-sql-runner/src/main/docker/Dockerfile | 1 + flink-sql-runner/src/main/docker/sql-runner | 40 +++++++++++++++++++ .../flinkrunner/AbstractITSupport.java | 6 +-- 4 files changed, 49 insertions(+), 7 deletions(-) create mode 100644 flink-sql-runner/src/main/docker/sql-runner diff --git a/README.md b/README.md index 15f9090..0b364cd 100644 --- a/README.md +++ b/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: diff --git a/flink-sql-runner/src/main/docker/Dockerfile b/flink-sql-runner/src/main/docker/Dockerfile index 9a5a9bb..1cf572d 100644 --- a/flink-sql-runner/src/main/docker/Dockerfile +++ b/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 \ diff --git a/flink-sql-runner/src/main/docker/sql-runner b/flink-sql-runner/src/main/docker/sql-runner new file mode 100644 index 0000000..e9d3fa7 --- /dev/null +++ b/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[@]}" diff --git a/flink-sql-runner/src/test/java/com/datasqrl/flinkrunner/AbstractITSupport.java b/flink-sql-runner/src/test/java/com/datasqrl/flinkrunner/AbstractITSupport.java index 87e942b..eea663d 100644 --- a/flink-sql-runner/src/test/java/com/datasqrl/flinkrunner/AbstractITSupport.java +++ b/flink-sql-runner/src/test/java/com/datasqrl/flinkrunner/AbstractITSupport.java @@ -158,16 +158,14 @@ protected String flinkRun(List sqlRunnerArgs) throws Exception { protected String flinkRun(List 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]));