refactor: unify DDL execution and dry-run into shared processCreate* methods

Extract TemporaryTable to a standalone public class. Add DdlMode enum
(CREATE/UPDATE/SPECIFY), processCreateMaterializedView(), and
processCreateTable() to HoptimatorDdlUtils so HoptimatorDdlExecutor
fully delegates to them. Add specifyFromSql() and specifyCreateTable()
as the unified !specify entry points used by QuidemTestBase and
HoptimatorAppConfig.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

Author: jogrogan

hoptimator-cli/src/main/java/sqlline/HoptimatorAppConfig.java

@@ -1,7 +1,5 @@
 package sqlline;
 
-import com.linkedin.hoptimator.Pipeline;
-import com.linkedin.hoptimator.Source;
 import com.linkedin.hoptimator.SqlDialect;
 import com.linkedin.hoptimator.jdbc.HoptimatorConnection;
 import com.linkedin.hoptimator.jdbc.HoptimatorDdlUtils;
@@ -278,56 +276,13 @@ public void execute(String line, DispatchCallback dispatchCallback) {
       }
       String sql = split[1];
       HoptimatorConnection conn = (HoptimatorConnection) sqlline.getConnection();
-      Pair<SchemaPlus, Table> schemaSnapshot = null;
-      String viewName = "sink";
       try {
-        String querySql = sql;
-        SqlCreateMaterializedView create = null;
-        SqlNode sqlNode = HoptimatorDriver.parseQuery(conn, sql);
-        if (sqlNode.getKind().belongsTo(SqlKind.DDL)) {
-          if (sqlNode instanceof SqlCreateMaterializedView) {
-            create = (SqlCreateMaterializedView) sqlNode;
-            final SqlNode q = HoptimatorDdlUtils.renameColumns(create.columnList, create.query);
-            querySql = q.toSqlString(CalciteSqlDialect.DEFAULT).getSql();
-            viewName = HoptimatorDdlUtils.viewName(create.name);
-          } else {
-            sqlline.error("Unsupported DDL statement: " + sql);
-            dispatchCallback.setToFailure();
-            return;
-          }
-        }
-
-        RelRoot root = HoptimatorDriver.convert(conn, querySql).root;
-        Properties connectionProperties = conn.connectionProperties();
-        RelOptTable table = root.rel.getTable();
-        if (table != null) {
-          connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, String.join(".", table.getQualifiedName()));
-        } else if (create != null) {
-          connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, create.name.toString());
-        }
-        PipelineRel.Implementor plan = DeploymentService.plan(root, conn.materializations(), connectionProperties);
-        if (create != null) {
-          schemaSnapshot = HoptimatorDdlUtils.snapshotAndSetSinkSchema(conn.createPrepareContext(),
-              new HoptimatorDriver.Prepare(conn), plan, create, querySql);
-        }
-        Pipeline pipeline = plan.pipeline(viewName, conn);
-        List<String> specs = new ArrayList<>();
-        for (Source source : pipeline.sources()) {
-          specs.addAll(DeploymentService.specify(source, conn));
-        }
-        specs.addAll(DeploymentService.specify(pipeline.sink(), conn));
-        specs.addAll(DeploymentService.specify(pipeline.job(), conn));
+        List<String> specs = HoptimatorDdlUtils.specifyFromSql(sql, conn).specs;
         specs.forEach(x -> sqlline.output(x + "\n\n---\n\n"));
       } catch (SQLException e) {
         sqlline.error(e);
         dispatchCallback.setToFailure();
       }
-      if (schemaSnapshot != null) {
-        if (schemaSnapshot.right != null) {
-          schemaSnapshot.left.add(viewName, schemaSnapshot.right);
-        }
-        schemaSnapshot.left.removeTable(viewName);
-      }
     }
 
     @Override