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

[jogrogan]

What

Teach the pipeline planner to skip Flink SQL job generation when a table involved in the pipeline has no connector configuration (i.e. an empty WITH () clause). Such tables are moved by a non-SQL JobTemplate (e.g. a triggered/ETL-style job) rather than a Flink SqlJob, so emitting a SqlJob for them produces SQL that is unrunnable and fails at Flink runtime.

Why

A JobTemplate with no databases filter matches every sink, so a connector-less table would still get a Flink SqlJob rendered against it — e.g. CREATE TABLE … WITH () followed by an INSERT INTO that Flink cannot execute. This affects both directions:

• Connector-less sink (e.g. a table materialized by a downstream non-SQL job): the generated SqlJob writes into an empty-connector sink.
• Connector-less source (the reverse case): the generated SqlJob reads from an empty-connector source.

Both should render only their non-SQL JobTemplate, never a broken SqlJob.

How

• Add MissingConnectorException (hoptimator-api), a SQLNonTransientException signaling that a table has no connector and therefore 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, which drops SQL-based JobTemplates during rendering while non-SQL templates still render.
  • LogicalTableDeployer and K8sMaterializedViewDeployer — catch it and treat the pipeline as having no SQL (null → nullable Pipeline CR field).
• K8sPipelineDeployer stores "" instead of null for the Pipeline CR sql field.

Tests

• PipelineRelImplementorTest — new cases asserting sql() throws MissingConnectorException when the sink or a source has no connector; existing SQL-generation tests stub a non-empty connector.
• K8sJobDeployerTest — new cases asserting a connector-less sink skips the SQL template while a non-SQL template still renders, and that a sink with a connector still renders the SQL template.

Fixtures

• deploy/samples/demodb.yaml: 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 missing write connector is the fixture bug this surfaced. The nearline→offline logical-table integration graph now correctly renders a FlinkSessionJob for that tier.

Compatibility

No public API changes. sql()'s contract now allows throwing MissingConnectorException; all in-tree callers handle it.

[github-actions]

Code Coverage

Overall Project	84.74% -0.04%	🟢
Files changed	75.31%	🟢

File	Coverage
MissingConnectorException.java	100%	🟢
K8sJobDeployer.java	100%	🟢
PipelineRel.java	96.02%	🟢
K8sMaterializedViewDeployer.java	94.9% -1.18%	🟢
K8sPipelineDeployer.java	94.37% -2.82%	🟢
LogicalTableDeployer.java	74.04% -1.08%	❌

[ryannedolan]

Wow great idea. Nice writeup as well.