Skip SQL job generation when a pipeline source or sink has no connector

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