Validate if all environment variables are present

Signed-off-by: Marvin Froeder <velo.br@gmail.com>

Author: velo

fink-dockerfile-example/build.gradle

@@ -12,7 +12,7 @@ ext {
     flinkVersion = "1.19.1"
     jdbcVersion = "3.2.0-1.19"
     kafkaVersion = "3.2.0-1.19"
-    sqrlVersion = "0.5.6"
+    sqrlVersion = "0.5.7"
     icebergVersion = "1.6.0"
 }
 

src/main/java/com/datasqrl/EnvironmentVariablesUtils.java

@@ -0,0 +1,48 @@
+package com.datasqrl;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import lombok.experimental.UtilityClass;
+
+@UtilityClass
+public class EnvironmentVariablesUtils {
+
+  private static final Pattern ENVIRONMENT_VARIABLE_PATTERN = Pattern.compile("\\$\\{(.*?)\\}");
+
+  public static String replaceWithEnv(String command, Map<String, String> envVariables) {
+    String substitutedStr = command;
+    StringBuffer result = new StringBuffer();
+    // First pass to replace environment variables
+    Matcher matcher = ENVIRONMENT_VARIABLE_PATTERN.matcher(substitutedStr);
+    while (matcher.find()) {
+      String key = matcher.group(1);
+      String envValue = envVariables.getOrDefault(key, "");
+      matcher.appendReplacement(result, Matcher.quoteReplacement(envValue));
+    }
+    matcher.appendTail(result);
+
+    return result.toString();
+  }
+
+  public void validateEnvironmentVariables(TreeMap<String, String> envVariables, String script) {
+    Matcher matcher = ENVIRONMENT_VARIABLE_PATTERN.matcher(script);
+
+    Set<String> scriptEnvironmentVariables = new TreeSet<>();
+    while (matcher.find()) {
+      scriptEnvironmentVariables.add(matcher.group(1));
+    }
+
+    scriptEnvironmentVariables.removeAll(envVariables.keySet());
+
+    if (!scriptEnvironmentVariables.isEmpty()) {
+      throw new IllegalStateException(
+          String.format(
+              "Could not find the following environment variables: %s",
+              scriptEnvironmentVariables));
+    }
+  }
+}

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

@@ -24,8 +24,10 @@ class SqlExecutor {
       Pattern.compile("SET\\s+'(\\S+)'\\s*=\\s*'(.+)';?", Pattern.CASE_INSENSITIVE);
 
   private final TableEnvironment tableEnv;
+  private final Map<String, String> envVariables;
 
-  public SqlExecutor(Configuration configuration, String udfPath) {
+  public SqlExecutor(
+      Configuration configuration, String udfPath, Map<String, String> envVariables) {
     StreamExecutionEnvironment sEnv;
     try {
       sEnv = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
@@ -44,6 +46,7 @@ public SqlExecutor(Configuration configuration, String udfPath) {
     if (udfPath != null) {
       setupUdfPath(udfPath);
     }
+    this.envVariables = envVariables;
   }
 
   /**
@@ -92,7 +95,8 @@ private TableResult executeStatement(String statement) {
       } else {
         System.out.println(statement);
         log.info("Executing statement:\n{}", statement);
-        tableResult = tableEnv.executeSql(replaceWithEnv(statement));
+        tableResult =
+            tableEnv.executeSql(EnvironmentVariablesUtils.replaceWithEnv(statement, envVariables));
       }
     } catch (Exception e) {
       log.error("Failed to execute statement: {}", statement, e);

src/main/java/com/datasqrl/SqlRunner.java

@@ -18,7 +18,8 @@
 package com.datasqrl;
 
 import java.io.File;
-import java.util.*;
+import java.util.Map;
+import java.util.TreeMap;
 import java.util.concurrent.Callable;
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
@@ -29,7 +30,8 @@
 import org.apache.flink.table.api.TableResult;
 import org.apache.flink.util.FileUtils;
 import picocli.CommandLine;
-import picocli.CommandLine.*;
+import picocli.CommandLine.Command;
+import picocli.CommandLine.Option;
 
 /** Main class for executing SQL scripts using picocli. */
 @Command(
@@ -67,7 +69,7 @@ public class SqlRunner implements Callable<Integer> {
   private String udfPath;
 
   public static void main(String[] args) {
-    int exitCode = new CommandLine(new SqlRunner()).execute(args);
+    var exitCode = new CommandLine(new SqlRunner()).execute(args);
     System.exit(exitCode);
   }
 
@@ -79,22 +81,27 @@ public Integer call() throws Exception {
     }
 
     // Load configuration if configFile is provided
-    Configuration configuration = new Configuration();
+    var configuration = new Configuration();
     if (configFile != null) {
       configuration = loadConfigurationFromYaml(configFile);
     }
 
+    log.info("Environment variables");
+    TreeMap<String, String> envVariables = new TreeMap<>(System.getenv());
+    envVariables.forEach((name, value) -> log.info("{}: {}", name, value));
+
     // Initialize SqlExecutor
-    SqlExecutor sqlExecutor = new SqlExecutor(configuration, udfPath);
+    var sqlExecutor = new SqlExecutor(configuration, udfPath, envVariables);
     TableResult tableResult;
     // Input validation and execution logic
     if (sqlFile != null) {
       // Single SQL file mode
-      String script = FileUtils.readFileUtf8(sqlFile);
+      var script = FileUtils.readFileUtf8(sqlFile);
+      EnvironmentVariablesUtils.validateEnvironmentVariables(envVariables, script);
       tableResult = sqlExecutor.executeScript(script);
     } else if (planFile != null) {
       // Compiled plan JSON file
-      String planJson = FileUtils.readFileUtf8(planFile);
+      var planJson = FileUtils.readFileUtf8(planFile);
       planJson = replaceScriptWithEnv(planJson);
 
       tableResult = sqlExecutor.executeCompiledPlan(planJson);
@@ -120,10 +127,10 @@ private String replaceScriptWithEnv(String script) {
   }
 
   public static ObjectMapper getObjectMapper() {
-    ObjectMapper objectMapper = new ObjectMapper();
+    var objectMapper = new ObjectMapper();
 
     // Register the custom deserializer module
-    SimpleModule module = new SimpleModule();
+    var module = new SimpleModule();
     module.addDeserializer(String.class, new JsonEnvVarDeserializer());
     objectMapper.registerModule(module);
     return objectMapper;
@@ -138,8 +145,7 @@ public static ObjectMapper getObjectMapper() {
    */
   private Configuration loadConfigurationFromYaml(File configFile) throws Exception {
     log.info("Loading configuration from {}", configFile.getAbsolutePath());
-    Configuration configuration =
-        GlobalConfiguration.loadConfiguration(configFile.getAbsolutePath());
+    var configuration = GlobalConfiguration.loadConfiguration(configFile.getAbsolutePath());
     return configuration;
   }
 }

src/main/java/com/datasqrl/SqlUtils.java

@@ -2,8 +2,10 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import lombok.experimental.UtilityClass;
 
 /** Utility class for parsing SQL scripts. */
+@UtilityClass
 class SqlUtils {
 
   private static final String STATEMENT_DELIMITER = ";"; // a statement should end with `;`