Wrap results in spec

Author: jogrogan

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

@@ -277,7 +277,7 @@ public void execute(String line, DispatchCallback dispatchCallback) {
       String sql = split[1];
       HoptimatorConnection conn = (HoptimatorConnection) sqlline.getConnection();
       try {
-        List<String> specs = HoptimatorDdlUtils.specifyFromSql(sql, 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);

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/HoptimatorDdlUtils.java

@@ -89,6 +89,29 @@ public final class HoptimatorDdlUtils {
   private HoptimatorDdlUtils() {
   }
 
+  /**
+   * The result of a {@link #specifyFromSql} call: the YAML artifact specs, the sink row type,
+   * and the fully-qualified path of the sink (viewPath).
+   *
+   * <ul>
+   *   <li>{@code specs} — YAML artifacts produced by each deployer (empty for dry-run no-ops).
+   *   <li>{@code sinkRowType} — row type of the sink; same as the (renamed) query output.
+   *   <li>{@code viewPath} — fully-qualified path including catalog, schema, and view name.
+   * </ul>
+   */
+  public static final class SpecifyResult {
+    public final List<String> specs;
+    public final RelDataType sinkRowType;
+    /** Fully-qualified path of the sink (catalog + schema + table). */
+    public final List<String> viewPath;
+
+    SpecifyResult(List<String> specs, RelDataType sinkRowType, List<String> viewPath) {
+      this.specs = Collections.unmodifiableList(specs);
+      this.sinkRowType = sinkRowType;
+      this.viewPath = Collections.unmodifiableList(viewPath);
+    }
+  }
+
   /**
    * Controls whether a DDL operation performs a real deployment (CREATE or UPDATE)
    * or a dry-run (SPECIFY).
@@ -267,14 +290,23 @@ public static Pair<SchemaPlus, Table> snapshotAndSetSinkSchema(CalcitePrepare.Co
    * @return an empty list for CREATE/UPDATE, or the YAML spec strings for SPECIFY
    * @throws SQLException on validation or deployment errors
    */
-  static List<String> processCreateMaterializedView(CalcitePrepare.Context ctx,
+  static SpecifyResult processCreateMaterializedView(CalcitePrepare.Context ctx,
       HoptimatorDriver.Prepare prepare, HoptimatorConnection conn,
       SqlCreateMaterializedView create, DdlMode mode) throws SQLException {
     HoptimatorConnection.HoptimatorConnectionDualLogger logger = conn.getLogger(HoptimatorDdlUtils.class);
     // Validate the DDL statement.
     logger.info("Validating statement: {}", create);
     ValidationService.validateOrThrow(create);
 
+    // Extract query SQL (rename columns if a column list was provided) and plan the query.
+    // This is done first — before schema/conflict checks — so that:
+    //   1. sinkRowType is always available, even for IF NOT EXISTS early returns.
+    //   2. root is computed once and reused for pipeline planning below.
+    final SqlNode q = renameColumns(create.columnList, create.query);
+    final String sql = q.toSqlString(CalciteSqlDialect.DEFAULT).getSql();
+    RelRoot root = prepare.convert(ctx, sql).root;
+    RelDataType sinkRowType = root.rel.getRowType();
+
     // Navigate to the schema (mutable only when actually deploying).
     final Pair<CalciteSchema, String> pair = schema(ctx, mode.mutable(), create.name);
     if (pair.left == null) {
@@ -297,7 +329,9 @@ static List<String> processCreateMaterializedView(CalcitePrepare.Context ctx,
         schemaPlus.removeTable(pair.right);
       } else {
         // IF NOT EXISTS — nothing to do.
-        return Collections.emptyList();
+        List<String> viewPath = new ArrayList<>(pair.left.path(null));
+        viewPath.add(pair.right);
+        return new SpecifyResult(Collections.emptyList(), sinkRowType, viewPath);
       }
     }
 
@@ -306,9 +340,6 @@ static List<String> processCreateMaterializedView(CalcitePrepare.Context ctx,
     }
     String database = ((Database) pair.left.schema).databaseName();
 
-    // Extract query SQL (rename columns if a column list was provided).
-    final SqlNode q = renameColumns(create.columnList, create.query);
-    final String sql = q.toSqlString(CalciteSqlDialect.DEFAULT).getSql();
     final List<String> schemaPath = pair.left.path(null);
     String viewName = pair.right;
     List<String> viewPath = new ArrayList<>(schemaPath);
@@ -327,8 +358,7 @@ static List<String> processCreateMaterializedView(CalcitePrepare.Context ctx,
     Properties connectionProperties = conn.connectionProperties();
     connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, pipelineName);
 
-    // Plan the pipeline to materialize the view.
-    RelRoot root = prepare.convert(ctx, sql).root;
+    // Plan the pipeline
     PipelineRel.Implementor plan = DeploymentService.plan(root, conn.materializations(), connectionProperties);
 
     // Snapshot the current schema state and set the sink schema for planning.
@@ -358,12 +388,12 @@ static List<String> processCreateMaterializedView(CalcitePrepare.Context ctx,
       } else {
         logger.info("Specifying materialized view {}", viewName);
       }
-      List<String> result = mode.executeDeployers(deployers, conn);
+      List<String> specs = mode.executeDeployers(deployers, conn);
       if (mode.mutable()) {
         logger.info("Deployed materialized view {}", viewName);
       }
       success = true;
-      return result;
+      return new SpecifyResult(specs, sinkRowType, viewPath);
     } catch (SQLException | RuntimeException e) {
       logger.info("Failed to deploy materialized view {}", viewName);
       if (deployers != null) {
@@ -404,7 +434,7 @@ static List<String> processCreateMaterializedView(CalcitePrepare.Context ctx,
    * @return an empty list for CREATE/UPDATE, or the YAML spec strings for SPECIFY
    * @throws SQLException on validation or deployment errors
    */
-  static List<String> processCreateTable(CalcitePrepare.Context ctx, HoptimatorConnection conn,
+  static SpecifyResult processCreateTable(CalcitePrepare.Context ctx, HoptimatorConnection conn,
       SqlCreateTable create, DdlMode mode) throws SQLException {
     HoptimatorConnection.HoptimatorConnectionDualLogger logger = conn.getLogger(HoptimatorDdlUtils.class);
 
@@ -552,12 +582,12 @@ public RexNode newColumnDefaultValue(RelOptTable table, int iColumn,
       } else {
         logger.info("Specifying table {}", source);
       }
-      List<String> result = mode.executeDeployers(deployers, conn);
+      List<String> specs = mode.executeDeployers(deployers, conn);
       if (mode.mutable()) {
         logger.info("Deployed table {}", source);
       }
       success = true;
-      return result;
+      return new SpecifyResult(specs, rowType, tablePath);
     } catch (SQLException | RuntimeException e) {
       logger.info("Failed to deploy table {}", tableName);
       if (deployers != null) {
@@ -594,8 +624,19 @@ public RexNode newColumnDefaultValue(RelOptTable table, int iColumn,
    * <p>This is the shared implementation behind {@code !specify} in Quidem tests, the
    * interactive CLI, and any gRPC plan endpoint. It is a strict dry-run: no schema
    * mutations, no deployer create/update calls.
+   * Returns both the YAML artifact specs and the sink row type for the given SQL statement in
+   * a single pass.
+   *
+   * <p>Supported statement types:
+   * <ul>
+   *   <li><b>CREATE MATERIALIZED VIEW</b> — column renames applied; query planned once.
+   *   <li><b>CREATE TABLE</b> — row type from column declarations; no query planning.
+   *   <li><b>SELECT / INSERT INTO</b> — query output row type.
+   * </ul>
+   *
+   * @throws SQLException for unsupported DDL (e.g. DROP, CREATE VIEW)
    */
-  public static List<String> specifyFromSql(String sql, HoptimatorConnection conn) throws SQLException {
+  public static SpecifyResult specifyFromSql(String sql, HoptimatorConnection conn) throws SQLException {
     SqlNode sqlNode = HoptimatorDriver.parseQuery(conn, sql);
 
     if (sqlNode instanceof SqlCreateTable) {
@@ -612,19 +653,18 @@ public static List<String> specifyFromSql(String sql, HoptimatorConnection conn)
     }
 
     // Plain SELECT / INSERT INTO path.
-    String querySql = sql;
     String viewName = "sink";
-
-    RelRoot root = HoptimatorDriver.convert(conn, querySql).root;
+    RelRoot root = HoptimatorDriver.convert(conn, sql).root;
+    RelDataType sinkRowType = root.rel.getRowType();
     Properties connectionProperties = conn.connectionProperties();
     RelOptTable table = root.rel.getTable();
     if (table != null) {
       List<String> qualifiedName = table.getQualifiedName();
       connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION,
           String.join(".", qualifiedName));
-      // Use the sink table's simple name as the pipeline/job name (matches !specify TABLE behavior)
       viewName = qualifiedName.get(qualifiedName.size() - 1);
     }
+    List<String> viewPath = new ArrayList<>(List.of("DEFAULT", viewName));
 
     PipelineRel.Implementor plan = DeploymentService.plan(root, Collections.emptyList(), connectionProperties);
 
@@ -635,7 +675,7 @@ public static List<String> specifyFromSql(String sql, HoptimatorConnection conn)
     }
     specs.addAll(DeploymentService.specify(pipeline.sink(), conn));
     specs.addAll(DeploymentService.specify(pipeline.job(), conn));
-    return specs;
+    return new SpecifyResult(specs, sinkRowType, viewPath);
   }
 
   public static Map<String, String> options(SqlNodeList optionList) {