Unify DDL execution and dry-run into shared processCreate methods

[jogrogan]

Problem

HoptimatorDdlExecutor and the !specify dry-run path (used by the CLI, Quidem tests, and more internally) implemented the same DDL pipeline — schema validation, query planning, sink schema setup, deployer invocation, and rollback — in several completely separate code paths. Any fix or feature had to be applied many times, and paths have already drifted:

• The executor had detailed step-level logging the dry-run path lacked
• TemporaryTable was a private inner class of HoptimatorDdlExecutor, inaccessible to the shared path
• The plain SELECT path passed Collections.emptyList() as materializations instead of conn.materializations(), causing source table connector configs to be absent from the generated SQL
• The dry-run path did not support CREATE TABLE statements or INSERT INTO statements properly, only CREATE MATERIALIZED VIEW and SELECT.
• Complicated error-prone "rollback" logic.

Solution

Extract the shared logic into HoptimatorDdlUtils and parameterize the divergence point with a DdlMode enum. This gives us a common entry point to support other SQL DDL types as well.

New / changed production code

TemporaryTable — extracted from a private inner class of HoptimatorDdlExecutor to a standalone public class with a simplified 2-arg constructor. Retains the full InitializerExpressionFactory support (via a 3-arg overload) so that column default expressions and generation strategies are preserved. Note the intent is to reuse this piece later within

DdlMode enum (CREATE / UPDATE / SPECIFY) — executeDeployers() either calls DeploymentService.create/update or collects deployer.specify() output; mutable() gates the schema conflict check.

processCreateMaterializedView() — single 11-step pipeline for CREATE MATERIALIZED VIEW, shared by HoptimatorDdlExecutor and all dry-run callers. Logging routed through conn.getLogger() so the connection's dual-logger (SLF4J + connection hooks) is used, matching the original executor behavior.

processCreateTable() — same for CREATE TABLE. Includes the full ColumnDef / InitializerExpressionFactory machinery for column default expressions so no behavior is dropped.

specifyFromSql() — unified entry point for !specify in the CLI, Quidem tests, and more internally. Handles CREATE TABLE, CREATE MATERIALIZED VIEW, plain SELECT, and INSERT INTO. Returns a SpecifyResult containing:

• specs — YAML artifacts from each deployer
• sinkRowType — output row type, derived from the query plan (no double planning: for CREATE MV, root is computed once and reused for both DeploymentService.plan and the row type)
• viewPath — fully-qualified path of the sink, used by callers to construct Avro namespace and schema name

HoptimatorDdlExecutor — simplified to a thin dispatcher: validates, picks DdlMode.CREATE vs UPDATE, delegates to processCreate*, wraps exceptions as DdlException.

Bug fixes

Bug	Fix
Schema rollback always calls removeTable unconditionally	Delegating to processCreateMaterializedView which uses the correct conditional rollback
pipeline.job().sink() is null for plain SELECT	specifyFromSql SELECT path now registers a virtual DEFAULT.SINK and calls plan.setSink()
Source table connector configs missing in generated SQL	SELECT path now passes conn.materializations() instead of Collections.emptyList()

Test changes

HoptimatorDdlUtilsTest — 40+ new tests covering:

• processCreateMaterializedView and processCreateTable validation paths (schema not found, non-database schema, duplicate view, IF NOT EXISTS, overwrite physical table)
• DdlMode behavior and ColumnDef preconditions
• specifyFromSql SELECT path: correct SpecifyResult fields asserted via
  @Mock MockedStatic<DeploymentService>; schema restoration verified after call

HoptimatorDdlExecutorTest — consolidated HoptimatorDdlExecutorDdlTest and HoptimatorDdlExecutorMockServiceTest into a single file. Trimmed to 2 tests per delegated DDL method (one happy path, one DdlException wrapping) since the validation logic now lives in HoptimatorDdlUtilsTest.

kafka-ddl-create-table.id — added CREATE TABLE IF NOT EXISTS integration test verifying the mutable no-op path returns (0 rows modified) cleanly.

This PR is also a precursor to the coming Logical table work which also needs similar functionality.

[github-actions]

Code Coverage

Overall Project	84.77% -0.7%	🟢
Files changed	83.99%	🟢

File	Coverage
TemporaryTable.java	100%	🟢
HoptimatorDdlExecutor.java	95.83% -0.15%	🟢
HoptimatorDdlUtils.java	86.16% -13.55%	🟢

[ryannedolan]

neat! Much needed improvement.