Use TableTriggers As TableTemplates With Hint-able Expressions (#149)

srnand · shrinandthakkar · web-flow · commit 50e4f13823b9 · 2025-07-29T18:35:35.000-07:00
* Use TableTriggers As TableTemplates With Hintable Expressions

* Addressing Comments

* Revert testing version change

* Rebase Master

* Revert testing version change

* Do not use try with resources, otherwise, the resource is closed at the block end

* Update integration test to include the new trigger linked via hints

* Fix Integration testing cleanup and sorting

---------

Co-authored-by: Shrinand Thakkar &lt;sthakkar@linkedin.com&gt;
diff --git a/deploy/samples/demodb.yaml b/deploy/samples/demodb.yaml
@@ -48,3 +48,37 @@ spec:
   - Modify
   connector: |
     connector = blackhole
+
+---
+
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: TableTemplate
+metadata:
+  name: ads-source-trigger-template
+spec:
+  databases:
+    - ads-database
+  methods:
+    - Scan
+  yaml: |
+    apiVersion: hoptimator.linkedin.com/v1alpha1
+    kind: TableTrigger
+    metadata:
+      name: {{name}}-trigger
+    spec:
+      schema: KAFKA
+      table: {{offline.table.name}}
+      yaml: |
+        apiVersion: batch/v1
+        kind: Job
+        metadata:
+          name: {{name}}-job
+        spec:
+          template:
+            spec:
+              containers:
+              - name: hello
+                image: alpine/k8s:1.33.0
+                command: ["bash",  "-c", "echo {{name}}-trigger fired at `date`"]
+              restartPolicy: Never
+          backoffLimit: 4
diff --git a/hoptimator b/hoptimator
@@ -8,5 +8,5 @@ $BASEDIR/hoptimator-cli/build/install/hoptimator-cli/bin/hoptimator-cli \
   sqlline.SqlLine \
   -ac sqlline.HoptimatorAppConfig \
   --verbose \
-  -u "jdbc:hoptimator://fun=mysql" -n "" -p "" -nn "Hoptimator" $@
+  -u "jdbc:hoptimator://fun=mysql;hints=offline.table.name=ads_offline" -n "" -p "" -nn "Hoptimator" $@
 
diff --git a/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sSourceDeployer.java b/hoptimator-k8s/src/main/java/com/linkedin/hoptimator/k8s/K8sSourceDeployer.java
@@ -6,34 +6,39 @@
 import java.util.stream.Collectors;
 
 import com.linkedin.hoptimator.Source;
+import com.linkedin.hoptimator.jdbc.HoptimatorConnection;
 import com.linkedin.hoptimator.k8s.models.V1alpha1TableTemplate;
 import com.linkedin.hoptimator.k8s.models.V1alpha1TableTemplateList;
 import com.linkedin.hoptimator.k8s.models.V1alpha1TableTemplateSpec;
+import com.linkedin.hoptimator.util.DeploymentService;
 import com.linkedin.hoptimator.util.Template;
 
-
 /** Specifies an abstract Source with concrete YAML by applying TableTemplates. */
 class K8sSourceDeployer extends K8sYamlDeployer {
-
+  private final K8sContext context;
   private final Source source;
   private final K8sApi<V1alpha1TableTemplate, V1alpha1TableTemplateList> tableTemplateApi;
 
   K8sSourceDeployer(Source source, K8sContext context) {
     super(context);
+    this.context = context;
     this.source = source;
     this.tableTemplateApi = new K8sApi<>(context, K8sApiEndpoints.TABLE_TEMPLATES);
   }
 
   @Override
   public List<String> specify() throws SQLException {
     String name = K8sUtils.canonicalizeName(source.database(), source.table());
+    HoptimatorConnection connection = context.connection();
     Template.Environment env =
         new Template.SimpleEnvironment()
             .with("name", name)
             .with("database", source.database())
             .with("schema", source.schema())
             .with("table", source.table())
-            .with(source.options());
+            .with(source.options())
+            .with(DeploymentService.parseHints(connection.connectionProperties()));
+
     return tableTemplateApi.list()
         .stream()
         .map(V1alpha1TableTemplate::getSpec)
@@ -43,6 +48,7 @@ public List<String> specify() throws SQLException {
         .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());
   }
 }
diff --git a/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/TestSqlScripts.java b/hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/TestSqlScripts.java
@@ -29,6 +29,6 @@ public void k8sValidationScript() throws Exception {
   @Test
   @Tag("integration")
   public void k8sMetadataTables() throws Exception {
-    run("k8s-metadata.id");
+    run("k8s-metadata.id", "hints=offline.table.name=ads_offline");
   }
 }
diff --git a/hoptimator-k8s/src/test/resources/k8s-metadata.id b/hoptimator-k8s/src/test/resources/k8s-metadata.id
@@ -61,49 +61,66 @@ select name, failed from "k8s".pipelines order by name;
 !ok
 
 select name, failed from "k8s".pipeline_elements order by name;
-+----------------------------------------+--------+
-| NAME                                   | FAILED |
-+----------------------------------------+--------+
-| FlinkSessionJob/ads-database-audience2 | false  |
-| FlinkSessionJob/ads-database-pages     | false  |
-+----------------------------------------+--------+
-(2 rows)
++---------------------------------------------+--------+
+| NAME                                        | FAILED |
++---------------------------------------------+--------+
+| FlinkSessionJob/ads-database-audience2      | false  |
+| FlinkSessionJob/ads-database-pages          | false  |
+| TableTrigger/ads-database-pageviews-trigger | false  |
++---------------------------------------------+--------+
+(3 rows)
 
 !ok
 
-select * from "k8s".pipeline_element_map order by element_name;
-+----------------------------------------+---------------+
-| ELEMENT_NAME                           | PIPELINE_NAME |
-+----------------------------------------+---------------+
-| FlinkSessionJob/ads-database-audience2 | ads-audience2 |
-| FlinkSessionJob/ads-database-pages     | ads-pages     |
-+----------------------------------------+---------------+
-(2 rows)
+select * from "k8s".pipeline_element_map order by element_name, pipeline_name;
++---------------------------------------------+---------------+
+| ELEMENT_NAME                                | PIPELINE_NAME |
++---------------------------------------------+---------------+
+| FlinkSessionJob/ads-database-audience2      | ads-audience2 |
+| FlinkSessionJob/ads-database-pages          | ads-pages     |
+| TableTrigger/ads-database-pageviews-trigger | ads-audience2 |
+| TableTrigger/ads-database-pageviews-trigger | ads-pages     |
++---------------------------------------------+---------------+
+(4 rows)
 
 !ok
 
-select pl.name as pipeline_name, pe.element_name, pe.failed as element_failed from "k8s".pipelines pl inner join (select t2.element_name, t1.failed, t2.pipeline_name from "k8s".pipeline_elements t1 inner join "k8s".pipeline_element_map t2 on t1.name = t2.element_name) pe on pl.name = pe.pipeline_name order by pipeline_name;
-+---------------+----------------------------------------+----------------+
-| PIPELINE_NAME | ELEMENT_NAME                           | ELEMENT_FAILED |
-+---------------+----------------------------------------+----------------+
-| ads-audience2 | FlinkSessionJob/ads-database-audience2 | false          |
-| ads-pages     | FlinkSessionJob/ads-database-pages     | false          |
-+---------------+----------------------------------------+----------------+
-(2 rows)
+select pl.name as pipeline_name, pe.element_name, pe.failed as element_failed from "k8s".pipelines pl inner join (select t2.element_name, t1.failed, t2.pipeline_name from "k8s".pipeline_elements t1 inner join "k8s".pipeline_element_map t2 on t1.
+name = t2.element_name) pe on pl.name = pe.pipeline_name order by pipeline_name, element_name;
++---------------+---------------------------------------------+----------------+
+| PIPELINE_NAME | ELEMENT_NAME                                | ELEMENT_FAILED |
++---------------+---------------------------------------------+----------------+
+| ads-audience2 | FlinkSessionJob/ads-database-audience2      | false          |
+| ads-audience2 | TableTrigger/ads-database-pageviews-trigger | false          |
+| ads-pages     | FlinkSessionJob/ads-database-pages          | false          |
+| ads-pages     | TableTrigger/ads-database-pageviews-trigger | false          |
++---------------+---------------------------------------------+----------------+
+(4 rows)
 
 !ok
 
 select name, "SCHEMA", "TABLE" from "k8s".TABLE_TRIGGERS;
-+--------------------+--------+------------------+
-| NAME               | SCHEMA | TABLE            |
-+--------------------+--------+------------------+
-| test-table-trigger | KAFKA  | existing-topic-1 |
-+--------------------+--------+------------------+
-(1 row)
++--------------------------------+--------+------------------+
+| NAME                           | SCHEMA | TABLE            |
++--------------------------------+--------+------------------+
+| test-table-trigger             | KAFKA  | existing-topic-1 |
+| ads-database-pageviews-trigger | KAFKA  | ads_offline      |
++--------------------------------+--------+------------------+
+(2 rows)
 
 !ok
 
 drop materialized view ads.pages;
 (0 rows modified)
 
 !update
+
+drop materialized view ads.audience;
+(0 rows modified)
+
+!update
+
+drop materialized view ads.audience2;
+(0 rows modified)
+
+!update
diff --git a/hoptimator-util/src/main/java/com/linkedin/hoptimator/util/Template.java b/hoptimator-util/src/main/java/com/linkedin/hoptimator/util/Template.java
@@ -1,5 +1,8 @@
 package com.linkedin.hoptimator.util;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.LinkedHashMap;
 import java.util.Locale;
 import java.util.Map;
@@ -11,7 +14,7 @@
 
 /** A convenient way to generate K8s YAML. */
 public interface Template {
-
+  Logger log = LoggerFactory.getLogger(Template.class);
   String render(Environment env);
 
   /** Exposes environment variables to templates */
@@ -201,9 +204,16 @@ public String render(Environment env) {
         String key = m.group(2);
         String defaultValue = m.group(4);
         String transform = m.group(5);
-        String value = env.getOrDefault(key, () -> defaultValue);
-        if (value == null) {
-          throw new IllegalArgumentException(template + " has no value for variable " + key + ".");
+        String value;
+        try {
+          value = env.getOrDefault(key, () -> defaultValue);
+          if (value == null) {
+            log.warn("Template variable '{}' resolved to null. Skipping template.", key);
+            return null;
+          }
+        } catch (IllegalArgumentException e) {
+          log.warn("Missing template variable '{}' in environment: {}. Skipping template.", key, e.getMessage());
+          return null;
         }
         String transformedValue = applyTransform(value, transform);
         String quotedPrefix = Matcher.quoteReplacement(prefix);

Original file line number	Diff line number	Diff line change
`@@ -29,6 +29,6 @@ public void k8sValidationScript() throws Exception {`
`29`	`29`	`@Test`
`30`	`30`	`@Tag("integration")`
`31`	`31`	`public void k8sMetadataTables() throws Exception {`
`32`		`- run("k8s-metadata.id");`
	`32`	`+ run("k8s-metadata.id", "hints=offline.table.name=ads_offline");`
`33`	`33`	`}`
`34`	`34`	`}`