INSERT INTO / SELECT fixes

Author: jogrogan

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

@@ -622,7 +622,7 @@ public RexNode newColumnDefaultValue(RelOptTable table, int iColumn,
    * {@code CREATE TABLE}, {@code CREATE MATERIALIZED VIEW}, or {@code INSERT INTO}.
    *
    * <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
+   * interactive CLI, and more. 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.
@@ -653,29 +653,65 @@ public static SpecifyResult specifyFromSql(String sql, HoptimatorConnection conn
     }
 
     // Plain SELECT / INSERT INTO path.
-    String viewName = "sink";
+    String viewName = "SINK";
     RelRoot root = HoptimatorDriver.convert(conn, sql).root;
     RelDataType sinkRowType = root.rel.getRowType();
     Properties connectionProperties = conn.connectionProperties();
     RelOptTable table = root.rel.getTable();
+    List<String> viewPath;
     if (table != null) {
       List<String> qualifiedName = table.getQualifiedName();
       connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION,
           String.join(".", qualifiedName));
       viewName = qualifiedName.get(qualifiedName.size() - 1);
+      viewPath = new ArrayList<>(qualifiedName);
+    } else {
+      // No INSERT INTO target — name the virtual sink "SINK" and record it as the pipeline.
+      connectionProperties.setProperty(DeploymentService.PIPELINE_OPTION, viewName);
+      viewPath = new ArrayList<>(List.of("DEFAULT", viewName));
     }
-    List<String> viewPath = new ArrayList<>(List.of("DEFAULT", viewName));
 
-    PipelineRel.Implementor plan = DeploymentService.plan(root, Collections.emptyList(), connectionProperties);
+    PipelineRel.Implementor plan = DeploymentService.plan(root, conn.materializations(), connectionProperties);
 
-    Pipeline pipeline = plan.pipeline(viewName, conn);
-    List<String> specs = new ArrayList<>();
-    for (Source source : pipeline.sources()) {
-      specs.addAll(DeploymentService.specify(source, conn));
+    // For plain SELECT queries (no INSERT INTO target), the planner has no sink.
+    // Set up a virtual sink so that pipeline.job().sink() is non-null
+    CalcitePrepare.Context ctx = conn.createPrepareContext();
+    Pair<SchemaPlus, Table> sinkSnapshot = null;
+    if (table == null) {
+      ViewTable sinkViewTable = viewTable(ctx, sql, new HoptimatorDriver.Prepare(conn),
+          List.of("DEFAULT"), viewPath);
+      sinkRowType = sinkViewTable.getRowType(ctx.getTypeFactory());
+      plan.setSink("DEFAULT", viewPath, sinkRowType, Collections.emptyMap());
+      // Register the virtual sink in the schema so deployers can resolve it; snapshot for rollback.
+      Pair<CalciteSchema, String> sinkSchemaPair = schema(ctx, false,
+          new SqlIdentifier(viewName, SqlParserPos.ZERO));
+      if (sinkSchemaPair.left != null) {
+        SchemaPlus sinkSchemaPlus = sinkSchemaPair.left.plus();
+        Table existing = sinkSchemaPlus.tables().get(sinkSchemaPair.right);
+        sinkSnapshot = Pair.of(sinkSchemaPlus, existing);
+        sinkSchemaPlus.add(sinkSchemaPair.right, new MaterializedViewTable(sinkViewTable));
+      }
+    }
+
+    try {
+      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));
+      return new SpecifyResult(specs, sinkRowType, viewPath);
+    } finally {
+      // Restore the schema — the virtual sink must not persist after this call.
+      if (sinkSnapshot != null) {
+        if (sinkSnapshot.right != null) {
+          sinkSnapshot.left.add(viewName, sinkSnapshot.right);
+        } else {
+          sinkSnapshot.left.removeTable(viewName);
+        }
+      }
     }
-    specs.addAll(DeploymentService.specify(pipeline.sink(), conn));
-    specs.addAll(DeploymentService.specify(pipeline.job(), conn));
-    return new SpecifyResult(specs, sinkRowType, viewPath);
   }
 
   public static Map<String, String> options(SqlNodeList optionList) {

hoptimator-jdbc/src/test/java/com/linkedin/hoptimator/jdbc/HoptimatorDdlUtilsTest.java

@@ -1,7 +1,11 @@
 package com.linkedin.hoptimator.jdbc;
 
 import com.linkedin.hoptimator.Deployer;
+import com.linkedin.hoptimator.Job;
+import com.linkedin.hoptimator.Pipeline;
+import com.linkedin.hoptimator.Sink;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateMaterializedView;
+import com.linkedin.hoptimator.util.DeploymentService;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateTable;
 import com.linkedin.hoptimator.util.planner.PipelineRel;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@@ -26,6 +30,8 @@
 import org.apache.calcite.util.Pair;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
 import org.mockito.junit.jupiter.MockitoExtension;
 
 import java.sql.SQLException;
@@ -41,6 +47,7 @@
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -50,6 +57,9 @@
     justification = "Mock objects created in stubbing setup don't need resource management")
 class HoptimatorDdlUtilsTest {
 
+  @Mock
+  MockedStatic<DeploymentService> mockDeploymentService;
+
   @Test
   void testRenameColumnsReturnsQueryWhenColumnListNull() {
     SqlNode query = SqlLiteral.createCharString("test", SqlParserPos.ZERO);
@@ -528,6 +538,93 @@ void specifyFromSqlWithCreateMaterializedViewDryRunDoesNotThrowOnSchemaFailure()
     }
   }
 
+  // ── specifyFromSql() plain SELECT path ───────────────────────────────────────
+
+  @Test
+  void specifyFromSqlForPlainSelectReturnsSinkPathAndRowType() throws Exception {
+    // Regression test: before the fix, pipeline.job().sink() was null for plain SELECT
+    // (no INSERT INTO target), causing NullPointerException. Now setSink() anchors a virtual
+    // "DEFAULT"."sink" so the pipeline can be fully constructed.
+    Pipeline mockPipeline = mock(Pipeline.class);
+    Sink mockSink = mock(Sink.class);
+    Job mockJob = mock(Job.class);
+    when(mockPipeline.sources()).thenReturn(Collections.emptyList());
+    when(mockPipeline.sink()).thenReturn(mockSink);
+    when(mockPipeline.job()).thenReturn(mockJob);
+
+    PipelineRel.Implementor mockPlan = mock(PipelineRel.Implementor.class);
+    when(mockPlan.pipeline(any(), any())).thenReturn(mockPipeline);
+
+    mockDeploymentService.when(() -> DeploymentService.plan(any(), any(), any())).thenReturn(mockPlan);
+    mockDeploymentService.when(() -> DeploymentService.specify(any(), any()))
+        .thenReturn(Collections.emptyList());
+
+    HoptimatorDriver driver = new HoptimatorDriver();
+    try (HoptimatorConnection conn =
+        (HoptimatorConnection) driver.connect("jdbc:hoptimator://catalogs=util", new Properties())) {
+      HoptimatorDdlUtils.SpecifyResult result = HoptimatorDdlUtils.specifyFromSql(
+          "SELECT 1 AS \"KEY\", 'value' AS \"VAL\"", conn);
+
+      // Specs: empty (no deployers return anything)
+      assertNotNull(result.specs);
+      assertTrue(result.specs.isEmpty());
+      // viewPath last element is "SINK" (uppercase) for a bare SELECT, matching original plan() behavior
+      assertNotNull(result.viewPath);
+      assertEquals("SINK", result.viewPath.get(result.viewPath.size() - 1));
+      // sinkRowType matches the SELECT output columns
+      assertNotNull(result.sinkRowType);
+      assertEquals(2, result.sinkRowType.getFieldCount());
+      assertEquals("KEY", result.sinkRowType.getFieldList().get(0).getName());
+      assertEquals("VAL", result.sinkRowType.getFieldList().get(1).getName());
+    }
+  }
+
+  @Test
+  void specifyFromSqlCreateTableViewPathMatchesSchemaAndName() throws Exception {
+    // For CREATE TABLE, the SpecifyResult.viewPath should include the schema path + table name.
+    HoptimatorDriver driver = new HoptimatorDriver();
+    try (HoptimatorConnection conn =
+        (HoptimatorConnection) driver.connect("jdbc:hoptimator://catalogs=util", new Properties())) {
+      conn.calciteConnection().getRootSchema().add("VP_DB", new TestDatabaseSchema("vp-db"));
+      HoptimatorDdlUtils.SpecifyResult result = HoptimatorDdlUtils.specifyFromSql(
+          "CREATE TABLE \"VP_DB\".\"events\" (\"id\" INTEGER, \"msg\" VARCHAR)", conn);
+      assertNotNull(result.viewPath);
+      assertFalse(result.viewPath.isEmpty());
+      assertEquals("events", result.viewPath.get(result.viewPath.size() - 1));
+      assertTrue(result.viewPath.contains("VP_DB"),
+          "viewPath should contain the schema: " + result.viewPath);
+    }
+  }
+
+  @Test
+  void specifyFromSqlForPlainSelectRestoresSchemaAfterCall() throws Exception {
+    // The virtual "SINK" table registered during the SELECT path must not persist after
+    // specifyFromSql returns (even when it throws at the deployer level).
+    HoptimatorDriver driver = new HoptimatorDriver();
+    try (HoptimatorConnection conn =
+        (HoptimatorConnection) driver.connect("jdbc:hoptimator://catalogs=util", new Properties())) {
+      CalcitePrepare.Context ctx = conn.createPrepareContext();
+
+      // Capture schema state before the call.
+      Pair<CalciteSchema, String> pair = HoptimatorDdlUtils.schema(ctx, false,
+          new SqlIdentifier("sink", SqlParserPos.ZERO));
+      boolean existedBefore = pair.left != null
+          && pair.left.plus().tables().get(pair.right) != null;
+
+      try {
+        HoptimatorDdlUtils.specifyFromSql("SELECT 1 AS \"COL1\"", conn);
+      } catch (Exception ignored) {
+        // Expected: planning/deployer failure in the test environment.
+      }
+
+      // Schema must be in the same state as before.
+      boolean existsAfter = pair.left != null
+          && pair.left.plus().tables().get(pair.right) != null;
+      assertEquals(existedBefore, existsAfter,
+          "Virtual SINK table must be removed after specifyFromSql returns");
+    }
+  }
+
   @Test
   void optionsParsesSingleKeyValue() {
     SqlNodeList list = new SqlNodeList(SqlParserPos.ZERO);