Add support for Flink Beam templates (#163)

* Add new FlinkBeam type and regenerate k8s models

* Add new template option to avoid matching if a specific field is present

* Add field map logic for trivial expressions + lazy evaluation of template functions

* Add template equality checks

* Add beam template, some testing, and some fixes

* Add fieldMap support for nested field references and add more thorough testing

* Hide magic strings

Author: jogrogan

Makefile

@@ -56,6 +56,7 @@ deploy-flink: deploy
 	kubectl apply -f ./deploy/dev/flink-session-cluster.yaml
 	kubectl apply -f ./deploy/dev/flink-sql-gateway.yaml
 	kubectl apply -f ./deploy/samples/flink-template.yaml
+	kubectl apply -f ./deploy/samples/flink-beam-template.yaml
 
 undeploy-flink:
 	kubectl delete flinksessionjobs.flink.apache.org --all || echo "skipping"

deploy/config/hoptimator-configmap.yaml

@@ -13,7 +13,12 @@ data:
   game.properties: |
     enemy.types=aliens,monsters
     player.maximum-lives=5
+
   user-interface.properties: |
     color.good=purple
     color.bad=yellow
-    allow.textmode=true
+    allow.textmode=true
+
+  flink.config: |
+    flink.app.name=hoptimator-flink-runner
+    flink.app.type=SQL

deploy/samples/flink-beam-template.yaml

@@ -0,0 +1,26 @@
+## This template adds Flink Beam support.
+
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: JobTemplate
+metadata:
+  name: flink-beam-sqljob-template
+spec:
+  yaml: |
+    apiVersion: hoptimator.linkedin.com/v1alpha1
+    kind: SqlJob
+    metadata:
+      name: {{name}}
+    spec:
+      dialect: FlinkBeam
+      executionMode: Streaming
+      sql:
+        - PLACEHOLDER
+      configs:
+        {{flink.app.type==BEAM}}
+        source.group.id: hoptimator-flink-beam-{{pipeline}}
+        source.schemas: {{sourceSchemas}}
+        source.tables: {{sourceTables}}
+        sink.schema: {{schema}}
+        sink.table: {{table}}
+        fields: {{fieldMap}}
+        {{flinkconfigs}}

deploy/samples/flink-template.yaml

@@ -16,7 +16,8 @@ spec:
         entryClass: com.linkedin.hoptimator.flink.runner.FlinkRunner
         args:
         - {{flinksql}}
-        jarURI: file:///opt/hoptimator-flink-runner.jar
+        jarURI: file:///opt/{{flink.app.name}}.jar
         parallelism: {{flink.parallelism:1}}
         upgradeMode: stateless
         state: running
+        {{flink.app.type==SQL}}

hoptimator-api/src/main/java/com/linkedin/hoptimator/Job.java

@@ -1,30 +1,90 @@
 package com.linkedin.hoptimator;
 
-import java.util.function.Function;
+import java.util.Map;
+import java.util.Set;
 
 
+/**
+ * Represents a data pipeline job with lazy-evaluated template functions.
+ *
+ * <p>A Job encapsulates the configuration and execution logic for a data pipeline,
+ * including the destination sink and a collection of lazy-evaluated functions that
+ * generate SQL scripts, field mappings, and other template values on demand.
+ */
 public class Job implements Deployable {
 
   private final String name;
+  private final Set<Source> sources;
   private final Sink sink;
-  private final Function<SqlDialect, String> sql;
 
-  public Job(String name, Sink sink, Function<SqlDialect, String> sql) {
+  /**
+   * Lazy-evaluated template functions that generate various outputs for the job.
+   *
+   * <p>This map contains functions that are evaluated on-demand when specific
+   * template values are needed. Each function takes a {@link SqlDialect} parameter
+   * and returns a string representation of the requested output.
+   *
+   * @see ThrowingFunction
+   * @see SqlDialect
+   */
+  private final Map<String, ThrowingFunction<SqlDialect, String>> lazyEvals;
+
+  public Job(String name, Set<Source> sources, Sink sink, Map<String, ThrowingFunction<SqlDialect, String>> lazyEvals) {
     this.name = name;
+    this.sources = sources;
     this.sink = sink;
-    this.sql = sql;
+    this.lazyEvals = lazyEvals;
   }
 
   public String name() {
     return name;
   }
 
+  public Set<Source> sources() {
+    return sources;
+  }
+
   public Sink sink() {
     return sink;
   }
 
-  public Function<SqlDialect, String> sql() {
-    return sql;
+  public ThrowingFunction<SqlDialect, String> sql() {
+    return eval("sql");
+  }
+
+  public ThrowingFunction<SqlDialect, String> query() {
+    return eval("query");
+  }
+
+  public ThrowingFunction<SqlDialect, String> fieldMap() {
+    return eval("fieldMap");
+  }
+
+  /**
+   * Retrieves a lazy-evaluated template function by key.
+   *
+   * <p>This method provides access to the template functions stored in {@link #lazyEvals}.
+   * The returned function can be called with a {@link SqlDialect} to generate the
+   * corresponding output string.
+   *
+   * <p><strong>Available Keys:</strong>
+   * <ul>
+   *   <li><code>"sql"</code> - Complete SQL script with INSERT INTO statements</li>
+   *   <li><code>"query"</code> - SELECT query portion only</li>
+   *   <li><code>"fieldMap"</code> - JSON mapping of source to destination fields</li>
+   * </ul>
+   *
+   * @param key The template function key to retrieve
+   * @return The lazy-evaluated function, or {@code null} if the key doesn't exist
+   *
+   * @see #lazyEvals
+   * @see ThrowingFunction#apply(Object)
+   */
+  private ThrowingFunction<SqlDialect, String> eval(String key) {
+    if (!lazyEvals.containsKey(key)) {
+      throw new IllegalArgumentException("Unknown eval key: " + key + ". Available keys: " + lazyEvals.keySet());
+    }
+    return lazyEvals.get(key);
   }
 
   @Override

hoptimator-api/src/main/java/com/linkedin/hoptimator/MaterializedView.java

@@ -1,16 +1,15 @@
 package com.linkedin.hoptimator;
 
 import java.util.List;
-import java.util.function.Function;
 
 
 public class MaterializedView extends View implements Deployable {
 
   private final String database;
-  private final Function<SqlDialect, String> pipelineSql;
+  private final ThrowingFunction<SqlDialect, String> pipelineSql;
   private final Pipeline pipeline;
 
-  public MaterializedView(String database, List<String> path, String viewSql, Function<SqlDialect, String> pipelineSql,
+  public MaterializedView(String database, List<String> path, String viewSql, ThrowingFunction<SqlDialect, String> pipelineSql,
       Pipeline pipeline) {
     super(path, viewSql);
     this.database = database;
@@ -22,7 +21,7 @@ public Pipeline pipeline() {
     return pipeline;
   }
 
-  public Function<SqlDialect, String> pipelineSql() {
+  public ThrowingFunction<SqlDialect, String> pipelineSql() {
     return pipelineSql;
   }
 

hoptimator-api/src/main/java/com/linkedin/hoptimator/ThrowingFunction.java

@@ -0,0 +1,12 @@
+package com.linkedin.hoptimator;
+
+import java.sql.SQLException;
+
+
+/**
+ * Functional interface that allows functions to throw SQLException.
+ */
+@FunctionalInterface
+public interface ThrowingFunction<T, R> {
+  R apply(T t) throws SQLException;
+}

hoptimator-api/src/main/java/com/linkedin/hoptimator/ThrowingSupplier.java

@@ -0,0 +1,12 @@
+package com.linkedin.hoptimator;
+
+import java.sql.SQLException;
+
+
+/**
+ * Functional interface that allows suppliers to throw SQLException.
+ */
+@FunctionalInterface
+public interface ThrowingSupplier<T> {
+  T get() throws SQLException;
+}

hoptimator-cli/src/main/java/sqlline/HoptimatorAppConfig.java

@@ -127,6 +127,8 @@ public void execute(String line, DispatchCallback dispatchCallback) {
         RelOptTable table = root.rel.getTable();
         if (table != null) {
           connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, String.join(".", table.getQualifiedName()));
+        } else if (create != null) {
+          connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, create.name.toString());
         }
         PipelineRel.Implementor plan = DeploymentService.plan(root, conn.materializations(), connectionProperties);
         if (create != null) {
@@ -301,6 +303,8 @@ public void execute(String line, DispatchCallback dispatchCallback) {
         RelOptTable table = root.rel.getTable();
         if (table != null) {
           connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, String.join(".", table.getQualifiedName()));
+        } else if (create != null) {
+          connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, create.name.toString());
         }
         PipelineRel.Implementor plan = DeploymentService.plan(root, conn.materializations(), connectionProperties);
         if (create != null) {

hoptimator-jdbc/src/testFixtures/java/com/linkedin/hoptimator/jdbc/QuidemTestBase.java

@@ -17,6 +17,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.Properties;
 import java.util.stream.Collectors;
 
 import org.apache.calcite.rel.RelRoot;
@@ -90,7 +91,10 @@ public void execute(Context context, boolean execute) throws Exception {
                 RelRoot root = HoptimatorDriver.convert(conn, sql).root;
                 String[] parts = line.split(" ", 2);
                 String pipelineName = parts.length == 2 ? parts[1] : "test";
-                Pipeline pipeline = DeploymentService.plan(root, Collections.emptyList(), conn.connectionProperties())
+                Properties properties = new Properties();
+                properties.putAll(conn.connectionProperties());
+                properties.put(DeploymentService.PIPELINE_OPTION, pipelineName);
+                Pipeline pipeline = DeploymentService.plan(root, Collections.emptyList(), properties)
                     .pipeline(pipelineName, conn);
                 List<String> specs = new ArrayList<>();
                 for (Source source : pipeline.sources()) {