Skip SQL job generation when a pipeline source or sink has no connector (#247)

A JobTemplate with no `databases` filter matches every sink, so a table
with no connector configuration (empty `WITH ()`) would still get a Flink
SqlJob rendered against it -- e.g. `CREATE TABLE ... WITH ()` plus an
`INSERT INTO` that Flink cannot execute. Such tables are moved by a
non-SQL JobTemplate rather than a SqlJob. This happens in both directions:
a connector-less sink (materialized by a downstream non-SQL job) and a
connector-less source (the reverse case).

Add MissingConnectorException (SQLNonTransientException) to signal that a
table has no connector and cannot participate in a generated SQL job.
PipelineRel throws it while resolving connector configs when the sink
(sql()) or any source (script()) has an empty config map. Callers that
generate SQL treat it as 'skip the SQL job', not an error:

  - K8sJobDeployer: a sqlOrNull helper catches it so {{sql}}/{{flinksql}}
    resolve to null, dropping SQL-based JobTemplates during rendering while
    non-SQL templates still render.
  - LogicalTableDeployer and K8sMaterializedViewDeployer: catch it and treat
    the pipeline as having no SQL.
  - K8sPipelineDeployer stores "" instead of null for the Pipeline CR sql.

Fixture: add ads-catalog-database to demodb-write-template's databases. The
offline demo tier (ADS_CATALOG) previously had only a read/trigger template,
so its sink had no write connector; the nearline->offline logical-table
graph now correctly renders a FlinkSessionJob for that tier.

Tests (per testing-best-practices.md): PipelineRelImplementorTest asserts
sql() throws when the sink or a source has no connector; K8sJobDeployerTest
asserts a connector-less sink skips the SQL template while a non-SQL template
still renders.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: jogrogan

deploy/samples/demodb.yaml

@@ -56,6 +56,7 @@ spec:
   databases:
   - profile-database
   - ads-database
+  - ads-catalog-database
   methods:
   - Modify
   connector: |

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

@@ -0,0 +1,20 @@
+package com.linkedin.hoptimator;
+
+import java.sql.SQLNonTransientException;
+
+
+/**
+ * Signals that a table has no connector configuration, and therefore cannot participate
+ * in a generated SQL job. Callers that generate SQL-based jobs are expected to catch this
+ * and skip SQL generation, while still emitting any non-SQL jobs.
+ *
+ * <p>This is not necessarily an error: some tables are moved by means other than a SQL job.
+ * For example, a JobTemplate may render a non-SQL job (rather than {@code SqlJob}) to move data
+ * into or out of such a table.
+ */
+public class MissingConnectorException extends SQLNonTransientException {
+
+  public MissingConnectorException(String path) {
+    super("No connector configured for '" + path + "'.");
+  }
+}

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

@@ -1,6 +1,7 @@
 package com.linkedin.hoptimator.k8s;
 
 import com.linkedin.hoptimator.Job;
+import com.linkedin.hoptimator.MissingConnectorException;
 import com.linkedin.hoptimator.Source;
 import com.linkedin.hoptimator.SqlDialect;
 import com.linkedin.hoptimator.ThrowingFunction;
@@ -45,6 +46,7 @@ public List<String> specify() throws SQLException {
     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())
@@ -55,8 +57,8 @@ public List<String> specify() throws SQLException {
         .with("sourceCatalogs", () -> job.sources().stream().map(Source::catalog).filter(Objects::nonNull).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("sql", () -> sqlOrNull(sql, SqlDialect.ANSI))
+        .with("flinksql", () -> sqlOrNull(sql, SqlDialect.FLINK))
         .with("flinkconfigs", properties)
         .with("fieldMap", () -> "'" + fieldMap.apply(SqlDialect.ANSI) + "'")
         .with(properties);
@@ -77,4 +79,18 @@ public List<String> specify() throws SQLException {
     }
     return renderedTemplates;
   }
+
+  /**
+   * Renders the pipeline SQL for the given dialect, returning {@code null} if a source or the
+   * sink has no connector. Such a pipeline cannot be materialized by a SQL job; returning
+   * {@code null} causes SQL-based JobTemplates to be skipped, while non-SQL JobTemplates still
+   * render.
+   */
+  private static String sqlOrNull(ThrowingFunction<SqlDialect, String> sql, SqlDialect dialect) throws SQLException {
+    try {
+      return sql.apply(dialect);
+    } catch (MissingConnectorException e) {
+      return null;
+    }
+  }
 }

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

@@ -2,6 +2,7 @@
 
 import com.linkedin.hoptimator.Deployer;
 import com.linkedin.hoptimator.MaterializedView;
+import com.linkedin.hoptimator.MissingConnectorException;
 import com.linkedin.hoptimator.Sink;
 import com.linkedin.hoptimator.Source;
 import com.linkedin.hoptimator.SqlDialect;
@@ -111,7 +112,13 @@ String name() {
   }
 
   String sql() throws SQLException {
-    return view.pipelineSql().apply(SqlDialect.ANSI);
+    try {
+      return view.pipelineSql().apply(SqlDialect.ANSI);
+    } catch (MissingConnectorException e) {
+      // A source or sink has no connector (the pipeline is moved by a non-SQL job rather than
+      // Flink SQL), so there is no pipeline SQL to stamp on the Pipeline resource.
+      return null;
+    }
   }
 
   @Override

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

@@ -38,7 +38,7 @@ class K8sPipelineDeployer extends K8sDeployer<V1alpha1Pipeline, V1alpha1Pipeline
     super(context, K8sApiEndpoints.PIPELINES);
     this.name = name;
     this.yaml = String.join("\n---\n", specs);
-    this.sql = sql;
+    this.sql = sql == null ? "" : sql;
     this.sources = sources == null ? Collections.emptyList() : sources;
     this.sink = sink;
   }

hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/K8sJobDeployerTest.java

@@ -3,6 +3,7 @@
 import com.linkedin.hoptimator.DeploymentContext;
 
 import com.linkedin.hoptimator.Job;
+import com.linkedin.hoptimator.MissingConnectorException;
 import com.linkedin.hoptimator.Sink;
 import com.linkedin.hoptimator.Source;
 import com.linkedin.hoptimator.SqlDialect;
@@ -75,8 +76,12 @@ K8sYamlApi createYamlApi(K8sContext context) {
   }
 
   private Job createTestJob(Sink sink) {
+    return createTestJob(sink, dialect -> "INSERT INTO sink SELECT * FROM source");
+  }
+
+  private Job createTestJob(Sink sink, ThrowingFunction<SqlDialect, String> sql) {
     Map<String, ThrowingFunction<SqlDialect, String>> lazyEvals = new HashMap<>();
-    lazyEvals.put("sql", dialect -> "INSERT INTO sink SELECT * FROM source");
+    lazyEvals.put("sql", sql);
     lazyEvals.put("fieldMap", dialect -> "{\"a\":\"b\"}");
     Source source = new Source("srcdb", Arrays.asList("schema", "src_table"), Collections.emptyMap());
     return new Job("test-job", new HashSet<>(Collections.singleton(source)), sink, lazyEvals);
@@ -273,4 +278,57 @@ void specifyConditionalRenderedTemplateNotNull() throws SQLException {
     // The name should be canonicalized from "sinkdb" + "test-job"
     assertTrue(specs.get(0).contains("sinkdb"), "rendered template must contain database name");
   }
+
+  @Test
+  void specifyWithoutSinkConnectorSkipsSqlTemplate() throws SQLException {
+    // Arrange: the sink has no connector, so the pipeline SQL function throws.
+    ThrowingFunction<SqlDialect, String> throwingSql = dialect -> {
+      throw new MissingConnectorException("sinkdb.schema.sink_table");
+    };
+    // A SQL-based JobTemplate (references {{flinksql}})...
+    templates.add(new V1alpha1JobTemplate()
+        .metadata(new V1ObjectMeta().name("flink-template"))
+        .spec(new V1alpha1JobTemplateSpec()
+            .yaml("kind: SqlJob\nname: {{name}}\nsql:\n  - {{flinksql}}")));
+    // ...and a non-SQL JobTemplate (references no SQL).
+    templates.add(new V1alpha1JobTemplate()
+        .metadata(new V1ObjectMeta().name("nonsql-template"))
+        .spec(new V1alpha1JobTemplateSpec()
+            .yaml("kind: BatchJob\nname: {{name}}-job\nsinkTable: {{table}}")));
+
+    Sink sink = new Sink("sinkdb", Arrays.asList("schema", "sink_table"),
+        Collections.emptyMap());
+    Job job = createTestJob(sink, throwingSql);
+    K8sJobDeployer deployer = makeDeployer(job);
+
+    // Act
+    List<String> specs = deployer.specify();
+
+    // Assert: the SQL template is skipped; only the non-SQL template renders.
+    assertEquals(1, specs.size());
+    assertTrue(specs.get(0).contains("BatchJob"), "only the non-SQL template should render");
+    assertFalse(specs.get(0).contains("SqlJob"), "SQL-based template must be skipped when sink has no connector");
+  }
+
+  @Test
+  void specifyWithSinkConnectorRendersSqlTemplate() throws SQLException {
+    // Arrange: the sink has a connector, so the pipeline SQL function returns SQL.
+    templates.add(new V1alpha1JobTemplate()
+        .metadata(new V1ObjectMeta().name("flink-template"))
+        .spec(new V1alpha1JobTemplateSpec()
+            .yaml("kind: SqlJob\nname: {{name}}\nsql:\n  - {{flinksql}}")));
+
+    Sink sink = new Sink("sinkdb", Arrays.asList("schema", "sink_table"),
+        Collections.emptyMap());
+    Job job = createTestJob(sink);
+    K8sJobDeployer deployer = makeDeployer(job);
+
+    // Act
+    List<String> specs = deployer.specify();
+
+    // Assert
+    assertEquals(1, specs.size());
+    assertTrue(specs.get(0).contains("SqlJob"), "SQL-based template must render when sink has a connector");
+    assertTrue(specs.get(0).contains("INSERT INTO sink SELECT * FROM source"));
+  }
 }

hoptimator-logical/src/main/java/com/linkedin/hoptimator/logical/LogicalTableDeployer.java

@@ -11,6 +11,7 @@
 import java.util.Map;
 import java.util.Properties;
 
+import com.linkedin.hoptimator.MissingConnectorException;
 import com.linkedin.hoptimator.SqlDialect;
 import com.linkedin.hoptimator.k8s.models.V1alpha1DatabaseSpec;
 import com.linkedin.hoptimator.k8s.models.V1alpha1JobTemplate;
@@ -520,7 +521,15 @@ void deployPipelineBundle(String fromTier, String toTier, Map<String, Source> ti
       throw new SQLNonTransientException(message, e);
     }
 
-    String pipelineSql = pipeline.job().sql().apply(SqlDialect.ANSI);
+    String pipelineSql;
+    try {
+      pipelineSql = pipeline.job().sql().apply(SqlDialect.ANSI);
+    } catch (MissingConnectorException e) {
+      // A tier has no connector so there is no pipeline SQL. The non-SQL job specs are
+      // still emitted below via DeploymentService.specify(pipeline.job(), ...).
+      log.info("No connector for pipeline {}; skipping pipeline SQL.", pipelineName);
+      pipelineSql = null;
+    }
     List<String> pipelineSpecs = new ArrayList<>();
     for (Source src : pipeline.sources()) {
       pipelineSpecs.addAll(DeploymentService.specify(src, context.deploymentContext()));

hoptimator-util/src/main/java/com/linkedin/hoptimator/util/planner/PipelineRel.java

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.linkedin.hoptimator.DeploymentContext;
 import com.linkedin.hoptimator.Job;
+import com.linkedin.hoptimator.MissingConnectorException;
 import com.linkedin.hoptimator.Pipeline;
 import com.linkedin.hoptimator.Sink;
 import com.linkedin.hoptimator.Source;
@@ -126,6 +127,12 @@ private ScriptImplementor script(DeploymentContext context) throws SQLException
         script = script.catalog(source.getKey().catalog());
         script = script.database(source.getKey().catalog(), source.getKey().schema());
         Map<String, String> configs = ConnectionService.configure(source.getKey(), context);
+        // A source with no connector configuration cannot be read by a SQL job. As with a
+        // connector-less sink, such a table is moved by a non-SQL job (with source and sink
+        // reversed). Signal this to callers so they can skip SQL generation.
+        if (configs.isEmpty()) {
+          throw new MissingConnectorException(source.getKey().pathString());
+        }
         String suffix = needsSuffixes ? "_source" : null;
         script = script.connector(source.getKey().catalog(), source.getKey().schema(), source.getKey().table(), suffix, source.getValue(), configs);
       }
@@ -161,6 +168,11 @@ public ThrowingFunction<SqlDialect, String> sql(DeploymentContext context) throw
           validateFieldMapping(targetRowType);
         }
         Map<String, String> sinkConfigs = ConnectionService.configure(sink, context);
+        // A sink with no connector configuration cannot be materialized by a SQL job. Signal
+        // this to callers so they can skip SQL generation while still emitting non-SQL jobs.
+        if (sinkConfigs.isEmpty()) {
+          throw new MissingConnectorException(sink.pathString());
+        }
         script = script.catalog(sink.catalog());
         script = script.database(sink.catalog(), sink.schema());
         // Check if we need to add suffixes to avoid table name collisions