Add apply mode for declarative, idempotent CREATE DDL

docs/user-guide/ddl-reference.md

@@ -14,6 +14,39 @@ This page documents the DDL Hoptimator adds on top.
 > Reading them is a fast way to see currently-passing examples of every
 > DDL form.
 
+## CREATE semantics: strict vs. apply mode
+
+`CREATE` has two flavors, selected per-connection via the `mode` property:
+
+| Mode               | `CREATE`                                | `CREATE OR REPLACE`  |
+|--------------------|-----------------------------------------|----------------------|
+| `create` (default) | Fail if the resource exists.            | Update if it exists. |
+| `apply`            | Converge the resource to the definition. Idempotent. | Same as `CREATE`. |
+
+`create` mode is the imperative default — good for interactive sessions where
+you want a real error if you accidentally redefine something.
+
+`apply` mode is the declarative, K8s-style flavor: a `.sql` file becomes a
+manifest, and re-running it converges resources to the declared state. It's
+designed for check-in-and-reconcile workflows where CI runs the same script
+on every merge. Set it on the JDBC URL:
+
+```
+jdbc:hoptimator://...;mode=apply
+```
+
+Both modes only ever touch **metadata**. Underlying data is never destroyed by
+any CREATE form — even `OR REPLACE` and `DROP` are metadata-only operations,
+and the data store retains its rows. This is the safety guarantee that makes
+apply mode usable in production.
+
+**Out of scope today:** prune-by-absence (the K8s `--prune` equivalent — a
+resource missing from the declared script does *not* trigger a drop). `DROP`
+remains imperative and explicit. Detection of *incompatible* metadata changes
+(e.g. dropping a primary key) is also not yet wired in; in apply mode both
+`CREATE` and `CREATE OR REPLACE` apply the new definition regardless. Use
+caution when changing schemas of in-flight pipelines.
+
 ## CREATE VIEW
 
 ```

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

@@ -123,9 +123,10 @@ public void execute(SqlCreateView create, CalcitePrepare.Context context) {
       throw new DdlException(create,
           "Cannot overwrite physical table " + pair.right + " with a view.");
     }
+    HoptimatorDdlUtils.DdlMode mode = HoptimatorDdlUtils.effectiveMode(create.getReplace(), connection);
     for (Function function : schemaPlus.getFunctions(pair.right)) {
       if (function.getParameters().isEmpty()) {
-        if (!create.getReplace()) {
+        if (mode == HoptimatorDdlUtils.DdlMode.CREATE) {
           throw new DdlException(create,
               "View " + pair.right + " already exists. Use CREATE OR REPLACE to update.");
         }
@@ -151,7 +152,7 @@ public void execute(SqlCreateView create, CalcitePrepare.Context context) {
       deployers = DeploymentService.deployers(view, connection);
       ValidationService.validateOrThrow(deployers, connection);
       logger.info("Validated view {}", viewName);
-      if (create.getReplace()) {
+      if (mode == HoptimatorDdlUtils.DdlMode.UPDATE) {
         logger.info("Deploying update view {}", viewName);
         DeploymentService.update(deployers);
       } else {
@@ -177,8 +178,7 @@ public void execute(SqlCreateView create, CalcitePrepare.Context context) {
   /** Executes a {@code CREATE MATERIALIZED VIEW} command. */
   public void execute(SqlCreateMaterializedView create, CalcitePrepare.Context context) {
     logger.info("Validating statement: {}", create);
-    HoptimatorDdlUtils.DdlMode mode = create.getReplace()
-        ? HoptimatorDdlUtils.DdlMode.UPDATE : HoptimatorDdlUtils.DdlMode.CREATE;
+    HoptimatorDdlUtils.DdlMode mode = HoptimatorDdlUtils.effectiveMode(create.getReplace(), connection);
     try {
       HoptimatorDdlUtils.processCreateMaterializedView(
           context,
@@ -240,7 +240,8 @@ public void execute(SqlCreateTrigger create, CalcitePrepare.Context context) {
       deployers = DeploymentService.deployers(trigger, connection);
       ValidationService.validateOrThrow(deployers, connection);
       logger.info("Validated trigger {}", name);
-      if (create.getReplace()) {
+      HoptimatorDdlUtils.DdlMode mode = HoptimatorDdlUtils.effectiveMode(create.getReplace(), connection);
+      if (mode == HoptimatorDdlUtils.DdlMode.UPDATE) {
         logger.info("Updating trigger {}", name);
         DeploymentService.update(deployers);
       } else {
@@ -278,8 +279,7 @@ private static String databaseOf(Table target) {
 
   /** Executes a {@code CREATE TABLE} command. */
   public void execute(SqlCreateTable create, CalcitePrepare.Context context) {
-    HoptimatorDdlUtils.DdlMode mode = create.getReplace()
-        ? HoptimatorDdlUtils.DdlMode.UPDATE : HoptimatorDdlUtils.DdlMode.CREATE;
+    HoptimatorDdlUtils.DdlMode mode = HoptimatorDdlUtils.effectiveMode(create.getReplace(), connection);
     try {
       HoptimatorDdlUtils.processCreateTable(context, connection, create, mode);
     } catch (SQLException | RuntimeException e) {
@@ -291,8 +291,7 @@ public void execute(SqlCreateTable create, CalcitePrepare.Context context) {
 
   /** Executes a {@code CREATE DATABASE} command. */
   public void execute(SqlCreateDatabase create, CalcitePrepare.Context context) {
-    HoptimatorDdlUtils.DdlMode mode = create.getReplace()
-        ? HoptimatorDdlUtils.DdlMode.UPDATE : HoptimatorDdlUtils.DdlMode.CREATE;
+    HoptimatorDdlUtils.DdlMode mode = HoptimatorDdlUtils.effectiveMode(create.getReplace(), connection);
     try {
       HoptimatorDdlUtils.processCreateDatabase(connection, create, mode);
     } catch (SQLException | RuntimeException e) {

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

@@ -89,6 +89,53 @@ public final class HoptimatorDdlUtils {
   private HoptimatorDdlUtils() {
   }
 
+  /**
+   * Connection property that controls how plain {@code CREATE} statements behave.
+   *
+   * <ul>
+   *   <li>{@code create} (default) — strict create. {@code CREATE} fails if the resource
+   *       already exists; {@code CREATE OR REPLACE} updates.</li>
+   *   <li>{@code apply} — declarative, K8s-style reconciliation. Both {@code CREATE} and
+   *       {@code CREATE OR REPLACE} converge the resource to the declared definition, so
+   *       running the same script twice is a no-op. Idempotent by design.</li>
+   * </ul>
+   *
+   * <p>The property may also be set per statement via {@code SET mode = 'apply'} on a
+   * compatible session — see {@code docs/user-guide/ddl-reference.md}.
+   */
+  public static final String MODE_PROPERTY = "mode";
+
+  /** Default value of {@link #MODE_PROPERTY}. */
+  public static final String MODE_CREATE = "create";
+
+  /** Apply-mode value of {@link #MODE_PROPERTY}. */
+  public static final String MODE_APPLY = "apply";
+
+  /**
+   * Resolves the effective {@link DdlMode} for a {@code CREATE} statement, combining the
+   * statement's {@code OR REPLACE} flag with the connection's {@link #MODE_PROPERTY}.
+   *
+   * <p>In {@code create} mode (the default): {@code CREATE} → {@link DdlMode#CREATE} and
+   * {@code CREATE OR REPLACE} → {@link DdlMode#UPDATE}. In {@code apply} mode: both forms
+   * resolve to {@link DdlMode#UPDATE}, making CREATE idempotent.
+   */
+  static DdlMode effectiveMode(boolean orReplace, HoptimatorConnection conn) {
+    if (isApplyMode(conn)) {
+      return DdlMode.UPDATE;
+    }
+    return orReplace ? DdlMode.UPDATE : DdlMode.CREATE;
+  }
+
+  /** Whether the connection is configured for apply-mode DDL. */
+  static boolean isApplyMode(HoptimatorConnection conn) {
+    Properties props = conn.connectionProperties();
+    if (props == null) {
+      return false;
+    }
+    String mode = props.getProperty(MODE_PROPERTY, MODE_CREATE);
+    return MODE_APPLY.equalsIgnoreCase(mode);
+  }
+
   /**
    * The result of a {@link #specifyFromSql} call: the YAML artifact specs, the sink row type,
    * and the fully-qualified path of the sink (viewPath).
@@ -320,12 +367,27 @@ static SpecifyResult processCreateMaterializedView(CalcitePrepare.Context ctx,
         throw new SQLException(
             "Cannot overwrite physical table " + pair.right + " with a view.");
       }
-      // Materialized view exists.
-      if (!create.ifNotExists && !create.getReplace()) {
-        throw new SQLException(
-            "View " + pair.right + " already exists. Use CREATE OR REPLACE to update.");
+      // A view already exists. The executor pre-resolved apply-mode into DdlMode.UPDATE,
+      // so UPDATE always means "converge to this definition". Strict CREATE is the only
+      // path that errors. SPECIFY (dry-run) keeps the original syntax-driven preview so
+      // it accurately reflects what a real run would do.
+      boolean replaceExisting;
+      if (mode == DdlMode.UPDATE) {
+        replaceExisting = true;
+      } else if (mode == DdlMode.CREATE) {
+        if (!create.ifNotExists) {
+          throw new SQLException(
+              "View " + pair.right + " already exists. Use CREATE OR REPLACE to update.");
+        }
+        replaceExisting = false;
+      } else { // SPECIFY
+        if (!create.ifNotExists && !create.getReplace()) {
+          throw new SQLException(
+              "View " + pair.right + " already exists. Use CREATE OR REPLACE to update.");
+        }
+        replaceExisting = create.getReplace();
       }
-      if (create.getReplace()) {
+      if (replaceExisting) {
         schemaPlus.removeTable(pair.right);
       } else {
         // IF NOT EXISTS — nothing to do.
@@ -479,8 +541,18 @@ static SpecifyResult processCreateTable(CalcitePrepare.Context ctx, HoptimatorCo
     }
 
     if (!isNewSchema && schemaPlus.tables().get(tableName) != null) {
-      if (!create.ifNotExists && !create.getReplace()) {
-        // They did not specify IF NOT EXISTS, so give error.
+      // Strict CREATE without IF NOT EXISTS is the only path that errors. UPDATE
+      // (apply mode or explicit OR REPLACE) targets the existing table; SPECIFY
+      // (dry-run) preserves its syntax-driven semantics.
+      boolean wouldFail;
+      if (mode == DdlMode.UPDATE) {
+        wouldFail = false;
+      } else if (mode == DdlMode.CREATE) {
+        wouldFail = !create.ifNotExists;
+      } else { // SPECIFY
+        wouldFail = !create.ifNotExists && !create.getReplace();
+      }
+      if (wouldFail) {
         throw new SQLException(
             "Table " + tableName + " already exists. Use CREATE OR REPLACE to update.");
       }

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

@@ -1393,8 +1393,9 @@ void processCreateMaterializedViewCreateOrReplaceReplacesExistingViewWithMateria
 
       CalcitePrepare.Context ctx = conn.createPrepareContext();
       HoptimatorDriver.Prepare prepare = new HoptimatorDriver.Prepare(conn);
+      // The executor pre-resolves OR REPLACE to DdlMode.UPDATE before calling this helper.
       HoptimatorDdlUtils.processCreateMaterializedView(
-          ctx, prepare, conn, create, HoptimatorDdlUtils.DdlMode.CREATE);
+          ctx, prepare, conn, create, HoptimatorDdlUtils.DdlMode.UPDATE);
 
       // After CREATE OR REPLACE, the old ViewTable is gone and a MaterializedViewTable takes its place
       Table table = replaceDbSchema.tables().get("existingView");
@@ -1697,4 +1698,90 @@ void removeTableFromSchemaIsNoOpWhenEntriesMissing() throws SQLException {
           "missing-table call must not create the table");
     }
   }
+
+  /** Helper: a HoptimatorConnection mock that returns the given Properties. */
+  private HoptimatorConnection connectionWith(Properties props) {
+    HoptimatorConnection conn = mock(HoptimatorConnection.class);
+    lenient().when(conn.connectionProperties()).thenReturn(props);
+    return conn;
+  }
+
+  @Test
+  void testEffectiveModeDefaultsToStrictCreate() {
+    HoptimatorConnection conn = connectionWith(new Properties());
+
+    assertEquals(HoptimatorDdlUtils.DdlMode.CREATE,
+        HoptimatorDdlUtils.effectiveMode(false, conn));
+    assertEquals(HoptimatorDdlUtils.DdlMode.UPDATE,
+        HoptimatorDdlUtils.effectiveMode(true, conn));
+  }
+
+  @Test
+  void testEffectiveModeExplicitCreateMatchesDefault() {
+    Properties props = new Properties();
+    props.setProperty(HoptimatorDdlUtils.MODE_PROPERTY, HoptimatorDdlUtils.MODE_CREATE);
+    HoptimatorConnection conn = connectionWith(props);
+
+    assertEquals(HoptimatorDdlUtils.DdlMode.CREATE,
+        HoptimatorDdlUtils.effectiveMode(false, conn));
+    assertEquals(HoptimatorDdlUtils.DdlMode.UPDATE,
+        HoptimatorDdlUtils.effectiveMode(true, conn));
+  }
+
+  @Test
+  void testEffectiveModeApplyMapsBothCreateFormsToUpdate() {
+    Properties props = new Properties();
+    props.setProperty(HoptimatorDdlUtils.MODE_PROPERTY, HoptimatorDdlUtils.MODE_APPLY);
+    HoptimatorConnection conn = connectionWith(props);
+
+    // The whole point of apply mode: plain CREATE becomes idempotent (UPDATE).
+    assertEquals(HoptimatorDdlUtils.DdlMode.UPDATE,
+        HoptimatorDdlUtils.effectiveMode(false, conn));
+    // CREATE OR REPLACE keeps converging behavior in apply mode.
+    assertEquals(HoptimatorDdlUtils.DdlMode.UPDATE,
+        HoptimatorDdlUtils.effectiveMode(true, conn));
+  }
+
+  @Test
+  void testEffectiveModeApplyIsCaseInsensitive() {
+    Properties props = new Properties();
+    props.setProperty(HoptimatorDdlUtils.MODE_PROPERTY, "APPLY");
+    HoptimatorConnection conn = connectionWith(props);
+
+    assertEquals(HoptimatorDdlUtils.DdlMode.UPDATE,
+        HoptimatorDdlUtils.effectiveMode(false, conn));
+  }
+
+  @Test
+  void testEffectiveModeUnknownValueFallsBackToStrictCreate() {
+    Properties props = new Properties();
+    props.setProperty(HoptimatorDdlUtils.MODE_PROPERTY, "nonsense");
+    HoptimatorConnection conn = connectionWith(props);
+
+    // Unknown values must not silently promote to apply — typos shouldn't change behavior.
+    assertEquals(HoptimatorDdlUtils.DdlMode.CREATE,
+        HoptimatorDdlUtils.effectiveMode(false, conn));
+  }
+
+  @Test
+  void testEffectiveModeTolerantOfNullProperties() {
+    HoptimatorConnection conn = mock(HoptimatorConnection.class);
+    lenient().when(conn.connectionProperties()).thenReturn(null);
+
+    assertEquals(HoptimatorDdlUtils.DdlMode.CREATE,
+        HoptimatorDdlUtils.effectiveMode(false, conn));
+  }
+
+  @Test
+  void testIsApplyMode() {
+    Properties applyProps = new Properties();
+    applyProps.setProperty(HoptimatorDdlUtils.MODE_PROPERTY, HoptimatorDdlUtils.MODE_APPLY);
+    assertTrue(HoptimatorDdlUtils.isApplyMode(connectionWith(applyProps)));
+
+    Properties createProps = new Properties();
+    createProps.setProperty(HoptimatorDdlUtils.MODE_PROPERTY, HoptimatorDdlUtils.MODE_CREATE);
+    assertFalse(HoptimatorDdlUtils.isApplyMode(connectionWith(createProps)));
+
+    assertFalse(HoptimatorDdlUtils.isApplyMode(connectionWith(new Properties())));
+  }
 }

hoptimator-k8s/src/test/java/com/linkedin/hoptimator/k8s/TestSqlScripts.java

@@ -24,6 +24,11 @@ public void k8sValidationScript() throws Exception {
     run("k8s-validation.id");
   }
 
+  @Test
+  public void k8sApplyMode() throws Exception {
+    run("k8s-apply-mode.id", "mode=apply");
+  }
+
   @Test
   public void k8sMetadataTables() throws Exception {
     run("k8s-metadata.id");

hoptimator-k8s/src/test/resources/k8s-apply-mode.id

@@ -0,0 +1,58 @@
+!set outputformat mysql
+!use k8s
+
+-- In apply mode, plain CREATE is idempotent: it converges the resource to the
+-- declared definition. The same script can run twice with no error.
+
+create view ads.audience as select first_name, last_name from ads.page_views natural join profile.members;
+(0 rows modified)
+
+!update
+
+-- Re-running the same CREATE is a no-op convergence, not an error. In the default
+-- (strict) mode this would fail with "View AUDIENCE already exists. Use CREATE OR REPLACE
+-- to update." — see k8s-validation.id for that contract.
+
+create view ads.audience as select first_name, last_name from ads.page_views natural join profile.members;
+(0 rows modified)
+
+!update
+
+-- Updating the definition through a plain CREATE works the same as CREATE OR REPLACE.
+
+create view ads.audience as select first_name from ads.page_views natural join profile.members;
+(0 rows modified)
+
+!update
+
+-- CREATE OR REPLACE remains valid (and equivalent to CREATE) in apply mode.
+
+create or replace view ads.audience as select last_name from ads.page_views natural join profile.members;
+(0 rows modified)
+
+!update
+
+-- Plain CREATE TABLE is likewise idempotent in apply mode. The default mode would
+-- reject a re-create with "Table X already exists. Use CREATE OR REPLACE to update."
+
+create table ads.apply_table ("k" varchar, "v" varchar);
+(0 rows modified)
+
+!update
+
+create table ads.apply_table ("k" varchar, "v" varchar);
+(0 rows modified)
+
+!update
+
+-- DROP semantics are unchanged: explicit, imperative, metadata-only.
+
+drop view ads.audience;
+(0 rows modified)
+
+!update
+
+drop table ads.apply_table;
+(0 rows modified)
+
+!update