content: /*
   * 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 org.apache.calcite.server;
  
  import org.apache.calcite.adapter.java.JavaTypeFactory;
  import org.apache.calcite.adapter.jdbc.JdbcSchema;
  import org.apache.calcite.avatica.AvaticaUtils;
  import org.apache.calcite.jdbc.CalcitePrepare;
  import org.apache.calcite.jdbc.CalciteSchema;
  import org.apache.calcite.jdbc.ContextSqlValidator;
  import org.apache.calcite.linq4j.Ord;
  import org.apache.calcite.materialize.MaterializationKey;
  import org.apache.calcite.materialize.MaterializationService;
  import org.apache.calcite.model.JsonSchema;
  import org.apache.calcite.plan.RelOptTable;
  import org.apache.calcite.rel.RelRoot;
  import org.apache.calcite.rel.type.RelDataType;
  import org.apache.calcite.rel.type.RelDataTypeFactory;
  import org.apache.calcite.rel.type.RelDataTypeField;
  import org.apache.calcite.rel.type.RelDataTypeImpl;
  import org.apache.calcite.rex.RexNode;
  import org.apache.calcite.schema.ColumnStrategy;
  import org.apache.calcite.schema.Function;
  import org.apache.calcite.schema.Schema;
  import org.apache.calcite.schema.SchemaFactory;
  import org.apache.calcite.schema.SchemaPlus;
  import org.apache.calcite.schema.Table;
  import org.apache.calcite.schema.TranslatableTable;
  import org.apache.calcite.schema.Wrapper;
  import org.apache.calcite.schema.impl.AbstractSchema;
  import org.apache.calcite.schema.impl.ViewTable;
  import org.apache.calcite.schema.impl.ViewTableMacro;
  import org.apache.calcite.sql.SqlCall;
  import org.apache.calcite.sql.SqlDataTypeSpec;
  import org.apache.calcite.sql.SqlIdentifier;
  import org.apache.calcite.sql.SqlKind;
  import org.apache.calcite.sql.SqlLiteral;
  import org.apache.calcite.sql.SqlNode;
  import org.apache.calcite.sql.SqlNodeList;
  import org.apache.calcite.sql.SqlSelect;
  import org.apache.calcite.sql.SqlUtil;
  import org.apache.calcite.sql.SqlWriterConfig;
  import org.apache.calcite.sql.ddl.SqlAttributeDefinition;
  import org.apache.calcite.sql.ddl.SqlColumnDeclaration;
  import org.apache.calcite.sql.ddl.SqlCreateForeignSchema;
  import org.apache.calcite.sql.ddl.SqlCreateFunction;
  import org.apache.calcite.sql.ddl.SqlCreateMaterializedView;
  import org.apache.calcite.sql.ddl.SqlCreateSchema;
  import org.apache.calcite.sql.ddl.SqlCreateTable;
  import org.apache.calcite.sql.ddl.SqlCreateTableLike;
  import org.apache.calcite.sql.ddl.SqlCreateType;
  import org.apache.calcite.sql.ddl.SqlCreateView;
  import org.apache.calcite.sql.ddl.SqlDropObject;
  import org.apache.calcite.sql.ddl.SqlDropSchema;
  import org.apache.calcite.sql.ddl.SqlTruncateTable;
  import org.apache.calcite.sql.dialect.CalciteSqlDialect;
  import org.apache.calcite.sql.fun.SqlStdOperatorTable;
  import org.apache.calcite.sql.parser.SqlAbstractParserImpl;
  import org.apache.calcite.sql.parser.SqlParseException;
  import org.apache.calcite.sql.parser.SqlParserImplFactory;
  import org.apache.calcite.sql.parser.SqlParserPos;
  import org.apache.calcite.sql.parser.ddl.SqlDdlParserImpl;
  import org.apache.calcite.sql.pretty.SqlPrettyWriter;
  import org.apache.calcite.sql.validate.SqlValidator;
  import org.apache.calcite.sql2rel.InitializerContext;
  import org.apache.calcite.sql2rel.InitializerExpressionFactory;
  import org.apache.calcite.sql2rel.NullInitializerExpressionFactory;
  import org.apache.calcite.tools.FrameworkConfig;
  import org.apache.calcite.tools.Frameworks;
  import org.apache.calcite.tools.Planner;
  import org.apache.calcite.tools.RelConversionException;
  import org.apache.calcite.tools.ValidationException;
  import org.apache.calcite.util.NlsString;
  import org.apache.calcite.util.Pair;
  import org.apache.calcite.util.Util;
  
  import com.google.common.collect.ImmutableList;
  
  import org.checkerframework.checker.nullness.qual.Nullable;
  
  import java.io.Reader;
  import java.sql.PreparedStatement;
  import java.sql.SQLException;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.LinkedHashMap;
  import java.util.List;
  import java.util.Locale;
  import java.util.Map;
  import java.util.Set;
  
  import static com.google.common.base.Preconditions.checkArgument;
  
  import static org.apache.calcite.util.Static.RESOURCE;
  
  import static java.util.Objects.requireNonNull;
  
  /** Executes DDL commands.
   *
   * <p>Given a DDL command that is a sub-class of {@link SqlNode}, dispatches
   * the command to an appropriate {@code execute} method. For example,
   * "CREATE TABLE" ({@link SqlCreateTable}) is dispatched to
   * {@link #execute(SqlCreateTable, CalcitePrepare.Context)}. */
  public class ServerDdlExecutor extends DdlExecutorImpl {
    /** Singleton instance. */
    public static final ServerDdlExecutor INSTANCE = new ServerDdlExecutor();
  
    /** Parser factory. */
    @SuppressWarnings("unused") // used via reflection
    public static final SqlParserImplFactory PARSER_FACTORY =
        new SqlParserImplFactory() {
          @Override public SqlAbstractParserImpl getParser(Reader stream) {
            return SqlDdlParserImpl.FACTORY.getParser(stream);
          }
  
          @Override public DdlExecutor getDdlExecutor() {
            return ServerDdlExecutor.INSTANCE;
          }
        };
  
    /** Creates a ServerDdlExecutor.
     * Protected only to allow sub-classing;
     * use {@link #INSTANCE} where possible. */
    protected ServerDdlExecutor() {
    }
  
    /** Returns the schema in which to create an object;
     * the left part is null if the schema does not exist. */
    static Pair<@Nullable CalciteSchema, String> schema(
        CalcitePrepare.Context context, boolean mutable, SqlIdentifier id) {
      final String name;
      final List<String> path;
      if (id.isSimple()) {
        path = context.getDefaultSchemaPath();
        name = id.getSimple();
      } else {
        path = Util.skipLast(id.names);
        name = Util.last(id.names);
      }
      CalciteSchema schema =
          mutable ? context.getMutableRootSchema()
              : context.getRootSchema();
      for (String p : path) {
        @Nullable CalciteSchema subSchema = schema.getSubSchema(p, true);
        if (subSchema == null) {
          return Pair.of(null, name);
        }
        schema = subSchema;
      }
      return Pair.of(schema, name);
    }
  
    /**
     * Returns the SqlValidator with the given {@code context} schema
     * and type factory.
     */
    static SqlValidator validator(CalcitePrepare.Context context,
        boolean mutable) {
      return new ContextSqlValidator(context, mutable);
    }
  
    /** Wraps a query to rename its columns. Used by CREATE VIEW and CREATE
     * MATERIALIZED VIEW. */
    static SqlNode renameColumns(@Nullable SqlNodeList columnList,
        SqlNode query) {
      if (columnList == null) {
        return query;
      }
      final SqlParserPos p = query.getParserPosition();
      final SqlNodeList selectList = SqlNodeList.SINGLETON_STAR;
      final SqlCall from =
          SqlStdOperatorTable.AS.createCall(p,
              ImmutableList.<SqlNode>builder()
                  .add(query)
                  .add(new SqlIdentifier("_", p))
                  .addAll(columnList)
                  .build());
      return new SqlSelect(p, null, selectList, from, null, null, null, null,
          null, null, null, null, null);
    }
  
    /** Erase the table date that calcite-sever created. */
    static void erase(SqlIdentifier name, CalcitePrepare.Context context) {
      // Directly clearing data is more efficient than executing SQL
      final Pair<@Nullable CalciteSchema, String> pair =
          schema(context, true, name);
      final CalciteSchema calciteSchema = requireNonNull(pair.left);
      final String tblName = pair.right;
      final CalciteSchema.TableEntry tableEntry =
          calciteSchema.getTable(tblName, context.config().caseSensitive());
      final Table table = requireNonNull(tableEntry, "tableEntry").getTable();
      if (table instanceof MutableArrayTable) {
        MutableArrayTable mutableArrayTable = (MutableArrayTable) table;
        mutableArrayTable.rows.clear();
      } else {
        // Not calcite-server created, so not support truncate.
        throw new UnsupportedOperationException("Only MutableArrayTable support truncate");
      }
    }
  
    /** Populates the table called {@code name} by executing {@code query}. */
    static void populate(SqlIdentifier name, SqlNode query,
        CalcitePrepare.Context context) {
      // Generate, prepare and execute an "INSERT INTO table query" statement.
      // (It's a bit inefficient that we convert from SqlNode to SQL and back
      // again.)
      final FrameworkConfig config = Frameworks.newConfigBuilder()
          .defaultSchema(context.getRootSchema().plus())
          .build();
      final Planner planner = Frameworks.getPlanner(config);
      try {
        final StringBuilder buf = new StringBuilder();
        final SqlWriterConfig writerConfig =
            SqlPrettyWriter.config().withAlwaysUseParentheses(false);
        final SqlPrettyWriter w = new SqlPrettyWriter(writerConfig, buf);
        buf.append("INSERT INTO ");
        name.unparse(w, 0, 0);
        buf.append(' ');
        query.unparse(w, 0, 0);
        final String sql = buf.toString();
        final SqlNode query1 = planner.parse(sql);
        final SqlNode query2 = planner.validate(query1);
        final RelRoot r = planner.rel(query2);
        final PreparedStatement prepare =
            context.getRelRunner().prepareStatement(r.rel);
        int rowCount = prepare.executeUpdate();
        Util.discard(rowCount);
        prepare.close();
      } catch (SqlParseException | ValidationException
          | RelConversionException | SQLException e) {
        throw Util.throwAsRuntime(e);
      }
    }
  
    /** Returns the value of a literal, converting
     * {@link NlsString} into String. */
    @SuppressWarnings("rawtypes")
    static @Nullable Comparable value(SqlNode node) {
      final Comparable v = SqlLiteral.value(node);
      return v instanceof NlsString ? ((NlsString) v).getValue() : v;
    }
  
    /** Executes a {@code CREATE FOREIGN SCHEMA} command. */
    public void execute(SqlCreateForeignSchema create,
        CalcitePrepare.Context context) {
      final Pair<@Nullable CalciteSchema, String> pair =
          schema(context, true, create.name);
      requireNonNull(pair.left); // TODO: should not assume parent schema exists
      if (pair.left.plus().subSchemas().get(pair.right) != null) {
        if (!create.getReplace() && !create.ifNotExists) {
          throw SqlUtil.newContextException(create.name.getParserPosition(),
              RESOURCE.schemaExists(pair.right));
        }
      }
      final Schema subSchema;
      final String libraryName;
      if (create.type != null) {
        checkArgument(create.library == null);
        final String typeName = (String) requireNonNull(value(create.type));
        final JsonSchema.Type type =
            Util.enumVal(JsonSchema.Type.class,
                typeName.toUpperCase(Locale.ROOT));
        if (type != null) {
          switch (type) {
          case JDBC:
            libraryName = JdbcSchema.Factory.class.getName();
            break;
          default:
            libraryName = null;
          }
        } else {
          libraryName = null;
        }
        if (libraryName == null) {
          throw SqlUtil.newContextException(create.type.getParserPosition(),
              RESOURCE.schemaInvalidType(typeName,
                  Arrays.toString(JsonSchema.Type.values())));
        }
      } else {
        libraryName =
            requireNonNull((String) value(requireNonNull(create.library)));
      }
      final SchemaFactory schemaFactory =
          AvaticaUtils.instantiatePlugin(SchemaFactory.class, libraryName);
      final Map<String, Object> operandMap = new LinkedHashMap<>();
      for (Pair<SqlIdentifier, SqlNode> option : create.options()) {
        operandMap.put(option.left.getSimple(),
            requireNonNull(value(option.right)));
      }
      subSchema =
          schemaFactory.create(pair.left.plus(), pair.right, operandMap);
      pair.left.add(pair.right, subSchema);
    }
  
    /** Executes a {@code CREATE FUNCTION} command. */
    public void execute(SqlCreateFunction create,
        CalcitePrepare.Context context) {
      throw new UnsupportedOperationException("CREATE FUNCTION is not supported");
    }
  
    /** Executes {@code DROP FUNCTION}, {@code DROP TABLE},
     * {@code DROP MATERIALIZED VIEW}, {@code DROP TYPE},
     * {@code DROP VIEW} commands. */
    public void execute(SqlDropObject drop,
        CalcitePrepare.Context context) {
      final Pair<@Nullable CalciteSchema, String> pair =
          schema(context, false, drop.name);
      final @Nullable CalciteSchema schema =
          pair.left; // null if schema does not exist
      final String objectName = pair.right;
  
      boolean existed;
      switch (drop.getKind()) {
      case DROP_TABLE:
      case DROP_MATERIALIZED_VIEW:
        Table materializedView =
            schema != null
                && drop.getKind() == SqlKind.DROP_MATERIALIZED_VIEW
                ? schema.plus().tables().get(objectName)
                : null;
  
        existed = schema != null && schema.removeTable(objectName);
        if (existed) {
          if (materializedView instanceof Wrapper) {
            ((Wrapper) materializedView).maybeUnwrap(MaterializationKey.class)
                .ifPresent(materializationKey ->
                    MaterializationService.instance()
                        .removeMaterialization(materializationKey));
          }
        } else if (!drop.ifExists) {
          throw SqlUtil.newContextException(drop.name.getParserPosition(),
              RESOURCE.tableNotFound(objectName));
        }
        break;
      case DROP_VIEW:
        // Not quite right: removes any other functions with the same name
        existed = schema != null && schema.removeFunction(objectName);
        if (!existed && !drop.ifExists) {
          throw SqlUtil.newContextException(drop.name.getParserPosition(),
              RESOURCE.viewNotFound(objectName));
        }
        break;
      case DROP_TYPE:
        existed = schema != null && schema.removeType(objectName);
        if (!existed && !drop.ifExists) {
          throw SqlUtil.newContextException(drop.name.getParserPosition(),
              RESOURCE.typeNotFound(objectName));
        }
        break;
      case DROP_FUNCTION:
        existed = schema != null && schema.removeFunction(objectName);
        if (!existed && !drop.ifExists) {
          throw SqlUtil.newContextException(drop.name.getParserPosition(),
              RESOURCE.functionNotFound(objectName));
        }
        break;
      case OTHER_DDL:
      default:
        throw new AssertionError(drop.getKind());
      }
    }
  
    /**
     * Executes a {@code TRUNCATE TABLE} command.
     */
    public void execute(SqlTruncateTable truncate,
        CalcitePrepare.Context context) {
      final Pair<@Nullable CalciteSchema, String> pair =
          schema(context, true, truncate.name);
      if (pair.left == null
          || pair.left.plus().tables().get(pair.right) == null) {
        throw SqlUtil.newContextException(truncate.name.getParserPosition(),
            RESOURCE.tableNotFound(pair.right));
      }
  
      if (!truncate.continueIdentify) {
        // Calcite not support RESTART IDENTIFY
        throw new UnsupportedOperationException("RESTART IDENTIFY is not supported");
      }
  
      erase(truncate.name, context);
    }
  
    /** Executes a {@code CREATE MATERIALIZED VIEW} command. */
    public void execute(SqlCreateMaterializedView create,
        CalcitePrepare.Context context) {
      final Pair<@Nullable CalciteSchema, String> pair =
          schema(context, true, create.name);
      if (pair.left != null
          && pair.left.plus().tables().get(pair.right) != null) {
        // Materialized view exists.
        if (!create.ifNotExists) {
          // They did not specify IF NOT EXISTS, so give error.
          throw SqlUtil.newContextException(create.name.getParserPosition(),
              RESOURCE.tableExists(pair.right));
        }
        return;
      }
      final SqlNode q = renameColumns(create.columnList, create.query);
      final String sql = q.toSqlString(CalciteSqlDialect.DEFAULT).getSql();
      requireNonNull(pair.left); // TODO: should not assume parent schema exists
      final List<String> schemaPath = pair.left.path(null);
      final ViewTableMacro viewTableMacro =
          ViewTable.viewMacro(pair.left.plus(), sql, schemaPath,
              context.getObjectPath(), false);
      final TranslatableTable x = viewTableMacro.apply(ImmutableList.of());
      final RelDataType rowType = x.getRowType(context.getTypeFactory());
  
      // Table does not exist. Create it.
      final MaterializedViewTable table =
          new MaterializedViewTable(pair.right, RelDataTypeImpl.proto(rowType));
      pair.left.add(pair.right, table);
      populate(create.name, create.query, context);
      table.key =
          MaterializationService.instance().defineMaterialization(pair.left, null,
              sql, schemaPath, pair.right, true, true);
    }
  
    /** Executes a {@code CREATE SCHEMA} command. */
    public void execute(SqlCreateSchema create,
        CalcitePrepare.Context context) {
      final Pair<@Nullable CalciteSchema, String> pair =
          schema(context, true, create.name);
      requireNonNull(pair.left); // TODO: should not assume parent schema exists
      if (pair.left.plus().subSchemas().get(pair.right) != null) {
        if (create.ifNotExists) {
          return;
        }
        if (!create.getReplace()) {
          throw SqlUtil.newContextException(create.name.getParserPosition(),
              RESOURCE.schemaExists(pair.right));
        }
      }
      final Schema subSchema = new AbstractSchema();
      pair.left.add(pair.right, subSchema);
    }
  
    /** Executes a {@code DROP SCHEMA} command. */
    public void execute(SqlDropSchema drop,
        CalcitePrepare.Context context) {
      final Pair<@Nullable CalciteSchema, String> pair =
          schema(context, false, drop.name);
      final String name = pair.right;
      final boolean existed = pair.left != null
          && pair.left.removeSubSchema(name);
      if (!existed && !drop.ifExists) {
        throw SqlUtil.newContextException(drop.name.getParserPosition(),
            RESOURCE.schemaNotFound(name));
      }
    }
  
    /** Executes a {@code CREATE TABLE} command. */
    public void execute(SqlCreateTable create,
        CalcitePrepare.Context context) {
      final Pair<@Nullable CalciteSchema, String> pair =
          schema(context, true, create.name);
      requireNonNull(pair.left); // TODO: should not assume parent schema exists
      final JavaTypeFactory typeFactory = context.getTypeFactory();
      final RelDataType queryRowType;
      if (create.query != null) {
        // A bit of a hack: pretend it's a view, to get its row type
        final String sql =
            create.query.toSqlString(CalciteSqlDialect.DEFAULT).getSql();
        final ViewTableMacro viewTableMacro =
            ViewTable.viewMacro(pair.left.plus(), sql, pair.left.path(null),
                context.getObjectPath(), false);
        final TranslatableTable x = viewTableMacro.apply(ImmutableList.of());
        queryRowType = x.getRowType(typeFactory);
  
        if (create.columnList != null
            && queryRowType.getFieldCount() != create.columnList.size()) {
          throw SqlUtil.newContextException(
              create.columnList.getParserPosition(),
              RESOURCE.columnCountMismatch());
        }
      } else {
        queryRowType = null;
      }
      final List<SqlNode> columnList;
      if (create.columnList != null) {
        columnList = create.columnList;
      } else {
        if (queryRowType == null) {
          // "CREATE TABLE t" is invalid; because there is no "AS query" we need
          // a list of column names and types, "CREATE TABLE t (INT c)".
          throw SqlUtil.newContextException(create.name.getParserPosition(),
              RESOURCE.createTableRequiresColumnList());
        }
        columnList = new ArrayList<>();
        for (String name : queryRowType.getFieldNames()) {
          columnList.add(new SqlIdentifier(name, SqlParserPos.ZERO));
        }
      }
      final ImmutableList.Builder<ColumnDef> b = ImmutableList.builder();
      final RelDataTypeFactory.Builder builder = typeFactory.builder();
      final RelDataTypeFactory.Builder storedBuilder = typeFactory.builder();
      // REVIEW 2019-08-19 Danny Chan: Should we implement the
      // #validate(SqlValidator) to get the SqlValidator instance?
      final SqlValidator validator = validator(context, true);
      for (Ord<SqlNode> c : Ord.zip(columnList)) {
        if (c.e instanceof SqlColumnDeclaration) {
          final SqlColumnDeclaration d = (SqlColumnDeclaration) c.e;
          final RelDataType type = d.dataType.deriveType(validator, true);
          builder.add(d.name.getSimple(), type);
          if (d.strategy != ColumnStrategy.VIRTUAL) {
            storedBuilder.add(d.name.getSimple(), type);
          }
          b.add(ColumnDef.of(d.expression, type, d.strategy));
        } else if (c.e instanceof SqlIdentifier) {
          final SqlIdentifier id = (SqlIdentifier) c.e;
          if (queryRowType == null) {
            throw SqlUtil.newContextException(id.getParserPosition(),
                RESOURCE.createTableRequiresColumnTypes(id.getSimple()));
          }
          final RelDataTypeField f = queryRowType.getFieldList().get(c.i);
          final ColumnStrategy strategy = f.getType().isNullable()
              ? ColumnStrategy.NULLABLE
              : ColumnStrategy.NOT_NULLABLE;
          b.add(ColumnDef.of(c.e, f.getType(), strategy));
          builder.add(id.getSimple(), f.getType());
          storedBuilder.add(id.getSimple(), f.getType());
        } else {
          throw new AssertionError(c.e.getClass());
        }
      }
      final RelDataType rowType = builder.build();
      final RelDataType storedRowType = storedBuilder.build();
      final List<ColumnDef> columns = b.build();
      final InitializerExpressionFactory ief =
          new NullInitializerExpressionFactory() {
            @Override public ColumnStrategy generationStrategy(RelOptTable table,
                int iColumn) {
              return columns.get(iColumn).strategy;
            }
  
            @Override public RexNode newColumnDefaultValue(RelOptTable table,
                int iColumn, InitializerContext context) {
              final ColumnDef c = columns.get(iColumn);
              if (c.expr != null) {
                // REVIEW Danny 2019-10-09: Should we support validation for DDL nodes?
                final SqlNode validated = context.validateExpression(storedRowType, c.expr);
                // The explicit specified type should have the same nullability
                // with the column expression inferred type,
                // actually they should be exactly the same.
                return context.convertExpression(validated);
              }
              return super.newColumnDefaultValue(table, iColumn, context);
            }
          };
      if (pair.left.plus().tables().get(pair.right) != null) {
        // Table exists.
        if (create.ifNotExists) {
          return;
        }
        if (!create.getReplace()) {
          // They did not specify IF NOT EXISTS, so give error.
          throw SqlUtil.newContextException(create.name.getParserPosition(),
              RESOURCE.tableExists(pair.right));
        }
      }
      // Table does not exist. Create it.
      pair.left.add(pair.right,
          new MutableArrayTable(pair.right,
              RelDataTypeImpl.proto(storedRowType),
              RelDataTypeImpl.proto(rowType), ief));
      if (create.query != null) {
        populate(create.name, create.query, context);
      }
    }
  
    /** Executes a {@code CREATE TABLE LIKE} command. */
    public void execute(SqlCreateTableLike create,
        CalcitePrepare.Context context) {
      final Pair<@Nullable CalciteSchema, String> pair =
          schema(context, true, create.name);
      requireNonNull(pair.left); // TODO: should not assume parent schema exists
      if (pair.left.plus().tables().get(pair.right) != null) {
        // Table exists.
        if (create.ifNotExists) {
          return;
        }
        if (!create.getReplace()) {
          // They did not specify IF NOT EXISTS, so give error.
          throw SqlUtil.newContextException(create.name.getParserPosition(),
              RESOURCE.tableExists(pair.right));
        }
      }
  
      final Pair<@Nullable CalciteSchema, String> sourceTablePair =
          schema(context, true, create.sourceTable);
      final CalciteSchema schema =
          // TODO: should not assume parent schema exists
          requireNonNull(sourceTablePair.left);
      final String tableName = sourceTablePair.right;
      final CalciteSchema.TableEntry tableEntry =
          schema.getTable(tableName, context.config().caseSensitive());
      final Table table = requireNonNull(tableEntry, "tableEntry").getTable();
  
      InitializerExpressionFactory ief = NullInitializerExpressionFactory.INSTANCE;
      if (table instanceof Wrapper) {
        final InitializerExpressionFactory sourceIef =
            ((Wrapper) table).unwrap(InitializerExpressionFactory.class);
        if (sourceIef != null) {
          final Set<SqlCreateTableLike.LikeOption> optionSet = create.options();
          final boolean includingGenerated =
              optionSet.contains(SqlCreateTableLike.LikeOption.GENERATED)
                  || optionSet.contains(SqlCreateTableLike.LikeOption.ALL);
          final boolean includingDefaults =
              optionSet.contains(SqlCreateTableLike.LikeOption.DEFAULTS)
                  || optionSet.contains(SqlCreateTableLike.LikeOption.ALL);
  
          // initializes columns based on the source table InitializerExpressionFactory
          // and like options.
          ief =
              new CopiedTableInitializerExpressionFactory(
                  includingGenerated, includingDefaults, sourceIef);
        }
      }
  
      final JavaTypeFactory typeFactory = context.getTypeFactory();
      final RelDataType rowType = table.getRowType(typeFactory);
      // Table does not exist. Create it.
      pair.left.add(pair.right,
          new MutableArrayTable(pair.right,
              RelDataTypeImpl.proto(rowType),
              RelDataTypeImpl.proto(rowType), ief));
    }
  
    /** Executes a {@code CREATE TYPE} command. */
    public void execute(SqlCreateType create,
        CalcitePrepare.Context context) {
      final Pair<@Nullable CalciteSchema, String> pair =
          schema(context, true, create.name);
      requireNonNull(pair.left); // TODO: should not assume parent schema exists
      final SqlValidator validator = validator(context, false);
      pair.left.add(pair.right, typeFactory -> {
        if (create.dataType != null) {
          return create.dataType.deriveType(validator);
        } else {
          final RelDataTypeFactory.Builder builder = typeFactory.builder();
          if (create.attributeDefs != null) {
            for (SqlNode def : create.attributeDefs) {
              final SqlAttributeDefinition attributeDef =
                  (SqlAttributeDefinition) def;
              final SqlDataTypeSpec typeSpec = attributeDef.dataType;
              final RelDataType type = typeSpec.deriveType(validator);
              builder.add(attributeDef.name.getSimple(), type);
            }
          }
          return builder.build();
        }
      });
    }
  
    /** Executes a {@code CREATE VIEW} command. */
    public void execute(SqlCreateView create,
        CalcitePrepare.Context context) {
      final Pair<@Nullable CalciteSchema, String> pair =
          schema(context, true, create.name);
      requireNonNull(pair.left); // TODO: should not assume parent schema exists
      final SchemaPlus schemaPlus = pair.left.plus();
      for (Function function : schemaPlus.getFunctions(pair.right)) {
        if (function.getParameters().isEmpty()) {
          if (!create.getReplace()) {
            throw SqlUtil.newContextException(create.name.getParserPosition(),
                RESOURCE.viewExists(pair.right));
          }
          pair.left.removeFunction(pair.right);
        }
      }
      final SqlNode q = renameColumns(create.columnList, create.query);
      final String sql = q.toSqlString(CalciteSqlDialect.DEFAULT).getSql();
      final ViewTableMacro viewTableMacro =
          ViewTable.viewMacro(schemaPlus, sql, pair.left.path(null),
              context.getObjectPath(), false);
      final TranslatableTable x = viewTableMacro.apply(ImmutableList.of());
      Util.discard(x);
      schemaPlus.add(pair.right, viewTableMacro);
    }
  
    /**
     * Initializes columns based on the source {@link InitializerExpressionFactory}
     * and like options.
     */
    private static class CopiedTableInitializerExpressionFactory
        extends NullInitializerExpressionFactory {
  
      private final boolean includingGenerated;
      private final boolean includingDefaults;
      private final InitializerExpressionFactory sourceIef;
  
      CopiedTableInitializerExpressionFactory(
          boolean includingGenerated,
          boolean includingDefaults,
          InitializerExpressionFactory sourceIef) {
        this.includingGenerated = includingGenerated;
        this.includingDefaults = includingDefaults;
        this.sourceIef = sourceIef;
      }
  
      @Override public ColumnStrategy generationStrategy(
          RelOptTable table, int iColumn) {
        final ColumnStrategy sourceStrategy = sourceIef.generationStrategy(table, iColumn);
        if (includingGenerated
            && (sourceStrategy == ColumnStrategy.STORED
            || sourceStrategy == ColumnStrategy.VIRTUAL)) {
          return sourceStrategy;
        }
        if (includingDefaults && sourceStrategy == ColumnStrategy.DEFAULT) {
          return ColumnStrategy.DEFAULT;
        }
  
        return super.generationStrategy(table, iColumn);
      }
  
      @Override public RexNode newColumnDefaultValue(
          RelOptTable table, int iColumn, InitializerContext context) {
        if (includingDefaults || includingGenerated) {
          return sourceIef.newColumnDefaultValue(table, iColumn, context);
        } else {
          return super.newColumnDefaultValue(table, iColumn, context);
        }
      }
    }
  
    /** Column definition. */
    private static class ColumnDef {
      final @Nullable SqlNode expr;
      final RelDataType type;
      final ColumnStrategy strategy;
  
      private ColumnDef(@Nullable SqlNode expr, RelDataType type,
          ColumnStrategy strategy) {
        this.expr = expr;
        this.type = type;
        this.strategy = requireNonNull(strategy, "strategy");
        checkArgument(
            strategy == ColumnStrategy.NULLABLE
                || strategy == ColumnStrategy.NOT_NULLABLE
                || expr != null);
      }
  
      static ColumnDef of(@Nullable SqlNode expr, RelDataType type,
          ColumnStrategy strategy) {
        return new ColumnDef(expr, type, strategy);
      }
    }
  }
  