Add support for Flink Beam templates

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()) {

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sConnector.java

@@ -5,7 +5,9 @@
 import java.io.IOException;
 import java.io.StringReader;
 import java.sql.SQLException;
+import java.util.ArrayList;
 import java.util.LinkedHashMap;
+import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Objects;
@@ -52,16 +54,23 @@ public Map<String, String> configure() throws SQLException {
             .with("database", source.database())
             .with("table", source.table())
             .with(options);
-    String configs = tableTemplateApi.list()
+    List<String> templates = tableTemplateApi.list()
         .stream()
         .map(V1alpha1TableTemplate::getSpec)
         .filter(Objects::nonNull)
         .filter(x -> x.getDatabases() == null || x.getDatabases().contains(source.database()))
         .filter(x -> x.getMethods() == null || x.getMethods().contains(K8sUtils.method(source)))
         .map(V1alpha1TableTemplateSpec::getConnector)
         .filter(Objects::nonNull)
-        .map(x -> new Template.SimpleTemplate(x).render(env))
-        .collect(Collectors.joining("\n"));
+        .collect(Collectors.toList());
+    List<String> renderedTemplates = new ArrayList<>();
+    for (String template : templates) {
+      String renderedTemplate = new Template.SimpleTemplate(template).render(env);
+      if (renderedTemplate != null) {
+        renderedTemplates.add(renderedTemplate);
+      }
+    }
+    String configs = String.join("\n", renderedTemplates);
     Properties props = new Properties();
     try {
       // Preload configs in order to check for 'connector'

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sJobDeployer.java

@@ -1,10 +1,12 @@
 package com.linkedin.hoptimator.k8s;
 
+import com.linkedin.hoptimator.Source;
+import com.linkedin.hoptimator.ThrowingFunction;
 import java.sql.SQLException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.Properties;
-import java.util.function.Function;
 import java.util.stream.Collectors;
 
 import com.linkedin.hoptimator.Job;
@@ -36,25 +38,37 @@ class K8sJobDeployer extends K8sYamlDeployer {
   public List<String> specify() throws SQLException {
     Properties properties = ConfigService.config(context.connection(), false, FLINK_CONFIG);
     properties.putAll(job.sink().options());
-    Function<SqlDialect, String> sql = job.sql();
+    ThrowingFunction<SqlDialect, String> sql = job.sql();
+    ThrowingFunction<SqlDialect, String> fieldMap = job.fieldMap();
     String name = K8sUtils.canonicalizeName(job.sink().database(), job.name());
     Template.Environment env = new Template.SimpleEnvironment()
         .with("name", name)
         .with("database", job.sink().database())
         .with("schema", job.sink().schema())
         .with("table", job.sink().table())
-        .with("sql", sql.apply(SqlDialect.ANSI))
-        .with("flinksql", sql.apply(SqlDialect.FLINK))
+        .with("sourceDatabases", () -> job.sources().stream().map(Source::database).collect(Collectors.joining(",")))
+        .with("sourceSchemas", () -> job.sources().stream().map(Source::schema).collect(Collectors.joining(",")))
+        .with("sourceTables", () -> job.sources().stream().map(Source::table).collect(Collectors.joining(",")))
+        .with("sql", () -> sql.apply(SqlDialect.ANSI))
+        .with("flinksql", () -> sql.apply(SqlDialect.FLINK))
         .with("flinkconfigs", properties)
-        .with(job.sink().options());
-    return jobTemplateApi.list()
+        .with("fieldMap", () -> "'" + fieldMap.apply(SqlDialect.ANSI) + "'")
+        .with(properties);
+    List<String> templates = jobTemplateApi.list()
         .stream()
         .map(V1alpha1JobTemplate::getSpec)
         .filter(Objects::nonNull)
         .filter(x -> x.getDatabases() == null || x.getDatabases().contains(job.sink().database()))
         .map(V1alpha1JobTemplateSpec::getYaml)
         .filter(Objects::nonNull)
-        .map(x -> new Template.SimpleTemplate(x).render(env))
         .collect(Collectors.toList());
+    List<String> renderedTemplates = new ArrayList<>();
+    for (String template : templates) {
+      String renderedTemplate = new Template.SimpleTemplate(template).render(env);
+      if (renderedTemplate != null) {
+        renderedTemplates.add(renderedTemplate);
+      }
+    }
+    return renderedTemplates;
   }
 }

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sMaterializedViewDeployer.java

@@ -98,7 +98,7 @@ String name() {
     return K8sUtils.canonicalizeName(view.path());
   }
 
-  String sql() {
+  String sql() throws SQLException {
     return view.pipelineSql().apply(SqlDialect.ANSI);
   }
 

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sSourceDeployer.java

@@ -1,6 +1,7 @@
 package com.linkedin.hoptimator.k8s;
 
 import java.sql.SQLException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
@@ -39,16 +40,22 @@ public List<String> specify() throws SQLException {
             .with(source.options())
             .with(DeploymentService.parseHints(connection.connectionProperties()));
 
-    return tableTemplateApi.list()
+    List<String> templates =  tableTemplateApi.list()
         .stream()
         .map(V1alpha1TableTemplate::getSpec)
         .filter(Objects::nonNull)
         .filter(x -> x.getDatabases() == null || x.getDatabases().contains(source.database()))
         .filter(x -> x.getMethods() == null || x.getMethods().contains(K8sUtils.method(source)))
         .map(V1alpha1TableTemplateSpec::getYaml)
         .filter(Objects::nonNull)
-        .map(x -> new Template.SimpleTemplate(x).render(env))
-        .filter(Objects::nonNull) // Filter out null templates (which might have errored out due to missing hint expressions)
         .collect(Collectors.toList());
+    List<String> renderedTemplates = new ArrayList<>();
+    for (String template : templates) {
+      String renderedTemplate = new Template.SimpleTemplate(template).render(env);
+      if (renderedTemplate != null) {
+        renderedTemplates.add(renderedTemplate);
+      }
+    }
+    return renderedTemplates;
   }
 }

hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/models/V1alpha1Database.java

@@ -30,7 +30,7 @@
  * Database metadata.
  */
 @ApiModel(description = "Database metadata.")
-@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2025-05-21T02:31:00.123Z[Etc/UTC]")
+@javax.annotation.Generated(value = "org.openapitools.codegen.languages.JavaClientCodegen", date = "2025-09-04T16:19:29.143Z[Etc/UTC]")
 public class V1alpha1Database implements io.kubernetes.client.common.KubernetesObject {
   public static final String SERIALIZED_NAME_API_VERSION = "apiVersion";
   @SerializedName(SERIALIZED_NAME_API_VERSION)