Add CREATE DATABASE DDL with Deployer mechanism for K8s Database objects

hoptimator-api/src/main/java/com/linkedin/hoptimator/DatabaseDeployable.java

@@ -0,0 +1,29 @@
+package com.linkedin.hoptimator;
+
+import java.util.Map;
+
+
+/** Represents a CREATE DATABASE request. */
+public class DatabaseDeployable implements Deployable {
+
+  private final String name;
+  private final Map<String, String> options;
+
+  public DatabaseDeployable(String name, Map<String, String> options) {
+    this.name = name;
+    this.options = options;
+  }
+
+  public String name() {
+    return name;
+  }
+
+  public Map<String, String> options() {
+    return options;
+  }
+
+  @Override
+  public String toString() {
+    return "Database[" + name + "]";
+  }
+}

hoptimator-jdbc/src/main/codegen/config.fmpp

@@ -32,6 +32,7 @@ data: {
       "org.apache.calcite.sql.SqlTruncate"
       "org.apache.calcite.sql.ddl.SqlCreateTableLike"
       "org.apache.calcite.sql.ddl.SqlDdlNodes"
+      "com.linkedin.hoptimator.jdbc.ddl.SqlCreateDatabase"
       "com.linkedin.hoptimator.jdbc.ddl.SqlCreateFunction"
       "com.linkedin.hoptimator.jdbc.ddl.SqlCreateMaterializedView"
       "com.linkedin.hoptimator.jdbc.ddl.SqlCreateTable"
@@ -91,6 +92,7 @@ data: {
     # Each must accept arguments "(SqlParserPos pos, boolean replace)".
     # Example: "SqlCreateForeignSchema".
     createStatementParserMethods: [
+      "SqlCreateDatabase"
       "SqlCreateMaterializedView"
       "SqlCreateTrigger"
       "SqlCreateTable"

hoptimator-jdbc/src/main/codegen/includes/parserImpls.ftl

@@ -334,6 +334,20 @@ SqlCreate SqlCreateTrigger(Span s, boolean replace) :
     }
 }
 
+SqlCreate SqlCreateDatabase(Span s, boolean replace) :
+{
+    final boolean ifNotExists;
+    final SqlIdentifier id;
+    SqlNodeList optionList = null;
+}
+{
+    <DATABASE> ifNotExists = IfNotExistsOpt() id = CompoundIdentifier()
+    [ optionList = Options() ]
+    {
+        return new SqlCreateDatabase(s.end(this), replace, ifNotExists, id, optionList);
+    }
+}
+
 SqlCreate SqlCreateFunction(Span s, boolean replace) :
 {
     final boolean ifNotExists;

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

@@ -25,6 +25,7 @@
 import com.linkedin.hoptimator.UserJob;
 import com.linkedin.hoptimator.View;
 import com.linkedin.hoptimator.jdbc.ddl.HoptimatorDdlParserImpl;
+import com.linkedin.hoptimator.jdbc.ddl.SqlCreateDatabase;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateMaterializedView;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateTable;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateTrigger;
@@ -268,6 +269,19 @@ public void execute(SqlCreateTable create, CalcitePrepare.Context context) {
     logger.info("CREATE TABLE {} completed", create.name);
   }
 
+  /** 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;
+    try {
+      HoptimatorDdlUtils.processCreateDatabase(connection, create, mode);
+    } catch (SQLException | RuntimeException e) {
+      logger.info("Failed to deploy database {}", create.name);
+      throw new DdlException(create, e.getMessage(), e);
+    }
+    logger.info("CREATE DATABASE {} completed", create.name);
+  }
+
   /** Executes a {@code PAUSE TRIGGER} command. */
   public void execute(SqlPauseTrigger pause, CalcitePrepare.Context context) {
     updateTriggerPausedState(pause, pause.name, true);

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

@@ -21,10 +21,12 @@
 
 import com.google.common.collect.ImmutableList;
 import com.linkedin.hoptimator.Database;
+import com.linkedin.hoptimator.DatabaseDeployable;
 import com.linkedin.hoptimator.Deployer;
 import com.linkedin.hoptimator.MaterializedView;
 import com.linkedin.hoptimator.Pipeline;
 import com.linkedin.hoptimator.Source;
+import com.linkedin.hoptimator.jdbc.ddl.SqlCreateDatabase;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateMaterializedView;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateTable;
 import com.linkedin.hoptimator.util.DeploymentService;
@@ -621,6 +623,55 @@ public RexNode newColumnDefaultValue(RelOptTable table, int iColumn,
     }
   }
 
+  /**
+   * Shared implementation of the {@code CREATE DATABASE} pipeline for both real deployment
+   * and dry-run (SPECIFY) modes.
+   *
+   * @param conn   the JDBC connection
+   * @param create the parsed DDL node
+   * @param mode   whether to CREATE, UPDATE, or SPECIFY
+   * @return a SpecifyResult (specs are empty for CREATE/UPDATE, YAML for SPECIFY)
+   * @throws SQLException on validation or deployment errors
+   */
+  static SpecifyResult processCreateDatabase(HoptimatorConnection conn,
+      SqlCreateDatabase create, DdlMode mode) throws SQLException {
+    HoptimatorConnection.HoptimatorConnectionDualLogger logger = conn.getLogger(HoptimatorDdlUtils.class);
+
+    logger.info("Validating statement: {}", create);
+    ValidationService.validateOrThrow(create);
+
+    if (create.name.names.size() > 1) {
+      throw new SQLException("Database names cannot be compound identifiers.");
+    }
+    String name = create.name.names.get(0);
+
+    Map<String, String> dbOptions = options(create.options);
+    DatabaseDeployable database = new DatabaseDeployable(name, dbOptions);
+
+    Collection<Deployer> deployers = null;
+    try {
+      logger.info("Validating database {}", name);
+      ValidationService.validateOrThrow(database);
+      deployers = DeploymentService.deployers(database, conn);
+      ValidationService.validateOrThrow(deployers);
+
+      List<String> specs = mode.executeDeployers(deployers, conn);
+      if (mode.mutable()) {
+        logger.info("Deployed database {}", name);
+      } else {
+        DeploymentService.restore(deployers);
+      }
+      return new SpecifyResult(specs, null, Collections.singletonList(name));
+    } catch (SQLException | RuntimeException e) {
+      logger.info("Failed to deploy database {}", name);
+      if (deployers != null) {
+        DeploymentService.restore(deployers);
+        logger.info("Restored deployable resources for database {}", name);
+      }
+      throw e;
+    }
+  }
+
   /**
    * Returns the YAML specs that would be created for any supported SQL statement —
    * {@code CREATE TABLE}, {@code CREATE MATERIALIZED VIEW}, or {@code INSERT INTO}.
@@ -643,6 +694,10 @@ public RexNode newColumnDefaultValue(RelOptTable table, int iColumn,
   public static SpecifyResult specifyFromSql(String sql, HoptimatorConnection conn) throws SQLException {
     SqlNode sqlNode = HoptimatorDriver.parseQuery(conn, sql);
 
+    if (sqlNode instanceof SqlCreateDatabase) {
+      return processCreateDatabase(conn, (SqlCreateDatabase) sqlNode, DdlMode.SPECIFY);
+    }
+
     if (sqlNode instanceof SqlCreateTable) {
       return processCreateTable(conn.createPrepareContext(), conn, (SqlCreateTable) sqlNode, DdlMode.SPECIFY);
     }

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/ddl/HoptimatorDdlParserImpl.java

@@ -7,6 +7,7 @@
 import org.apache.calcite.sql.SqlTruncate;
 import org.apache.calcite.sql.ddl.SqlCreateTableLike;
 import org.apache.calcite.sql.ddl.SqlDdlNodes;
+import com.linkedin.hoptimator.jdbc.ddl.SqlCreateDatabase;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateFunction;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateMaterializedView;
 import com.linkedin.hoptimator.jdbc.ddl.SqlCreateTable;
@@ -6730,6 +6731,24 @@ final public SqlCreate SqlCreateTrigger(Span s, boolean replace) throws ParseExc
     throw new Error("Missing return statement in function");
   }
 
+  final public SqlCreate SqlCreateDatabase(Span s, boolean replace) throws ParseException {
+    final boolean ifNotExists;
+    final SqlIdentifier id;
+    SqlNodeList optionList = null;
+    jj_consume_token(DATABASE);
+    ifNotExists = IfNotExistsOpt();
+    id = CompoundIdentifier();
+    switch ((jj_ntk==-1)?jj_ntk():jj_ntk) {
+    case WITH:
+      optionList = Options();
+      break;
+    default:
+      ;
+    }
+        {if (true) return new SqlCreateDatabase(s.end(this), replace, ifNotExists, id, optionList);}
+    throw new Error("Missing return statement in function");
+  }
+
   final public SqlCreate SqlCreateFunction(Span s, boolean replace) throws ParseException {
     final boolean ifNotExists;
     final SqlIdentifier id;
@@ -24461,6 +24480,9 @@ final public SqlCreate SqlCreate() throws ParseException {
       ;
     }
     switch ((jj_ntk==-1)?jj_ntk():jj_ntk) {
+    case DATABASE:
+      create = SqlCreateDatabase(s, replace);
+      break;
     case MATERIALIZED:
       create = SqlCreateMaterializedView(s, replace);
       break;

hoptimator-jdbc/src/main/java/com/linkedin/hoptimator/jdbc/ddl/SqlCreateDatabase.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to you under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.linkedin.hoptimator.jdbc.ddl;
+
+import org.apache.calcite.sql.SqlCreate;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlSpecialOperator;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.util.ImmutableNullableList;
+
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Parse tree for {@code CREATE DATABASE} statement.
+ */
+public class SqlCreateDatabase extends SqlCreate {
+  public final SqlIdentifier name;
+  public final @Nullable SqlNodeList options;
+
+  private static final SqlOperator OPERATOR =
+      new SqlSpecialOperator("CREATE DATABASE", SqlKind.OTHER_DDL);
+
+  /** Creates a SqlCreateDatabase. */
+  protected SqlCreateDatabase(SqlParserPos pos, boolean replace, boolean ifNotExists,
+      SqlIdentifier name, @Nullable SqlNodeList options) {
+    super(OPERATOR, pos, replace, ifNotExists);
+    this.name = requireNonNull(name, "name");
+    this.options = options;
+  }
+
+  @SuppressWarnings("nullness")
+  @Override public List<SqlNode> getOperandList() {
+    return ImmutableNullableList.of(name, options);
+  }
+
+  @Override public void unparse(SqlWriter writer, int leftPrec, int rightPrec) {
+    writer.keyword("CREATE");
+    writer.keyword("DATABASE");
+    if (ifNotExists) {
+      writer.keyword("IF NOT EXISTS");
+    }
+    name.unparse(writer, leftPrec, rightPrec);
+    if (options != null) {
+      writer.keyword("WITH");
+      SqlWriter.Frame frame = writer.startList("(", ")");
+      for (SqlNode c : options) {
+        writer.sep(",");
+        c.unparse(writer, 0, 0);
+      }
+      writer.endList(frame);
+    }
+  }
+}