Skip to content

Commit 2167c2d

Browse files
velovelo
committed
Validate if all environment variables are present
Signed-off-by: Marvin Froeder <velo.br@gmail.com>
1 parent 4bf8ef5 commit 2167c2d

File tree

5 files changed

+123
-56
lines changed

5 files changed

+123
-56
lines changed

‎fink-dockerfile-example/build.gradle

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ ext {
1212
flinkVersion = "1.19.1"
1313
jdbcVersion = "3.2.0-1.19"
1414
kafkaVersion = "3.2.0-1.19"
15-
sqrlVersion = "0.5.6"
15+
sqrlVersion = "0.5.7"
1616
icebergVersion = "1.6.0"
1717
}
1818

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

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
package com.datasqrl;
2+
3+
import java.util.Map;
4+
import java.util.Set;
5+
import java.util.TreeMap;
6+
import java.util.TreeSet;
7+
import java.util.regex.Matcher;
8+
import java.util.regex.Pattern;
9+
import lombok.experimental.UtilityClass;
10+
11+
@UtilityClass
12+
public class EnvironmentVariablesUtils {
13+
14+
private static final Pattern ENVIRONMENT_VARIABLE_PATTERN = Pattern.compile("\\$\\{(.*?)\\}");
15+
16+
public static String replaceWithEnv(String command, Map<String, String> envVariables) {
17+
String substitutedStr = command;
18+
StringBuffer result = new StringBuffer();
19+
// First pass to replace environment variables
20+
Matcher matcher = ENVIRONMENT_VARIABLE_PATTERN.matcher(substitutedStr);
21+
while (matcher.find()) {
22+
String key = matcher.group(1);
23+
String envValue = envVariables.getOrDefault(key, "");
24+
matcher.appendReplacement(result, Matcher.quoteReplacement(envValue));
25+
}
26+
matcher.appendTail(result);
27+
28+
return result.toString();
29+
}
30+
31+
public void validateEnvironmentVariables(TreeMap<String, String> envVariables, String script) {
32+
Matcher matcher = ENVIRONMENT_VARIABLE_PATTERN.matcher(script);
33+
34+
Set<String> scriptEnvironmentVariables = new TreeSet<>();
35+
while (matcher.find()) {
36+
scriptEnvironmentVariables.add(matcher.group(1));
37+
}
38+
39+
scriptEnvironmentVariables.removeAll(envVariables.keySet());
40+
41+
if (!scriptEnvironmentVariables.isEmpty()) {
42+
throw new IllegalStateException(String.format("Could not find the following environment variables: %s",
43+
scriptEnvironmentVariables));
44+
}
45+
}
46+
47+
}

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

Lines changed: 21 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -15,29 +15,28 @@
1515
import org.slf4j.Logger;
1616
import org.slf4j.LoggerFactory;
1717

18-
/**
19-
* Class for executing SQL scripts programmatically.
20-
*/
18+
/** Class for executing SQL scripts programmatically. */
2119
class SqlExecutor {
2220

2321
private static final Logger log = LoggerFactory.getLogger(SqlExecutor.class);
2422

25-
private static final Pattern SET_STATEMENT_PATTERN = Pattern.compile(
26-
"SET\\s+'(\\S+)'\\s*=\\s*'(.+)';?", Pattern.CASE_INSENSITIVE);
23+
private static final Pattern SET_STATEMENT_PATTERN =
24+
Pattern.compile("SET\\s+'(\\S+)'\\s*=\\s*'(.+)';?", Pattern.CASE_INSENSITIVE);
2725

2826
private final TableEnvironment tableEnv;
27+
private final Map<String, String> envVariables;
2928

30-
public SqlExecutor(Configuration configuration, String udfPath) {
29+
public SqlExecutor(
30+
Configuration configuration, String udfPath, Map<String, String> envVariables) {
3131
StreamExecutionEnvironment sEnv;
3232
try {
3333
sEnv = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
3434
} catch (Exception e) {
3535
throw e;
3636
}
3737

38-
EnvironmentSettings tEnvConfig = EnvironmentSettings.newInstance()
39-
.withConfiguration(configuration)
40-
.build();
38+
EnvironmentSettings tEnvConfig =
39+
EnvironmentSettings.newInstance().withConfiguration(configuration).build();
4140

4241
this.tableEnv = StreamTableEnvironment.create(sEnv, tEnvConfig);
4342

@@ -47,6 +46,7 @@ public SqlExecutor(Configuration configuration, String udfPath) {
4746
if (udfPath != null) {
4847
setupUdfPath(udfPath);
4948
}
49+
this.envVariables = envVariables;
5050
}
5151

5252
/**
@@ -62,14 +62,15 @@ public TableResult executeScript(String script) throws Exception {
6262
for (String statement : statements) {
6363
tableResult = executeStatement(statement);
6464
}
65-
//
66-
// TableEnvironmentImpl tEnv1 = (TableEnvironmentImpl) tableEnv;
67-
//
68-
// StatementSetOperation parse = (StatementSetOperation)tEnv1.getParser()
69-
// .parse(statements.get(statements.size()-1)).get(0);
70-
//
71-
// CompiledPlan plan = tEnv1.compilePlan(parse.getOperations());
72-
// plan.writeToFile("/Users/henneberger/flink-jar-runner/src/test/resources/sql/compiled-plan-udf.json");
65+
//
66+
// TableEnvironmentImpl tEnv1 = (TableEnvironmentImpl) tableEnv;
67+
//
68+
// StatementSetOperation parse = (StatementSetOperation)tEnv1.getParser()
69+
// .parse(statements.get(statements.size()-1)).get(0);
70+
//
71+
// CompiledPlan plan = tEnv1.compilePlan(parse.getOperations());
72+
//
73+
// plan.writeToFile("/Users/henneberger/flink-jar-runner/src/test/resources/sql/compiled-plan-udf.json");
7374

7475
return tableResult;
7576
}
@@ -94,7 +95,8 @@ private TableResult executeStatement(String statement) {
9495
} else {
9596
System.out.println(statement);
9697
log.info("Executing statement:\n{}", statement);
97-
tableResult = tableEnv.executeSql(replaceWithEnv(statement));
98+
tableResult =
99+
tableEnv.executeSql(EnvironmentVariablesUtils.replaceWithEnv(statement, envVariables));
98100
}
99101
} catch (Exception e) {
100102
log.error("Failed to execute statement: {}", statement, e);
@@ -168,4 +170,4 @@ protected TableResult executeCompiledPlan(String planJson) throws Exception {
168170
throw e;
169171
}
170172
}
171-
}
173+
}

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

Lines changed: 43 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -17,48 +17,59 @@
1717

1818
package com.datasqrl;
1919

20+
import java.io.File;
21+
import java.util.Map;
22+
import java.util.TreeMap;
23+
import java.util.concurrent.Callable;
2024
import lombok.SneakyThrows;
2125
import lombok.extern.slf4j.Slf4j;
2226
import org.apache.flink.configuration.Configuration;
2327
import org.apache.flink.configuration.GlobalConfiguration;
2428
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
2529
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.module.SimpleModule;
26-
import org.apache.flink.table.api.CompiledPlan;
2730
import org.apache.flink.table.api.TableResult;
28-
import org.apache.flink.table.api.internal.TableEnvironmentImpl;
29-
import org.apache.flink.table.operations.StatementSetOperation;
3031
import org.apache.flink.util.FileUtils;
3132
import picocli.CommandLine;
32-
import picocli.CommandLine.*;
33-
import java.io.File;
34-
import java.util.*;
35-
import java.util.concurrent.Callable;
36-
37-
/**
38-
* Main class for executing SQL scripts using picocli.
39-
*/
40-
@Command(name = "SqlRunner", mixinStandardHelpOptions = true, version = "1.0", description = "Runs SQL scripts using Flink TableEnvironment.")
33+
import picocli.CommandLine.Command;
34+
import picocli.CommandLine.Option;
35+
36+
/** Main class for executing SQL scripts using picocli. */
37+
@Command(
38+
name = "SqlRunner",
39+
mixinStandardHelpOptions = true,
40+
version = "1.0",
41+
description = "Runs SQL scripts using Flink TableEnvironment.")
4142
@Slf4j
4243
public class SqlRunner implements Callable<Integer> {
4344

44-
@Option(names = {"-s", "--sqlfile"}, description = "SQL file to execute.")
45+
@Option(
46+
names = {"-s", "--sqlfile"},
47+
description = "SQL file to execute.")
4548
private File sqlFile;
4649

47-
@Option(names = {"--block"}, description = "Wait for the flink job manager to exit.",
48-
defaultValue = "false")
50+
@Option(
51+
names = {"--block"},
52+
description = "Wait for the flink job manager to exit.",
53+
defaultValue = "false")
4954
private boolean block;
5055

51-
@Option(names = {"--planfile"}, description = "Compiled plan JSON file.")
56+
@Option(
57+
names = {"--planfile"},
58+
description = "Compiled plan JSON file.")
5259
private File planFile;
5360

54-
@Option(names = {"--configfile"}, description = "Configuration YAML file.")
61+
@Option(
62+
names = {"--configfile"},
63+
description = "Configuration YAML file.")
5564
private File configFile;
5665

57-
@Option(names = {"--udfpath"}, description = "Path to UDFs.")
66+
@Option(
67+
names = {"--udfpath"},
68+
description = "Path to UDFs.")
5869
private String udfPath;
5970

6071
public static void main(String[] args) {
61-
int exitCode = new CommandLine(new SqlRunner()).execute(args);
72+
var exitCode = new CommandLine(new SqlRunner()).execute(args);
6273
System.exit(exitCode);
6374
}
6475

@@ -70,22 +81,27 @@ public Integer call() throws Exception {
7081
}
7182

7283
// Load configuration if configFile is provided
73-
Configuration configuration = new Configuration();
84+
var configuration = new Configuration();
7485
if (configFile != null) {
7586
configuration = loadConfigurationFromYaml(configFile);
7687
}
7788

89+
log.info("Environment variables");
90+
TreeMap<String, String> envVariables = new TreeMap<>(System.getenv());
91+
envVariables.forEach((name, value) -> log.info("{}: {}", name, value));
92+
7893
// Initialize SqlExecutor
79-
SqlExecutor sqlExecutor = new SqlExecutor(configuration, udfPath);
94+
var sqlExecutor = new SqlExecutor(configuration, udfPath, envVariables);
8095
TableResult tableResult;
8196
// Input validation and execution logic
8297
if (sqlFile != null) {
8398
// Single SQL file mode
84-
String script = FileUtils.readFileUtf8(sqlFile);
99+
var script = FileUtils.readFileUtf8(sqlFile);
100+
EnvironmentVariablesUtils.validateEnvironmentVariables(envVariables, script);
85101
tableResult = sqlExecutor.executeScript(script);
86102
} else if (planFile != null) {
87103
// Compiled plan JSON file
88-
String planJson = FileUtils.readFileUtf8(planFile);
104+
var planJson = FileUtils.readFileUtf8(planFile);
89105
planJson = replaceScriptWithEnv(planJson);
90106

91107
tableResult = sqlExecutor.executeCompiledPlan(planJson);
@@ -110,12 +126,11 @@ private String replaceScriptWithEnv(String script) {
110126
return objectMapper.writeValueAsString(map);
111127
}
112128

113-
114129
public static ObjectMapper getObjectMapper() {
115-
ObjectMapper objectMapper = new ObjectMapper();
130+
var objectMapper = new ObjectMapper();
116131

117132
// Register the custom deserializer module
118-
SimpleModule module = new SimpleModule();
133+
var module = new SimpleModule();
119134
module.addDeserializer(String.class, new JsonEnvVarDeserializer());
120135
objectMapper.registerModule(module);
121136
return objectMapper;
@@ -130,7 +145,7 @@ public static ObjectMapper getObjectMapper() {
130145
*/
131146
private Configuration loadConfigurationFromYaml(File configFile) throws Exception {
132147
log.info("Loading configuration from {}", configFile.getAbsolutePath());
133-
Configuration configuration = GlobalConfiguration.loadConfiguration(configFile.getAbsolutePath());
148+
var configuration = GlobalConfiguration.loadConfiguration(configFile.getAbsolutePath());
134149
return configuration;
135150
}
136-
}
151+
}

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

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2,10 +2,10 @@
22

33
import java.util.ArrayList;
44
import java.util.List;
5+
import lombok.experimental.UtilityClass;
56

6-
/**
7-
* Utility class for parsing SQL scripts.
8-
*/
7+
/** Utility class for parsing SQL scripts. */
8+
@UtilityClass
99
class SqlUtils {
1010

1111
private static final String STATEMENT_DELIMITER = ";"; // a statement should end with `;`
@@ -25,10 +25,13 @@ class SqlUtils {
2525
* @return A list of individual SQL statements.
2626
*/
2727
public static List<String> parseStatements(String script) {
28-
String formatted = formatSqlFile(script).replaceAll(BEGIN_CERTIFICATE,
29-
ESCAPED_BEGIN_CERTIFICATE).replaceAll(END_CERTIFICATE, ESCAPED_END_CERTIFICATE)
30-
.replaceAll(COMMENT_PATTERN, "").replaceAll(ESCAPED_BEGIN_CERTIFICATE, BEGIN_CERTIFICATE)
31-
.replaceAll(ESCAPED_END_CERTIFICATE, END_CERTIFICATE);
28+
String formatted =
29+
formatSqlFile(script)
30+
.replaceAll(BEGIN_CERTIFICATE, ESCAPED_BEGIN_CERTIFICATE)
31+
.replaceAll(END_CERTIFICATE, ESCAPED_END_CERTIFICATE)
32+
.replaceAll(COMMENT_PATTERN, "")
33+
.replaceAll(ESCAPED_BEGIN_CERTIFICATE, BEGIN_CERTIFICATE)
34+
.replaceAll(ESCAPED_END_CERTIFICATE, END_CERTIFICATE);
3235

3336
List<String> statements = new ArrayList<>();
3437

@@ -74,4 +77,4 @@ public static String formatSqlFile(String content) {
7477
formatted.append(LINE_DELIMITER);
7578
return formatted.toString();
7679
}
77-
}
80+
}

0 commit comments

Comments
 (0)