diff --git a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/MSQTaskSqlEngine.java b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/MSQTaskSqlEngine.java
index cb331760ca34..2471d8c51b49 100644
--- a/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/MSQTaskSqlEngine.java
+++ b/extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/sql/MSQTaskSqlEngine.java
@@ -155,7 +155,8 @@ public OverlordClient overlordClient()
   public QueryMaker buildQueryMakerForInsert(
       final String targetDataSource,
       final RelRoot relRoot,
-      final PlannerContext plannerContext
+      final PlannerContext plannerContext,
+      final RelDataType targetType
   )
   {
     validateInsert(relRoot.rel, relRoot.fields, plannerContext);
diff --git a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java
index 0146fcf9bd8b..150e0b91de34 100644
--- a/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java
+++ b/extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/test/MSQTestBase.java
@@ -160,6 +160,9 @@
 import org.apache.druid.sql.calcite.BaseCalciteQueryTest;
 import org.apache.druid.sql.calcite.external.ExternalDataSource;
 import org.apache.druid.sql.calcite.external.ExternalOperatorConversion;
+import org.apache.druid.sql.calcite.external.HttpOperatorConversion;
+import org.apache.druid.sql.calcite.external.InlineOperatorConversion;
+import org.apache.druid.sql.calcite.external.LocalOperatorConversion;
 import org.apache.druid.sql.calcite.planner.CalciteRulesManager;
 import org.apache.druid.sql.calcite.planner.CatalogResolver;
 import org.apache.druid.sql.calcite.planner.PlannerConfig;
@@ -347,6 +350,9 @@ public void configure(Binder binder)
             binder.install(new NestedDataModule());
             NestedDataModule.registerHandlersAndSerde();
             SqlBindings.addOperatorConversion(binder, ExternalOperatorConversion.class);
+            SqlBindings.addOperatorConversion(binder, HttpOperatorConversion.class);
+            SqlBindings.addOperatorConversion(binder, InlineOperatorConversion.class);
+            SqlBindings.addOperatorConversion(binder, LocalOperatorConversion.class);
           }
 
           @Override
diff --git a/server/src/main/java/org/apache/druid/catalog/model/table/DatasourceDefn.java b/server/src/main/java/org/apache/druid/catalog/model/table/DatasourceDefn.java
index aa50dd3498e4..ef3f34ec7a7a 100644
--- a/server/src/main/java/org/apache/druid/catalog/model/table/DatasourceDefn.java
+++ b/server/src/main/java/org/apache/druid/catalog/model/table/DatasourceDefn.java
@@ -84,6 +84,7 @@ public void validate(Object value, ObjectMapper jsonMapper)
     {
       String gran = decode(value, jsonMapper);
       if (Strings.isNullOrEmpty(gran)) {
+        // TODO: allow null granularity
         throw new IAE("Segment granularity is required.");
       }
       CatalogUtils.validateGranularity(gran);
diff --git a/sql/src/main/codegen/includes/common.ftl b/sql/src/main/codegen/includes/common.ftl
index 59915bf09ed4..7b86f6e648e7 100644
--- a/sql/src/main/codegen/includes/common.ftl
+++ b/sql/src/main/codegen/includes/common.ftl
@@ -18,58 +18,51 @@
  */
 
 // Using fully qualified name for Pair class, since Calcite also has a same class name being used in the Parser.jj
-org.apache.druid.java.util.common.Pair<Granularity, String> PartitionGranularity() :
+SqlNode PartitionGranularity() :
 {
   SqlNode e;
-  Granularity granularity;
-  String unparseString;
+  SqlNode result;
 }
 {
   (
     <HOUR>
     {
-      granularity = Granularities.HOUR;
-      unparseString = "HOUR";
+      result = SqlLiteral.createSymbol(DruidSqlParserUtils.GranularityGrain.HOUR_GRAIN, getPos());
     }
   |
     <DAY>
     {
-      granularity = Granularities.DAY;
-      unparseString = "DAY";
+      result = SqlLiteral.createSymbol(DruidSqlParserUtils.GranularityGrain.DAY_GRAIN, getPos());
     }
   |
     <MONTH>
     {
-      granularity = Granularities.MONTH;
-      unparseString = "MONTH";
+      result = SqlLiteral.createSymbol(DruidSqlParserUtils.GranularityGrain.MONTH_GRAIN, getPos());
     }
   |
     <YEAR>
     {
-      granularity = Granularities.YEAR;
-      unparseString = "YEAR";
+      result = SqlLiteral.createSymbol(DruidSqlParserUtils.GranularityGrain.YEAR_GRAIN, getPos());
     }
   |
     <ALL>
     {
-      granularity = Granularities.ALL;
-      unparseString = "ALL";
+      result = SqlLiteral.createSymbol(DruidSqlParserUtils.GranularityGrain.ALL_GRAIN, getPos());
     }
     [
       <TIME>
       {
-        unparseString += " TIME";
+        result = SqlLiteral.createSymbol(DruidSqlParserUtils.GranularityGrain.ALL_TIME_GRAIN, getPos());
       }
     ]
   |
     e = Expression(ExprContext.ACCEPT_SUB_QUERY)
     {
-      granularity = DruidSqlParserUtils.convertSqlNodeToGranularityThrowingParseExceptions(e);
-      unparseString = e.toString();
+      result = e;
     }
   )
   {
-    return new org.apache.druid.java.util.common.Pair(granularity, unparseString);
+    return result;
   }
 }
 
diff --git a/sql/src/main/codegen/includes/insert.ftl b/sql/src/main/codegen/includes/insert.ftl
index 00133496d240..7ab2779f9875 100644
--- a/sql/src/main/codegen/includes/insert.ftl
+++ b/sql/src/main/codegen/includes/insert.ftl
@@ -78,13 +78,14 @@ SqlNode DruidSqlInsert() :
 SqlNode DruidSqlInsertEof() :
 {
   SqlNode insertNode;
-  org.apache.druid.java.util.common.Pair<Granularity, String> partitionedBy = new org.apache.druid.java.util.common.Pair(null, null);
+  SqlNode partitionedBy = null;
   SqlNodeList clusteredBy = null;
 }
 {
   insertNode = DruidSqlInsert()
-  // PARTITIONED BY is necessary, but is kept optional in the grammar. It is asserted that it is not missing in the
-  // DruidSqlInsert constructor so that we can return a custom error message.
+  // PARTITIONED BY is necessary. It can be provided either in this statement or in the catalog.
+  // As a result, it is optional in the grammar. It is asserted that it is not missing in the
+  // insert analysis step.
   [
     <PARTITIONED> <BY>
     partitionedBy = PartitionGranularity()
@@ -92,13 +93,6 @@ SqlNode DruidSqlInsertEof() :
   [
     clusteredBy = ClusteredBy()
   ]
-  {
-      if (clusteredBy != null && partitionedBy.lhs == null) {
-        throw org.apache.druid.sql.calcite.parser.DruidSqlParserUtils.problemParsing(
-          "CLUSTERED BY found before PARTITIONED BY, CLUSTERED BY must come after the PARTITIONED BY clause"
-        );
-      }
-  }
   // EOF is also present in SqlStmtEof but EOF is a special case and a single EOF can be consumed multiple times.
   // The reason for adding EOF here is to ensure that we create a DruidSqlInsert node after the syntax has been
   // validated and throw SQL syntax errors before performing validations in the DruidSqlInsert which can overshadow the
@@ -111,6 +105,6 @@ SqlNode DruidSqlInsertEof() :
       return insertNode;
     }
     SqlInsert sqlInsert = (SqlInsert) insertNode;
-    return new DruidSqlInsert(sqlInsert, partitionedBy.lhs, partitionedBy.rhs, clusteredBy);
+    return new DruidSqlInsert(sqlInsert, partitionedBy, clusteredBy);
   }
 }
diff --git a/sql/src/main/codegen/includes/replace.ftl b/sql/src/main/codegen/includes/replace.ftl
index f3ea3a567610..290fdd885625 100644
--- a/sql/src/main/codegen/includes/replace.ftl
+++ b/sql/src/main/codegen/includes/replace.ftl
@@ -25,8 +25,7 @@ SqlNode DruidSqlReplaceEof() :
     SqlNodeList columnList = null;
     final Span s;
     SqlInsert sqlInsert;
-    // Using fully qualified name for Pair class, since Calcite also has a same class name being used in the Parser.jj
-    org.apache.druid.java.util.common.Pair<Granularity, String> partitionedBy = new org.apache.druid.java.util.common.Pair(null, null);
+    SqlNode partitionedBy = null;
     SqlNodeList clusteredBy = null;
     final Pair<SqlNodeList, SqlNodeList> p;
     SqlNode replaceTimeQuery = null;
@@ -58,13 +57,6 @@ SqlNode DruidSqlReplaceEof() :
     [
       clusteredBy = ClusteredBy()
     ]
-    {
-        if (clusteredBy != null && partitionedBy.lhs == null) {
-          throw org.apache.druid.sql.calcite.parser.DruidSqlParserUtils.problemParsing(
-            "CLUSTERED BY found before PARTITIONED BY, CLUSTERED BY must come after the PARTITIONED BY clause"
-          );
-        }
-    }
     // EOF is also present in SqlStmtEof but EOF is a special case and a single EOF can be consumed multiple times.
     // The reason for adding EOF here is to ensure that we create a DruidSqlReplace node after the syntax has been
     // validated and throw SQL syntax errors before performing validations in the DruidSqlReplace which can overshadow the
@@ -72,7 +64,7 @@ SqlNode DruidSqlReplaceEof() :
     <EOF>
     {
         sqlInsert = new SqlInsert(s.end(source), SqlNodeList.EMPTY, table, source, columnList);
-        return new DruidSqlReplace(sqlInsert, partitionedBy.lhs, partitionedBy.rhs, clusteredBy, replaceTimeQuery);
+        return new DruidSqlReplace(sqlInsert, partitionedBy, clusteredBy, replaceTimeQuery);
     }
 }
 
diff --git a/sql/src/main/java/org/apache/calcite/sql/validate/ValidatorShim.java b/sql/src/main/java/org/apache/calcite/sql/validate/ValidatorShim.java
new file mode 100644
index 000000000000..a41a11563cb8
--- /dev/null
+++ b/sql/src/main/java/org/apache/calcite/sql/validate/ValidatorShim.java
@@ -0,0 +1,43 @@
+/*
+ * 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.
+ */
+
+//CHECKSTYLE.OFF: PackageName - Must be in Calcite
+
+package org.apache.calcite.sql.validate;
+
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlSelect;
+
+public class ValidatorShim
+{
+  /**
+   * Provide the ability to create an OrderBy scope to Druid-specific code.
+   * @param parent
+   * @param orderList
+   * @param select
+   * @return
+   */
+  public static OrderByScope newOrderByScope(
+      SqlValidatorScope parent,
+      SqlNodeList orderList,
+      SqlSelect select)
+  {
+    return new OrderByScope(parent, orderList, select);
+  }
+}
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlIngest.java b/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlIngest.java
index 146d13673bde..b9ca734b5268 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlIngest.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlIngest.java
@@ -19,11 +19,13 @@
 
 package org.apache.druid.sql.calcite.parser;
 
+import java.util.List;
+
 import org.apache.calcite.sql.SqlInsert;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.sql.parser.SqlParserPos;
-import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.calcite.util.ImmutableNullableList;
 
 import javax.annotation.Nullable;
 
@@ -34,10 +36,8 @@
  */
 public abstract class DruidSqlIngest extends SqlInsert
 {
-  protected final Granularity partitionedBy;
-
-  // Used in the unparse function to generate the original query since we convert the string to an enum
-  protected final String partitionedByStringForUnparse;
+  @Nullable
+  protected final SqlNode partitionedBy;
 
   @Nullable
   protected final SqlNodeList clusteredBy;
@@ -48,19 +48,17 @@ public DruidSqlIngest(
       SqlNode targetTable,
       SqlNode source,
       SqlNodeList columnList,
-      @Nullable Granularity partitionedBy,
-      @Nullable String partitionedByStringForUnparse,
+      @Nullable SqlNode partitionedBy,
       @Nullable SqlNodeList clusteredBy
   )
   {
     super(pos, keywords, targetTable, source, columnList);
 
-    this.partitionedByStringForUnparse = partitionedByStringForUnparse;
     this.partitionedBy = partitionedBy;
     this.clusteredBy = clusteredBy;
   }
 
-  public Granularity getPartitionedBy()
+  public SqlNode getPartitionedBy()
   {
     return partitionedBy;
   }
@@ -70,4 +68,14 @@ public SqlNodeList getClusteredBy()
   {
     return clusteredBy;
   }
+
+  @Override
+  public List<SqlNode> getOperandList()
+  {
+    return ImmutableNullableList.<SqlNode>builder()
+        .addAll(super.getOperandList())
+        .add(partitionedBy)
+        .add(clusteredBy)
+        .build();
+  }
 }
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlInsert.java b/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlInsert.java
index c2eeb2ed1e4d..4f67e4378269 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlInsert.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlInsert.java
@@ -24,7 +24,8 @@
 import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlWriter;
-import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.druid.sql.calcite.planner.DruidSqlIngestOperator;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -38,30 +39,40 @@ public class DruidSqlInsert extends DruidSqlIngest
 {
   public static final String SQL_INSERT_SEGMENT_GRANULARITY = "sqlInsertSegmentGranularity";
 
-  // This allows reusing super.unparse
-  public static final SqlOperator OPERATOR = SqlInsert.OPERATOR;
-
-  /**
-   * While partitionedBy and partitionedByStringForUnparse can be null as arguments to the constructor, this is
-   * disallowed (semantically) and the constructor performs checks to ensure that. This helps in producing friendly
-   * errors when the PARTITIONED BY custom clause is not present, and keeps its error separate from JavaCC/Calcite's
-   * custom errors which can be cryptic when someone accidentally forgets to explicitly specify the PARTITIONED BY clause
-   */
   public DruidSqlInsert(
       @Nonnull SqlInsert insertNode,
-      @Nullable Granularity partitionedBy,
-      @Nullable String partitionedByStringForUnparse,
+      @Nullable SqlNode partitionedBy,
       @Nullable SqlNodeList clusteredBy
   )
   {
-    super(
+    this(
         insertNode.getParserPosition(),
         (SqlNodeList) insertNode.getOperandList().get(0), // No better getter to extract this
         insertNode.getTargetTable(),
         insertNode.getSource(),
         insertNode.getTargetColumnList(),
         partitionedBy,
-        partitionedByStringForUnparse,
+        clusteredBy
+    );
+  }
+
+  public DruidSqlInsert(
+      SqlParserPos pos,
+      SqlNodeList keywords,
+      SqlNode targetTable,
+      SqlNode source,
+      SqlNodeList columnList,
+      @Nullable SqlNode partitionedBy,
+      @Nullable SqlNodeList clusteredBy
+  )
+  {
+    super(
+        pos,
+        keywords,
+        targetTable,
+        source,
+        columnList,
+        partitionedBy,
         clusteredBy
     );
   }
@@ -70,7 +81,7 @@ public DruidSqlInsert(
   @Override
   public SqlOperator getOperator()
   {
-    return OPERATOR;
+    return DruidSqlIngestOperator.INSERT_OPERATOR;
   }
 
   @Override
@@ -78,7 +89,7 @@ public void unparse(SqlWriter writer, int leftPrec, int rightPrec)
   {
     super.unparse(writer, leftPrec, rightPrec);
     writer.keyword("PARTITIONED BY");
-    writer.keyword(partitionedByStringForUnparse);
+    writer.keyword(partitionedBy.toString());
     if (getClusteredBy() != null) {
       writer.keyword("CLUSTERED BY");
       SqlWriter.Frame frame = writer.startList("", "");
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlParserUtils.java b/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlParserUtils.java
index 202df0f705c9..8f74f8281f99 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlParserUtils.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlParserUtils.java
@@ -19,8 +19,11 @@
 
 package org.apache.druid.sql.calcite.parser;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import org.apache.calcite.sql.SqlAsOperator;
 import org.apache.calcite.sql.SqlBasicCall;
@@ -42,7 +45,9 @@
 import org.apache.calcite.util.Pair;
 import org.apache.druid.error.DruidException;
 import org.apache.druid.error.InvalidSqlInput;
+import org.apache.druid.java.util.common.IAE;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularities;
 import org.apache.druid.java.util.common.granularity.Granularity;
 import org.apache.druid.java.util.common.granularity.GranularityType;
 import org.apache.druid.java.util.common.granularity.PeriodGranularity;
@@ -70,6 +75,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 
 public class DruidSqlParserUtils
@@ -77,6 +83,66 @@ public class DruidSqlParserUtils
   private static final Logger log = new Logger(DruidSqlParserUtils.class);
   public static final String ALL = "all";
 
+  public enum GranularityGrain
+  {
+    SECOND_GRAIN("SECOND", Granularities.SECOND),
+    MINUTE_GRAIN("MINUTE", Granularities.MINUTE),
+    HOUR_GRAIN("HOUR", Granularities.HOUR),
+    DAY_GRAIN("DAY", Granularities.DAY),
+    WEEK_GRAIN("WEEK", Granularities.WEEK),
+    MONTH_GRAIN("MONTH", Granularities.MONTH),
+    QUARTER_GRAIN("QUARTER", Granularities.QUARTER),
+    YEAR_GRAIN("YEAR", Granularities.YEAR),
+    ALL_GRAIN("ALL", Granularities.ALL),
+    ALL_TIME_GRAIN("ALL TIME", Granularities.ALL);
+
+    static Map<String, GranularityGrain> valueToGranularityGrain = ImmutableMap.of(
+        GranularityGrain.SECOND_GRAIN.value, GranularityGrain.SECOND_GRAIN,
+        GranularityGrain.MINUTE_GRAIN.value, GranularityGrain.MINUTE_GRAIN,
+        GranularityGrain.HOUR_GRAIN.value, GranularityGrain.HOUR_GRAIN,
+        GranularityGrain.DAY_GRAIN.value, GranularityGrain.DAY_GRAIN,
+        GranularityGrain.WEEK_GRAIN.value, GranularityGrain.WEEK_GRAIN,
+        GranularityGrain.MONTH_GRAIN.value, GranularityGrain.MONTH_GRAIN,
+        GranularityGrain.QUARTER_GRAIN.value, GranularityGrain.QUARTER_GRAIN,
+        GranularityGrain.YEAR_GRAIN.value, GranularityGrain.YEAR_GRAIN,
+        GranularityGrain.ALL_GRAIN.value, GranularityGrain.ALL_GRAIN,
+        GranularityGrain.ALL_TIME_GRAIN.value, GranularityGrain.ALL_TIME_GRAIN
+    );
+
+    private final String value;
+    private final Granularity granularity;
+    GranularityGrain(String value, Granularity granularity)
+    {
+      this.value = value;
+      this.granularity = granularity;
+    }
+
+    @Override
+    public String toString()
+    {
+      return value;
+    }
+
+    public String getValue()
+    {
+      return value;
+    }
+
+    public Granularity getGranularity()
+    {
+      return granularity;
+    }
+
+    public static GranularityGrain fromValue(String value)
+    {
+      GranularityGrain granularityGrain = valueToGranularityGrain.get(StringUtils.toUpperCase(value));
+      if (granularityGrain != null) {
+        return granularityGrain;
+      }
+      throw new IAE(PARTITION_ERROR_MESSAGE, "'" + value + "'");
+    }
+  }
+
   /**
    * Delegates to {@code convertSqlNodeToGranularity} and converts the exceptions to {@link ParseException}
    * with the underlying message
@@ -95,29 +161,72 @@ public static Granularity convertSqlNodeToGranularityThrowingParseExceptions(Sql
     }
   }
 
+  @VisibleForTesting
+  public static final String PARTITION_ERROR_MESSAGE =
+      "Encountered %s after PARTITIONED BY. " +
+      "Expected HOUR, DAY, MONTH, YEAR, ALL TIME, ISO 8601 period, FLOOR function or " +
+      TimeFloorOperatorConversion.SQL_FUNCTION_NAME + " function";
+
   /**
-   * This method is used to extract the granularity from a SqlNode representing following function calls:
-   * 1. FLOOR(__time TO TimeUnit)
-   * 2. TIME_FLOOR(__time, 'PT1H')
+   * This method is used to extract the granularity from a SqlNode which represents
+   * the argument to the {@code PARTITIONED BY} clause. The node can be any of the following:
+   * <ul>
+   * <li>A literal with a string that matches the SQL keywords
+   * {@code HOUR, DAY, MONTH, YEAR, ALL [TIME]}</li>
+   * <li>A literal string with a period in ISO 8601 format.</li>
+   * <li>Function call: {@code FLOOR(__time TO TimeUnit)}</li>
+   * <li>Function call: TIME_FLOOR(__time, 'PT1H')}</li>
+   * </ul>
    * <p>
-   * Validation on the sqlNode is contingent to following conditions:
-   * 1. sqlNode is an instance of SqlCall
-   * 2. Operator is either one of TIME_FLOOR or FLOOR
-   * 3. Number of operands in the call are 2
-   * 4. First operand is a SimpleIdentifier representing __time
-   * 5. If operator is TIME_FLOOR, the second argument is a literal, and can be converted to the Granularity class
-   * 6. If operator is FLOOR, the second argument is a TimeUnit, and can be mapped using {@link TimeUnits}
+   * Validation of the function sqlNode is contingent to following conditions:
+   * <ol>
+   * <li>sqlNode is an instance of SqlCall</li>
+   * <li>Operator is either one of TIME_FLOOR or FLOOR</li>
+   * <li>Number of operands in the call are 2</li>
+   * <li>First operand is a SimpleIdentifier representing __time</li>
+   * <li>If operator is TIME_FLOOR, the second argument is a literal, and can be converted to the Granularity class</li>
+   * <li>If operator is FLOOR, the second argument is a TimeUnit, and can be mapped using {@link TimeUnits}</li>
+   * </ol>
    * <p>
-   * Since it is to be used primarily while parsing the SqlNode, it is wrapped in {@code convertSqlNodeToGranularityThrowingParseExceptions}
+   * This method is called during validation, which will catch any errors. It is then called again
+   * during conversion, at which time we assume the node is valid.
    *
    * @param sqlNode SqlNode representing a call to a function
    *
    * @return Granularity as intended by the function call
    *
-   * @throws ParseException SqlNode cannot be converted a granularity
+   * @throws org.apache.druid.java.util.common.IAE if the SqlNode cannot be converted a granularity
    */
-  public static Granularity convertSqlNodeToGranularity(SqlNode sqlNode) throws ParseException
+  @Nullable
+  public static Granularity convertSqlNodeToGranularity(SqlNode sqlNode)
   {
+    if (sqlNode == null) {
+      return null;
+    }
+
+    if (sqlNode instanceof SqlLiteral) {
+      SqlLiteral literal = (SqlLiteral) sqlNode;
+      if (SqlLiteral.valueMatchesType(literal.getValue(), SqlTypeName.SYMBOL)) {
+        GranularityGrain value = literal.getValueAs(GranularityGrain.class);
+        if (value == null) {
+          throw new IAE(PARTITION_ERROR_MESSAGE, "NULL");
+        }
+        return value.getGranularity();
+      }
+
+      else if (SqlLiteral.valueMatchesType(literal.getValue(), SqlTypeName.CHAR)) {
+        String value = literal.getValueAs(String.class);
+        if (Strings.isNullOrEmpty(value)) {
+          throw new IAE(PARTITION_ERROR_MESSAGE, value == null ? "NULL" : "'" + value + "'");
+        }
+        return GranularityGrain.fromValue(value).getGranularity();
+      }
+
+      else {
+        throw new IAE(PARTITION_ERROR_MESSAGE, literal.getValue());
+      }
+    }
+
     if (!(sqlNode instanceof SqlCall)) {
       throw makeInvalidPartitionByException(sqlNode);
     }
@@ -166,7 +275,7 @@ public static Granularity convertSqlNodeToGranularity(SqlNode sqlNode) throws Pa
         period = new Period(granularityString);
       }
       catch (IllegalArgumentException e) {
-        throw new ParseException(StringUtils.format("%s is an invalid period string", granularitySqlNode.toString()));
+        throw new IAE(StringUtils.format("%s is an invalid period string", granularitySqlNode.toString()));
       }
       final PeriodGranularity retVal = new PeriodGranularity(period, null, null);
       validateSupportedGranularityForPartitionedBy(sqlNode, retVal);
@@ -210,7 +319,7 @@ private static DruidException makeInvalidPartitionByException(SqlNode sqlNode)
   }
 
   /**
-   * This method validates and converts a {@link SqlNode} representing a query into an optimized list of intervals to
+   * Validates and converts a {@link SqlNode} representing a query into an optimized list of intervals to
    * be used in creating an ingestion spec. If the sqlNode is an SqlLiteral of {@link #ALL}, returns a singleton list of
    * "ALL". Otherwise, it converts and optimizes the query using {@link MoveTimeFiltersToIntervals} into a list of
    * intervals which contain all valid values of time as per the query.
@@ -579,7 +688,10 @@ static String parseTimeStampWithTimeZone(SqlNode sqlNode, DateTimeZone timeZone)
     return String.valueOf(zonedTimestamp.toInstant().toEpochMilli());
   }
 
-  public static void validateSupportedGranularityForPartitionedBy(SqlNode originalNode, Granularity granularity)
+  public static void validateSupportedGranularityForPartitionedBy(
+      @Nullable SqlNode originalNode,
+      Granularity granularity
+  )
   {
     if (!GranularityType.isStandard(granularity)) {
       throw InvalidSqlInput.exception(
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlReplace.java b/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlReplace.java
index d527a08b59ec..086354bece79 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlReplace.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/parser/DruidSqlReplace.java
@@ -20,17 +20,18 @@
 package org.apache.druid.sql.calcite.parser;
 
 import org.apache.calcite.sql.SqlInsert;
-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.SqlOperator;
-import org.apache.calcite.sql.SqlSpecialOperator;
 import org.apache.calcite.sql.SqlWriter;
-import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.util.ImmutableNullableList;
+import org.apache.druid.sql.calcite.planner.DruidSqlIngestOperator;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
+import java.util.List;
 
 /**
  * Extends the 'replace' call to hold custom parameters specific to Druid i.e. PARTITIONED BY and the PARTITION SPECS
@@ -41,8 +42,6 @@ public class DruidSqlReplace extends DruidSqlIngest
 {
   public static final String SQL_REPLACE_TIME_CHUNKS = "sqlReplaceTimeChunks";
 
-  public static final SqlOperator OPERATOR = new SqlSpecialOperator("REPLACE", SqlKind.OTHER);
-
   private final SqlNode replaceTimeQuery;
 
   /**
@@ -53,20 +52,41 @@ public class DruidSqlReplace extends DruidSqlIngest
    */
   public DruidSqlReplace(
       @Nonnull SqlInsert insertNode,
-      @Nullable Granularity partitionedBy,
-      @Nullable String partitionedByStringForUnparse,
+      @Nullable SqlNode partitionedBy,
       @Nullable SqlNodeList clusteredBy,
       @Nullable SqlNode replaceTimeQuery
   )
   {
-    super(
+    this(
         insertNode.getParserPosition(),
         (SqlNodeList) insertNode.getOperandList().get(0), // No better getter to extract this
         insertNode.getTargetTable(),
         insertNode.getSource(),
         insertNode.getTargetColumnList(),
         partitionedBy,
-        partitionedByStringForUnparse,
+        clusteredBy,
+        replaceTimeQuery
+    );
+  }
+
+  public DruidSqlReplace(
+      SqlParserPos pos,
+      SqlNodeList keywords,
+      SqlNode targetTable,
+      SqlNode source,
+      SqlNodeList columnList,
+      @Nullable SqlNode partitionedBy,
+      @Nullable SqlNodeList clusteredBy,
+      @Nullable SqlNode replaceTimeQuery
+  )
+  {
+    super(
+        pos,
+        keywords,
+        targetTable,
+        source,
+        columnList,
+        partitionedBy,
         clusteredBy
     );
 
@@ -78,11 +98,19 @@ public SqlNode getReplaceTimeQuery()
     return replaceTimeQuery;
   }
 
-  @Nonnull
   @Override
   public SqlOperator getOperator()
   {
-    return OPERATOR;
+    return DruidSqlIngestOperator.REPLACE_OPERATOR;
+  }
+
+  @Override
+  public List<SqlNode> getOperandList()
+  {
+    return ImmutableNullableList.<SqlNode>builder()
+        .addAll(super.getOperandList())
+        .add(replaceTimeQuery)
+        .build();
   }
 
   @Override
@@ -111,7 +139,7 @@ public void unparse(SqlWriter writer, int leftPrec, int rightPrec)
     writer.newlineAndIndent();
 
     writer.keyword("PARTITIONED BY");
-    writer.keyword(partitionedByStringForUnparse);
+    writer.keyword(partitionedBy.toString());
 
     if (getClusteredBy() != null) {
       writer.keyword("CLUSTERED BY");
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlanner.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlanner.java
index 2cda011848b5..ea5a9991b0d2 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlanner.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlanner.java
@@ -86,6 +86,7 @@ public class CalcitePlanner implements Planner, ViewExpander
   private final @Nullable RelOptCostFactory costFactory;
   private final Context context;
   private final CalciteConnectionConfig connectionConfig;
+  private final DruidSqlValidator.ValidatorContext validatorContext;
   private final RelDataTypeSystem typeSystem;
 
   /**
@@ -118,9 +119,10 @@ public class CalcitePlanner implements Planner, ViewExpander
    * {@link org.apache.calcite.tools.Frameworks#getPlanner} instead.
    */
   @SuppressWarnings("method.invocation.invalid")
-  public CalcitePlanner(FrameworkConfig config)
+  public CalcitePlanner(FrameworkConfig config, DruidSqlValidator.ValidatorContext validatorContext)
   {
     this.costFactory = config.getCostFactory();
+    this.validatorContext = validatorContext;
     this.defaultSchema = config.getDefaultSchema();
     this.operatorTable = config.getOperatorTable();
     this.programs = config.getPrograms();
@@ -207,7 +209,7 @@ private void ready()
         planner.addRelTraitDef(RelCollationTraitDef.INSTANCE);
       }
     } else {
-      for (RelTraitDef def : this.traitDefs) {
+      for (RelTraitDef<?> def : this.traitDefs) {
         planner.addRelTraitDef(def);
       }
     }
@@ -240,7 +242,11 @@ public SqlNode validate(SqlNode sqlNode) throws ValidationException
       validatedSqlNode = validator.validate(sqlNode);
     }
     catch (RuntimeException e) {
-      throw new ValidationException(e);
+      // Change from Calcite: pass the message explicitly to avoid messages like:
+      // foo.bar.MyException: The real message here
+      // By passing the message, get get the simpler form:
+      // The real message here
+      throw new ValidationException(e.getMessage(), e);
     }
     state = CalcitePlanner.State.STATE_4_VALIDATED;
     return validatedSqlNode;
@@ -279,13 +285,21 @@ public final RelNode convert(SqlNode sql)
     return rel(sql).rel;
   }
 
+  /**
+   * Convert the given node to a relational operator. Converts the give node
+   * rather than the one given by {@link #validatedSqlNode}. This is a change
+   * from the original Calcite code. (Though, oddly, Calcite accepted an argument
+   * and didn't use it.) We use the passed-in argument because Druid ignores the
+   * INSERT node: using just the SELECT within the INSERT. The {@code validatedQueryNode}
+   * points to the (unwanted) INSERT, the passed in node is the actual query.
+   */
   @Override
   public RelRoot rel(SqlNode sql)
   {
     ensure(CalcitePlanner.State.STATE_4_VALIDATED);
     SqlNode validatedSqlNode = Objects.requireNonNull(
-        this.validatedSqlNode,
-        "validatedSqlNode is null. Need to call #validate() first"
+        sql,
+        "sql is null. Need to call #validate() first"
     );
     final RexBuilder rexBuilder = createRexBuilder();
     final RelOptCluster cluster = RelOptCluster.create(
@@ -295,8 +309,8 @@ public RelRoot rel(SqlNode sql)
     final SqlToRelConverter.Config config =
         sqlToRelConverterConfig.withTrimUnusedFields(false);
     final SqlToRelConverter sqlToRelConverter =
-        new SqlToRelConverter(this, validator,
-                              createCatalogReader(), cluster, convertletTable, config
+        new DruidSqlToRelConverter(this, validator,
+            createCatalogReader(), cluster, convertletTable, config
         );
     RelRoot root =
         sqlToRelConverter.convertQuery(validatedSqlNode, false, true);
@@ -403,12 +417,15 @@ private SqlValidator createSqlValidator(CalciteCatalogReader catalogReader)
         SqlValidator.Config.DEFAULT.withConformance(connectionConfig.conformance())
                                    .withLenientOperatorLookup(connectionConfig.lenientOperatorLookup())
                                    .withIdentifierExpansion(true);
+                                   // TODO: check on this
+                                   //.withTypeCoercionEnabled(false);
     return new DruidSqlValidator(
         opTab,
         catalogReader,
         getTypeFactory(),
         validatorConfig,
-        context.unwrapOrThrow(PlannerContext.class)
+        context.unwrapOrThrow(PlannerContext.class),
+        validatorContext
     );
   }
 
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlannerModule.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlannerModule.java
index 272fe5030286..12b3b2be6057 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlannerModule.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlannerModule.java
@@ -46,5 +46,11 @@ public void configure(Binder binder)
     binder.bind(PlannerFactory.class).in(LazySingleton.class);
     binder.bind(DruidOperatorTable.class).in(LazySingleton.class);
     Multibinder.newSetBinder(binder, ExtensionCalciteRuleProvider.class);
+
+    // Catalog resolver: the planner's interface into the (optional) catalog.
+    // This is the default no-op binding used unless the catalog is enabled.
+    binder.bind(CatalogResolver.class)
+        .to(CatalogResolver.NullCatalogResolver.class)
+        .in(LazySingleton.class);
   }
 }
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidCatalogReader.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidCatalogReader.java
new file mode 100644
index 000000000000..490ae8a60edc
--- /dev/null
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidCatalogReader.java
@@ -0,0 +1,166 @@
+/*
+ * 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.druid.sql.calcite.planner;
+
+import com.google.common.collect.Iterators;
+import org.apache.calcite.config.CalciteConnectionConfig;
+import org.apache.calcite.jdbc.CalciteSchema;
+import org.apache.calcite.prepare.CalciteCatalogReader;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.schema.Function;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlFunctionCategory;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.SqlSyntax;
+import org.apache.calcite.sql.validate.SqlNameMatcher;
+import org.apache.calcite.sql.validate.SqlValidatorUtil;
+import org.apache.druid.catalog.model.TableId;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.server.security.ResourceAction;
+import org.apache.druid.sql.calcite.external.DruidTableMacro;
+import org.apache.druid.sql.calcite.external.DruidUserDefinedTableMacro;
+import org.apache.druid.sql.calcite.external.Externals;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Druid-specific catalog reader which provides the special processing needed for
+ * external table macros which must be wrapped in a Druid-specific operator class.
+ */
+public class DruidCatalogReader extends CalciteCatalogReader
+{
+  public DruidCatalogReader(
+      final CalciteSchema rootSchema,
+      final List<String> defaultSchema,
+      final RelDataTypeFactory typeFactory,
+      final CalciteConnectionConfig config
+  )
+  {
+    super(rootSchema, defaultSchema, typeFactory, config);
+  }
+
+  @Override
+  public void lookupOperatorOverloads(final SqlIdentifier opName,
+                                      final SqlFunctionCategory category,
+                                      final SqlSyntax syntax,
+                                      final List<SqlOperator> operatorList,
+                                      final SqlNameMatcher nameMatcher
+  )
+  {
+    if (syntax != SqlSyntax.FUNCTION) {
+      return;
+    }
+
+    // A shame to do it this way...
+    // We need our own operator for external table functions. Calcite locks down the
+    // mapping from functions to operators and does not provide a way for us to
+    // customize. So, we have to intercept external schema lookups and do our own
+    // conversion to operators.
+    if (isExt(opName, nameMatcher)) {
+      lookupExtOperator(opName, operatorList, nameMatcher);
+    } else {
+      // Look in some other schema.
+      super.lookupOperatorOverloads(opName, category, syntax, operatorList, nameMatcher);
+    }
+  }
+
+  private boolean isExt(
+      SqlIdentifier opName,
+      SqlNameMatcher nameMatcher
+  )
+  {
+    if (opName.names.size() != 2) {
+      return false;
+    }
+    return nameMatcher.matches(TableId.EXTERNAL_SCHEMA, opName.names.get(0));
+  }
+
+  /**
+   * Resolve an external table as a function. The table may be "complete" which means it
+   * takes zero arguments. We allow this, it is OK, if verbose, to say
+   * {@code FROM(TABLE(ext.foo())} instead of the simpler {@code FROM ext.foo}. Most
+   * external tables are partial, meaning that they need extra parameters a query time,
+   * which means that, in Calcite terms, they are table functions. We wrap those functions
+   * in a Druid specific table macro, then wrap that macro in a Druid-specific operator
+   * which is {@code EXTEND}-aware so that we can pass the schema into the function.
+   */
+  private void lookupExtOperator(SqlIdentifier opName, List<SqlOperator> operatorList, SqlNameMatcher nameMatcher)
+  {
+    CalciteSchema schema =
+        SqlValidatorUtil.getSchema(rootSchema, opName.names.subList(0, 1), nameMatcher);
+    if (schema == null) {
+      return;
+    }
+    Collection<Function> fns = schema.getFunctions(opName.names.get(1), nameMatcher.isCaseSensitive());
+    if (fns.isEmpty()) {
+      return;
+    }
+    if (fns.size() > 1) {
+      throw new ISE("Found multiple table functions for %s", opName.toString());
+    }
+    Function fn = Iterators.getOnlyElement(fns.iterator());
+    if (!(fn instanceof DruidTableMacro)) {
+      throw new ISE("Table function %s is not a DruidTableMacro", opName.toString());
+    }
+    operatorList.add(new ExternalUserDefinedTableMacro(opName, (DruidTableMacro) fn));
+  }
+
+  /**
+   * Druid user defined table macro that preserves the full function name so that
+   * later re-resolutions of this function will search only in the ext schema
+   * rather than also searching in all schemas. (By default, Calcite preserves only
+   * the tail of the name, not the prefix.)
+   * <p>
+   * Also, we want the auth resource to use the actual table name, not just the
+   * generic `EXTERNAL`.
+   */
+  public static class ExternalUserDefinedTableMacro extends DruidUserDefinedTableMacro
+  {
+    private final SqlIdentifier fullName;
+
+    public ExternalUserDefinedTableMacro(SqlIdentifier name, DruidTableMacro macro)
+    {
+      super(macro);
+      this.fullName = name;
+    }
+
+    @Override
+    public SqlIdentifier getNameAsId()
+    {
+      return fullName;
+    }
+
+    @Override
+    public String toString()
+    {
+      return fullName.toString();
+    }
+
+    @Override
+    public Set<ResourceAction> computeResources(final SqlCall call, final boolean inputSourceTypeSecurityEnabled)
+    {
+      return Collections.singleton(Externals.externalRead(fullName.names.get(fullName.names.size() - 1)));
+    }
+  }
+}
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidPlanner.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidPlanner.java
index 4b697a0d5dfa..b5902b8bcee8 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidPlanner.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidPlanner.java
@@ -61,6 +61,10 @@
  * lifecycle defined as:
  * <p>
  * start --> validate [--> prepare] --> plan
+ * <p>
+ * Druid supports a variety of statements. Each statement (or statement family)
+ * has a "handler" for that statement type. To add a new statement, add a new
+ * handler (which might reuse one of the existing base handlers.)
  */
 public class DruidPlanner implements Closeable
 {
@@ -119,7 +123,7 @@ public AuthResult(
   )
   {
     this.frameworkConfig = frameworkConfig;
-    this.planner = new CalcitePlanner(frameworkConfig);
+    this.planner = new CalcitePlanner(frameworkConfig, new ValidatorContextImpl());
     this.plannerContext = plannerContext;
     this.engine = engine;
     this.hook = hook == null ? NoOpPlannerHook.INSTANCE : hook;
@@ -174,7 +178,7 @@ private SqlStatementHandler createHandler(final SqlNode node)
     }
 
     if (query.isA(SqlKind.QUERY)) {
-      return new QueryHandler.SelectHandler(handlerContext, query, explain);
+      return new SelectHandler(handlerContext, query, explain);
     }
     throw InvalidSqlInput.exception("Unsupported SQL statement [%s]", node.getKind());
   }
@@ -302,7 +306,7 @@ public SchemaPlus defaultSchema()
     @Override
     public ObjectMapper jsonMapper()
     {
-      return plannerContext.getJsonMapper();
+      return plannerContext.getPlannerToolbox().jsonMapper();
     }
 
     @Override
@@ -311,6 +315,12 @@ public DateTimeZone timeZone()
       return plannerContext.getTimeZone();
     }
 
+    @Override
+    public CatalogResolver catalog()
+    {
+      return plannerContext.getPlannerToolbox().catalogResolver();
+    }
+
     @Override
     public PlannerHook hook()
     {
@@ -318,6 +328,33 @@ public PlannerHook hook()
     }
   }
 
+  public class ValidatorContextImpl implements DruidSqlValidator.ValidatorContext
+  {
+    @Override
+    public Map<String, Object> queryContextMap()
+    {
+      return plannerContext.queryContextMap();
+    }
+
+    @Override
+    public CatalogResolver catalog()
+    {
+      return plannerContext.getPlannerToolbox().catalogResolver();
+    }
+
+    @Override
+    public String druidSchemaName()
+    {
+      return plannerContext.getPlannerToolbox().druidSchemaName();
+    }
+
+    @Override
+    public ObjectMapper jsonMapper()
+    {
+      return plannerContext.getPlannerToolbox().jsonMapper();
+    }
+  }
+
   public static DruidException translateException(Exception e)
   {
     try {
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlIngestOperator.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlIngestOperator.java
new file mode 100644
index 000000000000..26994c6261d4
--- /dev/null
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlIngestOperator.java
@@ -0,0 +1,120 @@
+/*
+ * 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.druid.sql.calcite.planner;
+
+import com.google.common.collect.ImmutableSet;
+import org.apache.calcite.sql.SqlCall;
+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.SqlSpecialOperator;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.Resource;
+import org.apache.druid.server.security.ResourceAction;
+import org.apache.druid.server.security.ResourceType;
+import org.apache.druid.sql.calcite.expression.AuthorizableOperator;
+import org.apache.druid.sql.calcite.parser.DruidSqlIngest;
+import org.apache.druid.sql.calcite.parser.DruidSqlInsert;
+import org.apache.druid.sql.calcite.parser.DruidSqlReplace;
+
+import java.util.Set;
+
+public class DruidSqlIngestOperator extends SqlSpecialOperator implements AuthorizableOperator
+{
+  public static final SqlSpecialOperator INSERT_OPERATOR =
+      new DruidSqlInsertOperator();
+  public static final SqlSpecialOperator REPLACE_OPERATOR =
+      new DruidSqlReplaceOperator();
+
+  public static class DruidSqlInsertOperator extends DruidSqlIngestOperator
+  {
+    public DruidSqlInsertOperator()
+    {
+      super("INSERT");
+    }
+
+    @Override
+    public SqlCall createCall(
+        SqlLiteral functionQualifier,
+        SqlParserPos pos,
+        SqlNode... operands
+    )
+    {
+      return new DruidSqlInsert(
+          pos,
+          // Must match SqlInsert.getOperandList()
+          (SqlNodeList) operands[0],
+          operands[1],
+          operands[2],
+          (SqlNodeList) operands[3],
+          // Must match DruidSqlIngest.getOperandList()
+          operands[4],
+          (SqlNodeList) operands[5]
+      );
+    }
+  }
+
+  public static class DruidSqlReplaceOperator extends DruidSqlIngestOperator
+  {
+    public DruidSqlReplaceOperator()
+    {
+      super("REPLACE");
+    }
+
+    @Override
+    public SqlCall createCall(
+        SqlLiteral functionQualifier,
+        SqlParserPos pos,
+        SqlNode... operands
+    )
+    {
+      return new DruidSqlReplace(
+          pos,
+          // Must match SqlInsert.getOperandList()
+          (SqlNodeList) operands[0],
+          operands[1],
+          operands[2],
+          (SqlNodeList) operands[3],
+          // Must match DruidSqlIngest.getOperandList()
+          operands[4],
+          (SqlNodeList) operands[5],
+          // Must match DruidSqlReplace.getOperandList()
+          operands[6]
+      );
+    }
+  }
+
+  public DruidSqlIngestOperator(String name)
+  {
+    super(name, SqlKind.INSERT);
+  }
+
+  @Override
+  public Set<ResourceAction> computeResources(SqlCall call, final boolean inputSourceTypeSecurityEnabled)
+  {
+    DruidSqlIngest ingestNode = (DruidSqlIngest) call;
+    final SqlIdentifier tableIdentifier = (SqlIdentifier) ingestNode.getTargetTable();
+    String targetDatasource = tableIdentifier.names.get(tableIdentifier.names.size() - 1);
+    return ImmutableSet.of(new ResourceAction(new Resource(targetDatasource, ResourceType.DATASOURCE), Action.WRITE));
+  }
+}
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlToRelConverter.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlToRelConverter.java
new file mode 100644
index 000000000000..d5f68748337a
--- /dev/null
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlToRelConverter.java
@@ -0,0 +1,69 @@
+/*
+ * 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.druid.sql.calcite.planner;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptTable.ViewExpander;
+import org.apache.calcite.prepare.Prepare.CatalogReader;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlInsert;
+import org.apache.calcite.sql.validate.SqlValidator;
+import org.apache.calcite.sql2rel.SqlRexConvertletTable;
+import org.apache.calcite.sql2rel.SqlToRelConverter;
+
+public class DruidSqlToRelConverter extends SqlToRelConverter
+{
+  public DruidSqlToRelConverter(
+      final ViewExpander viewExpander,
+      final SqlValidator validator,
+      final CatalogReader catalogReader,
+      RelOptCluster cluster,
+      final SqlRexConvertletTable convertletTable,
+      final Config config
+  )
+  {
+    super(viewExpander, validator, catalogReader, cluster, convertletTable, config);
+  }
+
+  /**
+   * Convert a Druid {@code INSERT} or {@code REPLACE} statement. The code is the same
+   * as the normal conversion, except we don't actually create the final modify node.
+   * Druid has its own special way to handle inserts. (This should probably change in
+   * some future, but doing so requires changes in the SQL engine and MSQ, which is a bit
+   * invasive.)
+   */
+  @Override
+  protected RelNode convertInsert(SqlInsert call)
+  {
+    // Get the target type: the column types we want to write into the target datasource.
+    final RelDataType targetRowType = validator.getValidatedNodeType(call);
+    assert targetRowType != null;
+
+    // Convert the underlying SELECT. We pushed the CLUSTERED BY clause into the SELECT
+    // as its ORDER BY. We claim this is the top query because MSQ doesn't actually
+    // use the Calcite insert node.
+    RelNode sourceRel = convertQueryRecursive(call.getSource(), true, targetRowType).project();
+
+    // We omit the column mapping and insert node that Calcite normally provides.
+    // Presumably MSQ does these its own way.
+    return sourceRel;
+  }
+}
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java
index 34f3c7410f38..8cba0712dafe 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/DruidSqlValidator.java
@@ -19,19 +19,67 @@
 
 package org.apache.druid.sql.calcite.planner;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.calcite.adapter.java.JavaTypeFactory;
 import org.apache.calcite.prepare.BaseDruidSqlValidator;
 import org.apache.calcite.prepare.CalciteCatalogReader;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rel.type.RelRecordType;
 import org.apache.calcite.runtime.CalciteContextException;
 import org.apache.calcite.runtime.CalciteException;
+import org.apache.calcite.sql.SqlBasicCall;
 import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlInsert;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlNumericLiteral;
 import org.apache.calcite.sql.SqlOperatorTable;
+import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlWith;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.calcite.sql.type.BasicSqlType;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.validate.IdentifierNamespace;
+import org.apache.calcite.sql.validate.OrderByScope;
+import org.apache.calcite.sql.validate.SqlValidatorException;
+import org.apache.calcite.sql.validate.SqlValidatorImpl;
+import org.apache.calcite.sql.validate.SqlValidatorNamespace;
 import org.apache.calcite.sql.validate.SqlValidatorScope;
+import org.apache.calcite.sql.validate.SqlValidatorTable;
+import org.apache.calcite.sql.validate.ValidatorShim;
+import org.apache.calcite.util.Pair;
+import org.apache.commons.lang.reflect.FieldUtils;
+import org.apache.druid.catalog.model.Columns;
+import org.apache.druid.catalog.model.facade.DatasourceFacade;
+import org.apache.druid.catalog.model.table.ClusterKeySpec;
+import org.apache.druid.common.utils.IdUtils;
+import org.apache.druid.error.InvalidSqlInput;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.common.granularity.Granularity;
+import org.apache.druid.java.util.common.granularity.PeriodGranularity;
+import org.apache.druid.sql.calcite.parser.DruidSqlIngest;
+import org.apache.druid.sql.calcite.parser.DruidSqlInsert;
+import org.apache.druid.sql.calcite.parser.DruidSqlParserUtils;
 import org.apache.druid.sql.calcite.run.EngineFeature;
+import org.apache.druid.sql.calcite.table.DatasourceTable;
+import org.apache.druid.utils.CollectionUtils;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Pattern;
 
 /**
  * Druid extended SQL validator. (At present, it doesn't actually
@@ -39,18 +87,696 @@
  */
 class DruidSqlValidator extends BaseDruidSqlValidator
 {
+  private static final Pattern UNNAMED_COLUMN_PATTERN = Pattern.compile("^EXPR\\$\\d+$", Pattern.CASE_INSENSITIVE);
+
+  // Copied here from MSQE since that extension is not visible here.
+  public static final String CTX_ROWS_PER_SEGMENT = "msqRowsPerSegment";
+
+  public interface ValidatorContext
+  {
+    Map<String, Object> queryContextMap();
+    CatalogResolver catalog();
+    String druidSchemaName();
+    ObjectMapper jsonMapper();
+  }
+
+  private final ValidatorContext validatorContext;
   private final PlannerContext plannerContext;
 
   protected DruidSqlValidator(
-      SqlOperatorTable opTab,
-      CalciteCatalogReader catalogReader,
-      JavaTypeFactory typeFactory,
-      Config validatorConfig,
-      PlannerContext plannerContext
+      final SqlOperatorTable opTab,
+      final CalciteCatalogReader catalogReader,
+      final JavaTypeFactory typeFactory,
+      final Config validatorConfig,
+      final PlannerContext plannerContext,
+      final ValidatorContext validatorContext
   )
   {
     super(opTab, catalogReader, typeFactory, validatorConfig);
     this.plannerContext = plannerContext;
+    this.validatorContext = validatorContext;
+  }
+
+  /**
+   * Druid-specific validation for an INSERT statement. In Druid, the columns are
+   * matched by name. A datasource, by default, allows the insertion of arbitrary columns,
+   * but the catalog may enforce a strict schema (all columns must exist). Destination
+   * types are set by the catalog, where available, else by the query.
+   * <p>
+   * The Druid {@code INSERT} statement is non-standard in a variety of ways:
+   * <ul>
+   * <li>Allows the target table to not yet exist. Instead, {@code INSERT}
+   * creates it.</li>
+   * <li>Does not allow specifying the list of columns:
+   * {@code INSERT INTO dst (a, b, c) ...}</li>
+   * <li>When given without target columns (the only form allowed), columns are
+   * not matched by schema position as in standard SQL, but rather by name.</li>
+   * <li>There is no requirement that the target columns already exist. In fact,
+   * even if the target column exists, any existing type is ignored if not specified
+   * in the catalog.</li>
+   * <li>The source can only be a {@code SELECT} statement, not {@code VALUES}.</li>
+   * <li>Types thus propagate upwards from the {@code SELECT} to the the target
+   * table. Standard SQL says that types propagate down from the target table to the
+   * source.</li>
+   * <li>The __time column is special in multiple ways.</li>
+   * <li>Includes the {@code CLUSTERED BY} and {@code PARTITIONED BY} clauses.</li>
+   * </ul>
+   * The result is that the validation for the Druid {@code INSERT} is wildly customized
+   * relative to standard SQL.
+   */
+  // TODO: Ensure the source and target are not the same
+  @Override
+  public void validateInsert(final SqlInsert insert)
+  {
+    final DruidSqlIngest ingestNode = (DruidSqlIngest) insert;
+    if (insert.isUpsert()) {
+      throw new IAE("UPSERT is not supported.");
+    }
+
+    // SQL-style INSERT INTO dst (a, b, c) is not (yet) supported.
+    final String operationName = insert.getOperator().getName();
+    if (insert.getTargetColumnList() != null) {
+      throw new IAE("%s with a target column list is not supported.", operationName);
+    }
+
+    // The target namespace is both the target table ID and the row type for that table.
+    final SqlValidatorNamespace targetNamespace = getNamespace(insert);
+    final IdentifierNamespace insertNs = (IdentifierNamespace) targetNamespace;
+
+    // The target is a new or existing datasource.
+    final DatasourceTable table = validateInsertTarget(targetNamespace, insertNs, operationName);
+
+    // An existing datasource may have metadata.
+    final DatasourceFacade tableMetadata = table == null ? null : table.effectiveMetadata().catalogMetadata();
+
+    // Validate segment granularity, which depends on nothing else.
+    validateSegmentGranularity(operationName, ingestNode, tableMetadata);
+
+    // The source must be a SELECT
+    final SqlNode source = insert.getSource();
+    ensureNoOrderBy(source, operationName);
+
+    // Convert CLUSTERED BY, or the catalog equivalent, to an ORDER BY clause
+    final SqlNodeList catalogClustering = convertCatalogClustering(tableMetadata);
+    rewriteClusteringToOrderBy(source, ingestNode, catalogClustering);
+
+    // Validate the source statement. Validates the ORDER BY pushed down in the above step.
+    // Because of the non-standard Druid semantics, we can't define the target type: we don't know
+    // the target columns yet, and we can't infer types when they must come from the SELECT.
+    // Normally, the target type is known, and is pushed into the SELECT. In Druid, the SELECT
+    // usually defines the target types, unless the catalog says otherwise. Since catalog entries
+    // are optional, we don't know the target type until we validate the SELECT. (Also, we won't
+    // know names and we match by name.) Thus, we'd have to validate (to know names and types)
+    // to get the target types, but we need the target types to validate. Catch-22. So, we punt.
+    final SqlValidatorScope scope;
+    if (source instanceof SqlSelect) {
+      final SqlSelect sqlSelect = (SqlSelect) source;
+      validateSelect(sqlSelect, unknownType);
+      scope = null;
+    } else {
+      scope = scopes.get(source);
+      validateQuery(source, scope, unknownType);
+    }
+
+    final SqlValidatorNamespace sourceNamespace = namespaces.get(source);
+    final RelRecordType sourceType = (RelRecordType) sourceNamespace.getRowType();
+
+    // Validate the __time column
+    int timeColumnIndex = sourceType.getFieldNames().indexOf(Columns.TIME_COLUMN);
+    if (timeColumnIndex != -1) {
+      validateTimeColumn(sourceType, timeColumnIndex);
+    }
+
+    // Validate clustering against the SELECT row type. Clustering has additional
+    // constraints beyond what was validated for the pushed-down ORDER BY.
+    // Though we pushed down clustering above, only now can we validate it after
+    // we've determined the SELECT row type.
+    validateClustering(sourceType, ingestNode, catalogClustering);
+
+    // Determine the output (target) schema.
+    final RelDataType targetType = validateTargetType(scope, insertNs, insert, sourceType, tableMetadata);
+
+    // Set the type for the INSERT/REPLACE node
+    setValidatedNodeType(insert, targetType);
+
+    // Segment size
+    if (tableMetadata != null && !validatorContext.queryContextMap().containsKey(CTX_ROWS_PER_SEGMENT)) {
+      final Integer targetSegmentRows = tableMetadata.targetSegmentRows();
+      if (targetSegmentRows != null) {
+        validatorContext.queryContextMap().put(CTX_ROWS_PER_SEGMENT, targetSegmentRows);
+      }
+    }
+  }
+
+  /**
+   * Validate the target table. Druid {@code INSERT/REPLACE} can create a new datasource,
+   * or insert into an existing one. If the target exists, it must be a datasource. If it
+   * does not exist, the target must be in the datasource schema, normally "druid".
+   */
+  private DatasourceTable validateInsertTarget(
+      final SqlValidatorNamespace targetNamespace,
+      final IdentifierNamespace insertNs,
+      final String operationName
+  )
+  {
+    // Get the target table ID
+    final SqlIdentifier destId = insertNs.getId();
+    if (destId.names.isEmpty()) {
+      // I don't think this can happen, but include a branch for it just in case.
+      throw new IAE("%s requires a target table.", operationName);
+    }
+
+    // Druid does not support 3+ part names.
+    final int n = destId.names.size();
+    if (n > 2) {
+      throw new IAE("Druid does not support 3+ part names: [%s]", destId);
+    }
+    String tableName = destId.names.get(n - 1);
+
+    // If this is a 2-part name, the first part must be the datasource schema.
+    if (n == 2 && !validatorContext.druidSchemaName().equals(destId.names.get(0))) {
+      throw new IAE("Cannot %s into [%s] because the table is not in the 'druid' schema",
+          operationName,
+          destId
+      );
+    }
+    try {
+      // Try to resolve the table. Will fail if this is an INSERT into a new table.
+      validateNamespace(targetNamespace, unknownType);
+      SqlValidatorTable target = insertNs.resolve().getTable();
+      try {
+        return target.unwrap(DatasourceTable.class);
+      }
+      catch (Exception e) {
+        throw new IAE("Cannot %s into [%s] because it is not a datasource", operationName, destId);
+      }
+    }
+    catch (CalciteContextException e) {
+      // Something failed. Let's make sure it was the table lookup.
+      // The check is kind of a hack, but its the best we can do given that Calcite
+      // didn't expect this non-SQL use case.
+      if (e.getCause() instanceof SqlValidatorException && e.getMessage().contains(StringUtils.format("Object '%s' not found", tableName))) {
+        // The catalog implementation may be "strict": and require that the target
+        // table already exists, rather than the default "lenient" mode that can
+        // create a new table.
+        if (validatorContext.catalog().ingestRequiresExistingTable()) {
+          throw new IAE("Cannot %s into [%s] because it does not exist", operationName, destId);
+        }
+        // New table. Validate the shape of the name.
+        IdUtils.validateId("table", tableName);
+        return null;
+      }
+      throw e;
+    }
+  }
+
+  private void validateSegmentGranularity(
+      final String operationName,
+      final DruidSqlIngest ingestNode,
+      final DatasourceFacade tableMetadata
+  )
+  {
+    final Granularity definedGranularity = tableMetadata == null ? null : tableMetadata.segmentGranularity();
+    if (definedGranularity != null) {
+      // Should already have been checked when creating the catalog entry
+      DruidSqlParserUtils.validateSupportedGranularityForPartitionedBy(null, definedGranularity);
+    }
+    final Granularity ingestionGranularity = DruidSqlParserUtils.convertSqlNodeToGranularity(ingestNode.getPartitionedBy());
+    if (ingestionGranularity != null) {
+      DruidSqlParserUtils.validateSupportedGranularityForPartitionedBy(ingestNode, ingestionGranularity);
+    }
+    final Granularity finalGranularity;
+    if (definedGranularity == null) {
+      // The catalog has no granularity: apply the query value
+      if (ingestionGranularity == null) {
+        // Neither have a value: error
+        throw InvalidSqlInput.exception(
+            "Operation [%s] requires a PARTITIONED BY to be explicitly defined, but none was found.",
+            operationName
+        );
+      } else {
+        finalGranularity = ingestionGranularity;
+      }
+    } else {
+      // The catalog has a granularity
+      if (ingestionGranularity == null) {
+        // The query has no granularity: just apply the catalog granularity.
+        finalGranularity = definedGranularity;
+      } else if (definedGranularity.equals(ingestionGranularity)) {
+        // Both have a setting and they are the same. We assume this would
+        // likely occur only when moving to the catalog, and old queries still
+        // contain the PARTITION BY clause.
+        finalGranularity = definedGranularity;
+      } else {
+        // Both have a setting but they are different. Since the user declared
+        // the grain, using a different one is an error. If the user wants to
+        // vary the grain across different (re)ingestions, then, at present, don't
+        // declare the grain in the catalog.
+        // TODO: allow mismatch
+        throw new IAE(
+            "PARTITIONED BY mismatch. Catalog: [%s], query: [%s]",
+            granularityToSqlString(definedGranularity),
+            granularityToSqlString(ingestionGranularity)
+        );
+      }
+    }
+
+    // Note: though this is the validator, we cheat a bit and write the target
+    // granularity into the query context. Perhaps this step should be done
+    // during conversion, however, we've just worked out the granularity, so we
+    // do it here instead.
+    try {
+      validatorContext.queryContextMap().put(
+          DruidSqlInsert.SQL_INSERT_SEGMENT_GRANULARITY,
+          validatorContext.jsonMapper().writeValueAsString(finalGranularity)
+      );
+    }
+    catch (JsonProcessingException e) {
+      throw new IAE("Invalid PARTITIONED BY granularity");
+    }
+  }
+
+  private String granularityToSqlString(final Granularity gran)
+  {
+    if (gran == null) {
+      return "NULL";
+    }
+    // The validation path will only ever see the ALL granularity or
+    // a period granularity. Neither the parser nor catalog can
+    // create a Duration granularity.
+    if (Granularities.ALL == gran) {
+      return "ALL TIME";
+    }
+    return ((PeriodGranularity) gran).getPeriod().toString();
+  }
+
+  private void ensureNoOrderBy(
+      SqlNode source,
+      final String operationName
+  )
+  {
+    // The source SELECT cannot include an ORDER BY clause. Ordering is given
+    // by the CLUSTERED BY clause, if any.
+    // Check that an ORDER BY clause is not provided by the underlying query
+    SqlNodeList orderByList;
+    if (source instanceof SqlOrderBy) {
+      throw new IAE(
+          "Cannot have ORDER BY on %s %s statement, use CLUSTERED BY instead.",
+          statementArticle(operationName),
+          operationName
+      );
+    }
+
+    // Pull the SELECT statement out of the WITH clause
+    if (source instanceof SqlWith) {
+      source = ((SqlWith) source).getOperandList().get(1);
+    }
+    // If the child of INSERT or WITH is not SELECT, then the statement is not valid.
+    if (!(source instanceof SqlSelect)) {
+      throw new IAE(
+          "%s is not supported within %s %s statement.",
+          source.getKind(),
+          statementArticle(operationName),
+          operationName
+      );
+    }
+
+    // Verify that the SELECT has no ORDER BY clause
+    SqlSelect select = (SqlSelect) source;
+    orderByList = select.getOrderList();
+    if (orderByList != null && orderByList.size() != 0) {
+      throw new IAE(
+          "ORDER BY is not supported within %s %s statement, use CLUSTERED BY instead.",
+          statementArticle(operationName),
+          operationName
+      );
+    }
+  }
+
+  private String statementArticle(final String operationName)
+  {
+    return "INSERT".equals(operationName) ? "an" : "a";
+  }
+
+  private SqlNodeList convertCatalogClustering(final DatasourceFacade tableMetadata)
+  {
+    if (tableMetadata == null) {
+      return null;
+    }
+    final List<ClusterKeySpec> keyCols = tableMetadata.clusterKeys();
+    if (CollectionUtils.isNullOrEmpty(keyCols)) {
+      return null;
+    }
+    final SqlNodeList keyNodes = new SqlNodeList(SqlParserPos.ZERO);
+    for (ClusterKeySpec keyCol : keyCols) {
+      final SqlIdentifier colIdent = new SqlIdentifier(
+          Collections.singletonList(keyCol.expr()),
+          null, SqlParserPos.ZERO,
+          Collections.singletonList(SqlParserPos.ZERO)
+      );
+      final SqlNode keyNode;
+      if (keyCol.desc()) {
+        keyNode = SqlStdOperatorTable.DESC.createCall(SqlParserPos.ZERO, colIdent);
+      } else {
+        keyNode = colIdent;
+      }
+      keyNodes.add(keyNode);
+    }
+    return keyNodes;
+  }
+
+  // This part is a bit sad. By the time we get here, the validator will have created
+  // the ORDER BY namespace if we had a real ORDER BY. We have to "catch up" and do the
+  // work that registerQuery() should have done. That's kind of OK. But, the orderScopes
+  // variable is private, so we have to play dirty tricks to get at it.
+  //
+  // Warning: this may no longer work if Java forbids access to private fields in a
+  // future release.
+  private static final Field ORDER_SCOPES_FIELD;
+
+  static {
+    try {
+      // TODO: this class has changed, and the orderScopes field no longer exists.
+      ORDER_SCOPES_FIELD = FieldUtils.getDeclaredField(
+          SqlValidatorImpl.class,
+          "orderScopes",
+          true
+      );
+    }
+    catch (RuntimeException e) {
+      throw new ISE(e, "SqlValidatorImpl.orderScopes is not accessible");
+    }
+  }
+
+  /**
+   * Clustering is a kind of ordering. We push the CLUSTERED BY clause into the source query as
+   * an ORDER BY clause. In an ideal world, clustering would be outside of SELECT: it is an operation
+   * applied after the data is selected. For now, we push it into the SELECT, then MSQ pulls it back
+   * out. This is unfortunate as errors will be attributed to ORDER BY, which the user did not
+   * actually specify (it is an error to do so.) However, with the current hybrid structure, it is
+   * not possible to add the ORDER by later: doing so requires access to the order by namespace
+   * which is not visible to subclasses.
+   */
+  private void rewriteClusteringToOrderBy(SqlNode source, DruidSqlIngest ingestNode, SqlNodeList catalogClustering)
+  {
+    SqlNodeList clusteredBy = ingestNode.getClusteredBy();
+    if (clusteredBy == null || clusteredBy.getList().isEmpty()) {
+      if (catalogClustering == null || catalogClustering.getList().isEmpty()) {
+        return;
+      }
+      clusteredBy = catalogClustering;
+    }
+    while (source instanceof SqlWith) {
+      source = ((SqlWith) source).getOperandList().get(1);
+    }
+    final SqlSelect select = (SqlSelect) source;
+
+    select.setOrderBy(clusteredBy);
+    final OrderByScope orderScope = ValidatorShim.newOrderByScope(scopes.get(select), clusteredBy, select);
+    try {
+      @SuppressWarnings("unchecked")
+      final Map<SqlSelect, SqlValidatorScope> orderScopes =
+          (Map<SqlSelect, SqlValidatorScope>) ORDER_SCOPES_FIELD.get(this);
+      orderScopes.put(select, orderScope);
+    }
+    catch (Exception e) {
+      throw new ISE(e, "orderScopes is not accessible");
+    }
+  }
+
+  private void validateTimeColumn(
+      final RelRecordType sourceType,
+      final int timeColumnIndex
+  )
+  {
+    final RelDataTypeField timeCol = sourceType.getFieldList().get(timeColumnIndex);
+    final RelDataType timeColType = timeCol.getType();
+    if (timeColType instanceof BasicSqlType) {
+      final BasicSqlType timeColSqlType = (BasicSqlType) timeColType;
+      final SqlTypeName timeColSqlTypeName = timeColSqlType.getSqlTypeName();
+      if (timeColSqlTypeName == SqlTypeName.BIGINT || timeColSqlTypeName == SqlTypeName.TIMESTAMP) {
+        return;
+      }
+    }
+    throw new IAE(
+        "Column [%s] is being used as the time column. It must be of type BIGINT or TIMESTAMP, got [%s]",
+        timeCol.getName(),
+        timeColType
+    );
+  }
+
+  /**
+   * Verify clustering which can come from the query, the catalog or both. If both,
+   * the two must match. In either case, the cluster keys must be present in the SELECT
+   * clause. The {@code __time} column cannot be included.
+   */
+  private void validateClustering(
+      final RelRecordType sourceType,
+      final DruidSqlIngest ingestNode,
+      final SqlNodeList catalogClustering
+  )
+  {
+    final SqlNodeList clusteredBy = ingestNode.getClusteredBy();
+
+    // Validate both the catalog and query definitions if present. This ensures
+    // that things are sane if we later check that the two are identical.
+    if (clusteredBy != null) {
+      validateClusteredBy(sourceType, clusteredBy);
+    }
+    if (catalogClustering != null) {
+      // Catalog defines the key columns. Verify that they are present in the query.
+      validateClusteredBy(sourceType, catalogClustering);
+    }
+    if (clusteredBy != null && catalogClustering != null) {
+      // Both the query and catalog have keys.
+      verifyQueryClusterByMatchesCatalog(sourceType, catalogClustering, clusteredBy);
+    }
+  }
+
+  /**
+   * Validate the CLUSTERED BY list. Members can be any of the following:
+   * <p>
+   * {@code CLUSTERED BY [<ordinal> | <id> | <expr>] DESC?}
+   * <p>
+   * Ensure that each id exists. Ensure each column is included only once.
+   * For an expression, just ensure it is valid; we don't check for duplicates.
+   */
+  private void validateClusteredBy(
+      final RelRecordType sourceType,
+      final SqlNodeList clusteredBy
+  )
+  {
+    // Keep track of fields which have been referenced.
+    final List<String> fieldNames = sourceType.getFieldNames();
+    final int fieldCount = fieldNames.size();
+    final boolean[] refs = new boolean[fieldCount];
+
+    // Process cluster keys
+    for (SqlNode clusterKey : clusteredBy) {
+      final Pair<Integer, Boolean> key = resolveClusterKey(clusterKey, fieldNames);
+      // If an expression, index is null. Validation was done in the ORDER BY check.
+      // Else, do additional MSQ-specific checks.
+      if (key != null) {
+        int index = key.left;
+        // No duplicate references
+        if (refs[index]) {
+          throw new IAE("Duplicate CLUSTERED BY key: [%s]", clusterKey);
+        }
+        refs[index] = true;
+      }
+    }
+  }
+
+  private Pair<Integer, Boolean> resolveClusterKey(SqlNode clusterKey, final List<String> fieldNames)
+  {
+    boolean desc = false;
+
+    // Check if the key is compound: only occurs for DESC. The ASC
+    // case is abstracted away by the parser.
+    if (clusterKey instanceof SqlBasicCall) {
+      SqlBasicCall basicCall = (SqlBasicCall) clusterKey;
+      if (basicCall.getOperator() == SqlStdOperatorTable.DESC) {
+        // Cluster key is compound: CLUSTERED BY foo DESC
+        // We check only the first element
+        clusterKey = ((SqlBasicCall) clusterKey).getOperandList().get(0);
+        desc = true;
+      }
+    }
+
+    // We now have the actual key. Handle the three cases.
+    if (clusterKey instanceof SqlNumericLiteral) {
+      // Key is an ordinal: CLUSTERED BY 2
+      // Ordinals are 1-based.
+      final int ord = ((SqlNumericLiteral) clusterKey).intValue(true);
+      final int index = ord - 1;
+
+      // The ordinal has to be in range.
+      if (index < 0 || fieldNames.size() <= index) {
+        throw new IAE(
+            "CLUSTERED BY ordinal [%d] should be non-negative and <= the number of fields [%d]",
+            ord,
+            fieldNames.size()
+        );
+      }
+      return new Pair<>(index, desc);
+    } else if (clusterKey instanceof SqlIdentifier) {
+      // Key is an identifier: CLUSTERED BY foo
+      final SqlIdentifier key = (SqlIdentifier) clusterKey;
+
+      // Only key of the form foo are allowed, not foo.bar
+      if (!key.isSimple()) {
+        throw new IAE("CLUSTERED BY keys must be a simple name with no dots: [%s]", key.toString());
+      }
+
+      // The name must match an item in the select list
+      final String keyName = key.names.get(0);
+      // Slow linear search. We assume that there are not many cluster keys.
+      final int index = fieldNames.indexOf(keyName);
+      if (index == -1) {
+        throw new IAE("Unknown column [%s] in CLUSTERED BY", keyName);
+      }
+      return new Pair<>(index, desc);
+    } else {
+      // Key is an expression: CLUSTERED BY CEIL(m2)
+      return null;
+    }
+  }
+
+  /**
+   * Both the catalog and query define clustering. This is allowed as long as they
+   * are identical.
+   */
+  private void verifyQueryClusterByMatchesCatalog(
+      final RelRecordType sourceType,
+      final SqlNodeList catalogClustering,
+      final SqlNodeList clusteredBy
+  )
+  {
+    if (clusteredBy.size() != catalogClustering.size()) {
+      throw clusterKeyMismatchException(catalogClustering, clusteredBy);
+    }
+    final List<String> fieldNames = sourceType.getFieldNames();
+    for (int i = 0; i < clusteredBy.size(); i++) {
+      final SqlNode catalogKey = catalogClustering.get(i);
+      final SqlNode clusterKey = clusteredBy.get(i);
+      final Pair<Integer, Boolean> catalogPair = resolveClusterKey(catalogKey, fieldNames);
+      final Pair<Integer, Boolean> queryPair = resolveClusterKey(clusterKey, fieldNames);
+
+      // Cluster keys in the catalog must be field references. If unresolved,
+      // we would have gotten an error above. Here we make sure that both
+      // indexes are the same. Since the catalog index can't be null, we're
+      // essentially checking that the indexes are the same: they name the same
+      // column.
+      if (!Objects.equals(catalogPair, queryPair)) {
+        throw clusterKeyMismatchException(catalogClustering, clusteredBy);
+      }
+    }
+  }
+
+  private RuntimeException clusterKeyMismatchException(SqlNodeList catalogClustering, SqlNodeList clusterKeys)
+  {
+    throw new IAE(
+        "CLUSTER BY mismatch. Catalog: [%s], query: [%s]",
+        catalogClustering,
+        clusterKeys
+    );
+  }
+
+  /**
+   * Compute and validate the target type. In normal SQL, the engine would insert
+   * a project operator after the SELECT before the write to cast columns from the
+   * input type to the (compatible) defined output type. Druid doesn't work that way.
+   * In MSQ, the output the just is the input type. If the user wants to control the
+   * output type, then the user must manually insert any required CAST: Druid is not
+   * in the business of changing the type to suit the catalog.
+   * <p>
+   * As a result, we first propagate column names and types using Druid rules: the
+   * output is exactly what SELECT says it is. We then apply restrictions from the
+   * catalog. If the table is strict, only column names from the catalog can be
+   * used.
+   */
+  private RelDataType validateTargetType(
+      SqlValidatorScope scope,
+      final IdentifierNamespace insertNs,
+      SqlInsert insert,
+      RelRecordType sourceType,
+      DatasourceFacade tableMetadata
+  )
+  {
+    final List<RelDataTypeField> sourceFields = sourceType.getFieldList();
+    for (int i = 0; i < sourceFields.size(); i++) {
+      final RelDataTypeField sourceField = sourceFields.get(i);
+      // Check that there are no unnamed columns in the insert.
+      if (UNNAMED_COLUMN_PATTERN.matcher(sourceField.getName()).matches()) {
+        throw new IAE(
+            "Expression [%d] must provide an alias to specify the target column: func(X) AS myColumn",
+            i + 1
+        );
+      }
+    }
+    if (tableMetadata == null) {
+      return sourceType;
+    }
+    final boolean isStrict = tableMetadata.isSealed();
+    final List<Map.Entry<String, RelDataType>> fields = new ArrayList<>();
+    for (RelDataTypeField sourceField : sourceFields) {
+      final String colName = sourceField.getName();
+      final DatasourceFacade.ColumnFacade definedCol = tableMetadata.column(colName);
+      if (definedCol == null) {
+        // No catalog definition for this column.
+        if (isStrict) {
+          // Table is strict: cannot add new columns at ingest time.
+          throw new IAE(
+              "Column [%s] is not defined in the target table [%s] strict schema",
+              colName,
+              insert.getTargetTable()
+          );
+        }
+
+        // Table is not strict: add a new column based on the SELECT column.
+        fields.add(Pair.of(colName, sourceField.getType()));
+        continue;
+      }
+
+      // If the column name is defined, but no type is given then, use the
+      // column type from SELECT.
+      if (!definedCol.hasType()) {
+        fields.add(Pair.of(colName, sourceField.getType()));
+        continue;
+      }
+
+      // Both the column name and type are provided. Use the name and type
+      // from the catalog.
+      // Note to future readers: this check is preliminary. It works for the
+      // simple column types and has not yet been extended to complex types, aggregates,
+      // types defined in extensions, etc. It may be that SQL
+      // has types that Druid cannot store. This may crop up with types defined in
+      // extensions which are not loaded. Those details are not known at the time
+      // of this code so we are not yet in a position to make the right decision.
+      // This is a task to be revisited when we have more information.
+      final String sqlTypeName = definedCol.sqlStorageType();
+      if (sqlTypeName == null) {
+        // Don't know the storage type. Just skip this one: Druid types are
+        // fluid so let Druid sort out what to store. This is probably not the right
+        // answer, but should avoid problems until full type system support is completed.
+        fields.add(Pair.of(colName, sourceField.getType()));
+        continue;
+      }
+      RelDataType relType = typeFactory.createSqlType(SqlTypeName.get(sqlTypeName));
+      fields.add(Pair.of(
+          colName,
+          typeFactory.createTypeWithNullability(relType, true)
+      ));
+    }
+
+    // Perform the SQL-standard check: that the SELECT column can be
+    // converted to the target type. This check is retained to mimic SQL
+    // behavior, but doesn't do anything because we enforced exact type
+    // matches above.
+    final RelDataType targetType = typeFactory.createStructType(fields);
+    final SqlValidatorTable target = insertNs.resolve().getTable();
+    checkTypeAssignment(scope, target, sourceType, targetType, insert);
+    return targetType;
   }
 
   @Override
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/IngestHandler.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/IngestHandler.java
index 3d38c6b3f2c1..d25018a3bf89 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/IngestHandler.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/IngestHandler.java
@@ -19,29 +19,16 @@
 
 package org.apache.druid.sql.calcite.planner;
 
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.google.common.collect.Iterables;
-import org.apache.calcite.jdbc.CalciteSchema;
 import org.apache.calcite.rel.RelRoot;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.sql.SqlExplain;
 import org.apache.calcite.sql.SqlIdentifier;
-import org.apache.calcite.sql.SqlInsert;
-import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlNode;
-import org.apache.calcite.sql.SqlNodeList;
-import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.validate.SqlValidator;
 import org.apache.calcite.tools.ValidationException;
-import org.apache.calcite.util.Pair;
-import org.apache.druid.common.utils.IdUtils;
-import org.apache.druid.error.DruidException;
 import org.apache.druid.error.InvalidSqlInput;
 import org.apache.druid.java.util.common.granularity.Granularity;
-import org.apache.druid.server.security.Action;
-import org.apache.druid.server.security.Resource;
-import org.apache.druid.server.security.ResourceAction;
-import org.apache.druid.server.security.ResourceType;
 import org.apache.druid.sql.calcite.parser.DruidSqlIngest;
 import org.apache.druid.sql.calcite.parser.DruidSqlInsert;
 import org.apache.druid.sql.calcite.parser.DruidSqlParserUtils;
@@ -50,50 +37,20 @@
 import org.apache.druid.sql.calcite.run.QueryMaker;
 
 import java.util.List;
-import java.util.regex.Pattern;
 
 public abstract class IngestHandler extends QueryHandler
 {
-  private static final Pattern UNNAMED_COLUMN_PATTERN = Pattern.compile("^EXPR\\$\\d+$", Pattern.CASE_INSENSITIVE);
-
-  protected final Granularity ingestionGranularity;
+  protected Granularity ingestionGranularity;
   protected String targetDatasource;
+  private SqlNode validatedQueryNode;
+  private RelDataType targetType;
 
   IngestHandler(
-      HandlerContext handlerContext,
-      DruidSqlIngest ingestNode,
-      SqlNode queryNode,
-      SqlExplain explain
+      final HandlerContext handlerContext,
+      final SqlExplain explain
   )
   {
-    super(handlerContext, queryNode, explain);
-    this.ingestionGranularity = ingestNode.getPartitionedBy();
-    handlerContext.hook().captureInsert(ingestNode);
-  }
-
-  protected static SqlNode convertQuery(DruidSqlIngest sqlNode)
-  {
-    SqlNode query = sqlNode.getSource();
-
-    // Check if ORDER BY clause is not provided to the underlying query
-    if (query instanceof SqlOrderBy) {
-      SqlOrderBy sqlOrderBy = (SqlOrderBy) query;
-      SqlNodeList orderByList = sqlOrderBy.orderList;
-      if (!(orderByList == null || orderByList.equals(SqlNodeList.EMPTY))) {
-        throw InvalidSqlInput.exception(
-            "Cannot use an ORDER BY clause on a Query of type [%s], use CLUSTERED BY instead",
-            sqlNode.getOperator().getName()
-        );
-      }
-    }
-    if (sqlNode.getClusteredBy() != null) {
-      query = DruidSqlParserUtils.convertClusterByToOrderBy(query, sqlNode.getClusteredBy());
-    }
-
-    if (!query.isA(SqlKind.QUERY)) {
-      throw InvalidSqlInput.exception("Unexpected SQL statement type [%s], expected it to be a QUERY", query.getKind());
-    }
-    return query;
+    super(handlerContext, explain);
   }
 
   protected String operationName()
@@ -106,25 +63,6 @@ protected String operationName()
   @Override
   public void validate()
   {
-    if (ingestNode().getPartitionedBy() == null) {
-      throw InvalidSqlInput.exception(
-          "Operation [%s] requires a PARTITIONED BY to be explicitly defined, but none was found.",
-          operationName()
-      );
-    }
-    try {
-      PlannerContext plannerContext = handlerContext.plannerContext();
-      if (ingestionGranularity != null) {
-        plannerContext.queryContextMap().put(
-            DruidSqlInsert.SQL_INSERT_SEGMENT_GRANULARITY,
-            plannerContext.getJsonMapper().writeValueAsString(ingestionGranularity)
-        );
-      }
-    }
-    catch (JsonProcessingException e) {
-      throw InvalidSqlInput.exception(e, "Invalid partition granularity [%s]", ingestionGranularity);
-    }
-    super.validate();
     // Check if CTX_SQL_OUTER_LIMIT is specified and fail the query if it is. CTX_SQL_OUTER_LIMIT being provided causes
     // the number of rows inserted to be limited which is likely to be confusing and unintended.
     if (handlerContext.queryContextMap().get(PlannerContext.CTX_SQL_OUTER_LIMIT) != null) {
@@ -134,8 +72,21 @@ public void validate()
           operationName()
       );
     }
-    targetDatasource = validateAndGetDataSourceForIngest();
-    resourceActions.add(new ResourceAction(new Resource(targetDatasource, ResourceType.DATASOURCE), Action.WRITE));
+    DruidSqlIngest ingestNode = ingestNode();
+    DruidSqlIngest validatedNode = (DruidSqlIngest) validate(ingestNode);
+    validatedQueryNode = validatedNode.getSource();
+    CalcitePlanner planner = handlerContext.planner();
+    final SqlValidator validator = planner.getValidator();
+    targetType = validator.getValidatedNodeType(validatedNode);
+    ingestionGranularity = DruidSqlParserUtils.convertSqlNodeToGranularity(ingestNode().getPartitionedBy());
+    final SqlIdentifier tableIdentifier = (SqlIdentifier) ingestNode().getTargetTable();
+    targetDatasource = tableIdentifier.names.get(tableIdentifier.names.size() - 1);
+  }
+
+  @Override
+  protected SqlNode validatedQueryNode()
+  {
+    return validatedQueryNode;
   }
 
   @Override
@@ -148,57 +99,6 @@ protected RelDataType returnedRowType()
     );
   }
 
-  /**
-   * Extract target datasource from a {@link SqlInsert}, and also validate that the ingestion is of a form we support.
-   * Expects the target datasource to be either an unqualified name, or a name qualified by the default schema.
-   */
-  private String validateAndGetDataSourceForIngest()
-  {
-    final SqlInsert insert = ingestNode();
-    if (insert.isUpsert()) {
-      throw InvalidSqlInput.exception("UPSERT is not supported.");
-    }
-
-    if (insert.getTargetColumnList() != null) {
-      throw InvalidSqlInput.exception(
-          "Operation [%s] cannot be run with a target column list, given [%s (%s)]",
-          operationName(),
-          insert.getTargetTable(), insert.getTargetColumnList()
-      );
-    }
-
-    final SqlIdentifier tableIdentifier = (SqlIdentifier) insert.getTargetTable();
-    final String dataSource;
-
-    if (tableIdentifier.names.isEmpty()) {
-      // I don't think this can happen, but include a branch for it just in case.
-      throw DruidException.forPersona(DruidException.Persona.USER)
-          .ofCategory(DruidException.Category.DEFENSIVE)
-          .build("Operation [%s] requires a target table", operationName());
-    } else if (tableIdentifier.names.size() == 1) {
-      // Unqualified name.
-      dataSource = Iterables.getOnlyElement(tableIdentifier.names);
-    } else {
-      // Qualified name.
-      final String defaultSchemaName =
-          Iterables.getOnlyElement(CalciteSchema.from(handlerContext.defaultSchema()).path(null));
-
-      if (tableIdentifier.names.size() == 2 && defaultSchemaName.equals(tableIdentifier.names.get(0))) {
-        dataSource = tableIdentifier.names.get(1);
-      } else {
-        throw InvalidSqlInput.exception(
-            "Table [%s] does not support operation [%s] because it is not a Druid datasource",
-            tableIdentifier,
-            operationName()
-        );
-      }
-    }
-
-    IdUtils.validateId("table", dataSource);
-
-    return dataSource;
-  }
-
   @Override
   protected PlannerResult planForDruid() throws ValidationException
   {
@@ -208,48 +108,33 @@ protected PlannerResult planForDruid() throws ValidationException
   @Override
   protected QueryMaker buildQueryMaker(final RelRoot rootQueryRel) throws ValidationException
   {
-    validateColumnsForIngestion(rootQueryRel);
+    final SqlIdentifier tableIdentifier = (SqlIdentifier) ingestNode().getTargetTable();
+    String targetDatasource = tableIdentifier.names.get(tableIdentifier.names.size() - 1);
     return handlerContext.engine().buildQueryMakerForInsert(
         targetDatasource,
         rootQueryRel,
-        handlerContext.plannerContext()
+        handlerContext.plannerContext(),
+        targetType
     );
   }
 
-  private void validateColumnsForIngestion(RelRoot rootQueryRel)
-  {
-    // Check that there are no unnamed columns in the insert.
-    for (Pair<Integer, String> field : rootQueryRel.fields) {
-      if (UNNAMED_COLUMN_PATTERN.matcher(field.right).matches()) {
-        throw InvalidSqlInput.exception(
-            "Insertion requires columns to be named, but at least one of the columns was unnamed.  This is usually "
-            + "the result of applying a function without having an AS clause, please ensure that all function calls"
-            + "are named with an AS clause as in \"func(X) as myColumn\"."
-        );
-      }
-    }
-  }
-
   /**
    * Handler for the INSERT statement.
    */
   protected static class InsertHandler extends IngestHandler
   {
+
     private final DruidSqlInsert sqlNode;
 
     public InsertHandler(
         SqlStatementHandler.HandlerContext handlerContext,
-        DruidSqlInsert sqlNode,
+        DruidSqlInsert insertNode,
         SqlExplain explain
     )
     {
-      super(
-          handlerContext,
-          sqlNode,
-          convertQuery(sqlNode),
-          explain
-      );
-      this.sqlNode = sqlNode;
+      super(handlerContext, explain);
+      this.sqlNode = insertNode;
+      handlerContext.hook().captureInsert(insertNode);
     }
 
     @Override
@@ -274,7 +159,7 @@ public void validate()
     public ExplainAttributes explainAttributes()
     {
       return new ExplainAttributes(
-          DruidSqlInsert.OPERATOR.getName(),
+          DruidSqlIngestOperator.INSERT_OPERATOR.getName(),
           targetDatasource,
           ingestionGranularity,
           DruidSqlParserUtils.resolveClusteredByColumnsToOutputColumns(sqlNode.getClusteredBy(), rootQueryRel.fields),
@@ -297,12 +182,7 @@ public ReplaceHandler(
         SqlExplain explain
     )
     {
-      super(
-          handlerContext,
-          sqlNode,
-          convertQuery(sqlNode),
-          explain
-      );
+      super(handlerContext, explain);
       this.sqlNode = sqlNode;
     }
 
@@ -348,7 +228,7 @@ public void validate()
     public ExplainAttributes explainAttributes()
     {
       return new ExplainAttributes(
-          DruidSqlReplace.OPERATOR.getName(),
+          DruidSqlIngestOperator.REPLACE_OPERATOR.getName(),
           targetDatasource,
           ingestionGranularity,
           DruidSqlParserUtils.resolveClusteredByColumnsToOutputColumns(sqlNode.getClusteredBy(), rootQueryRel.fields),
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/QueryHandler.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/QueryHandler.java
index 357ec61c6cdf..3782fa0edabc 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/QueryHandler.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/QueryHandler.java
@@ -93,27 +93,24 @@ public abstract class QueryHandler extends SqlStatementHandler.BaseStatementHand
 {
   static final EmittingLogger log = new EmittingLogger(QueryHandler.class);
 
-  protected SqlNode queryNode;
   protected SqlExplain explain;
-  protected SqlNode validatedQueryNode;
   private boolean isPrepared;
   protected RelRoot rootQueryRel;
   private PrepareResult prepareResult;
   protected RexBuilder rexBuilder;
 
-  public QueryHandler(SqlStatementHandler.HandlerContext handlerContext, SqlNode sqlNode, SqlExplain explain)
+  public QueryHandler(SqlStatementHandler.HandlerContext handlerContext, SqlExplain explain)
   {
     super(handlerContext);
-    this.queryNode = sqlNode;
     this.explain = explain;
   }
 
-  @Override
-  public void validate()
+  protected SqlNode validate(SqlNode root)
   {
     CalcitePlanner planner = handlerContext.planner();
+    SqlNode validatedQueryNode;
     try {
-      validatedQueryNode = planner.validate(rewriteParameters());
+      validatedQueryNode = planner.validate(rewriteParameters(root));
     }
     catch (ValidationException e) {
       throw DruidPlanner.translateException(e);
@@ -126,9 +123,12 @@ public void validate()
     );
     validatedQueryNode.accept(resourceCollectorShuttle);
     resourceActions = resourceCollectorShuttle.getResourceActions();
+    return validatedQueryNode;
   }
 
-  private SqlNode rewriteParameters()
+  protected abstract SqlNode validatedQueryNode();
+
+  private SqlNode rewriteParameters(SqlNode original)
   {
     // Uses {@link SqlParameterizerShuttle} to rewrite {@link SqlNode} to swap out any
     // {@link org.apache.calcite.sql.SqlDynamicParam} early for their {@link SqlLiteral}
@@ -140,9 +140,9 @@ private SqlNode rewriteParameters()
     // contains parameters, but no values were provided.
     PlannerContext plannerContext = handlerContext.plannerContext();
     if (plannerContext.getParameters().isEmpty()) {
-      return queryNode;
+      return original;
     } else {
-      return queryNode.accept(new SqlParameterizerShuttle(plannerContext));
+      return original.accept(new SqlParameterizerShuttle(plannerContext));
     }
   }
 
@@ -153,6 +153,7 @@ public void prepare()
       return;
     }
     isPrepared = true;
+    SqlNode validatedQueryNode = validatedQueryNode();
     rootQueryRel = handlerContext.planner().rel(validatedQueryNode);
     handlerContext.hook().captureQueryRel(rootQueryRel);
     final RelDataTypeFactory typeFactory = rootQueryRel.rel.getCluster().getTypeFactory();
@@ -337,12 +338,12 @@ private PlannerResult planWithBindableConvention()
         final Enumerator<?> enumerator = enumerable.enumerator();
         return QueryResponse.withEmptyContext(
             Sequences.withBaggage(new BaseSequence<>(
-                new BaseSequence.IteratorMaker<Object[], QueryHandler.EnumeratorIterator<Object[]>>()
+                new BaseSequence.IteratorMaker<Object[], Iterator<Object[]>>()
                 {
                   @Override
-                  public QueryHandler.EnumeratorIterator<Object[]> make()
+                  public Iterator<Object[]> make()
                   {
-                    return new QueryHandler.EnumeratorIterator<>(new Iterator<Object[]>()
+                    return new Iterator<Object[]>()
                     {
                       @Override
                       public boolean hasNext()
@@ -355,11 +356,11 @@ public Object[] next()
                       {
                         return (Object[]) enumerator.current();
                       }
-                    });
+                    };
                   }
 
                   @Override
-                  public void cleanup(QueryHandler.EnumeratorIterator<Object[]> iterFromMake)
+                  public void cleanup(Iterator<Object[]> iterFromMake)
                   {
 
                   }
@@ -401,7 +402,7 @@ protected PlannerResult planExplanation(
           .map(ResourceAction::getResource)
           .sorted(Comparator.comparing(Resource::getName))
           .collect(Collectors.toList());
-      resourcesString = plannerContext.getJsonMapper().writeValueAsString(resources);
+      resourcesString = plannerContext.getPlannerToolbox().jsonMapper().writeValueAsString(resources);
     }
     catch (JsonProcessingException jpe) {
       // this should never happen, we create the Resources here, not a user
@@ -701,72 +702,4 @@ private DruidException buildSQLPlanningError(RelOptPlanner.CannotPlanException e
                           );
     }
   }
-
-  public static class SelectHandler extends QueryHandler
-  {
-    public SelectHandler(
-        HandlerContext handlerContext,
-        SqlNode sqlNode,
-        SqlExplain explain
-    )
-    {
-      super(handlerContext, sqlNode, explain);
-    }
-
-    @Override
-    public void validate()
-    {
-      if (!handlerContext.plannerContext().featureAvailable(EngineFeature.CAN_SELECT)) {
-        throw InvalidSqlInput.exception("Cannot execute SELECT with SQL engine [%s]", handlerContext.engine().name());
-      }
-      super.validate();
-    }
-
-    @Override
-    protected RelDataType returnedRowType()
-    {
-      final RelDataTypeFactory typeFactory = rootQueryRel.rel.getCluster().getTypeFactory();
-      return handlerContext.engine().resultTypeForSelect(
-          typeFactory,
-          rootQueryRel.validatedRowType
-      );
-    }
-
-    @Override
-    protected PlannerResult planForDruid() throws ValidationException
-    {
-      return planWithDruidConvention();
-    }
-
-    @Override
-    protected QueryMaker buildQueryMaker(final RelRoot rootQueryRel) throws ValidationException
-    {
-      return handlerContext.engine().buildQueryMakerForSelect(
-          rootQueryRel,
-          handlerContext.plannerContext()
-      );
-    }
-  }
-
-  private static class EnumeratorIterator<T> implements Iterator<T>
-  {
-    private final Iterator<T> it;
-
-    EnumeratorIterator(Iterator<T> it)
-    {
-      this.it = it;
-    }
-
-    @Override
-    public boolean hasNext()
-    {
-      return it.hasNext();
-    }
-
-    @Override
-    public T next()
-    {
-      return it.next();
-    }
-  }
 }
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/SelectHandler.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/SelectHandler.java
new file mode 100644
index 000000000000..45c999b1eb02
--- /dev/null
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/SelectHandler.java
@@ -0,0 +1,85 @@
+/*
+ * 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.druid.sql.calcite.planner;
+
+import org.apache.calcite.rel.RelRoot;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.sql.SqlExplain;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.tools.ValidationException;
+import org.apache.druid.error.InvalidSqlInput;
+import org.apache.druid.sql.calcite.run.EngineFeature;
+import org.apache.druid.sql.calcite.run.QueryMaker;
+
+public class SelectHandler extends QueryHandler
+{
+  private final SqlNode queryNode;
+  private SqlNode validatedQueryNode;
+
+  public SelectHandler(
+      HandlerContext handlerContext,
+      SqlNode sqlNode,
+      SqlExplain explain
+  )
+  {
+    super(handlerContext, explain);
+    this.queryNode = sqlNode;
+  }
+
+  @Override
+  public void validate()
+  {
+    if (!handlerContext.plannerContext().featureAvailable(EngineFeature.CAN_SELECT)) {
+      throw InvalidSqlInput.exception("Cannot execute SELECT with SQL engine [%s]", handlerContext.engine().name());
+    }
+    validatedQueryNode = validate(queryNode);
+  }
+
+  @Override
+  protected SqlNode validatedQueryNode()
+  {
+    return validatedQueryNode;
+  }
+
+  @Override
+  protected RelDataType returnedRowType()
+  {
+    final RelDataTypeFactory typeFactory = rootQueryRel.rel.getCluster().getTypeFactory();
+    return handlerContext.engine().resultTypeForSelect(
+        typeFactory,
+        rootQueryRel.validatedRowType
+    );
+  }
+
+  @Override
+  protected PlannerResult planForDruid() throws ValidationException
+  {
+    return planWithDruidConvention();
+  }
+
+  @Override
+  protected QueryMaker buildQueryMaker(final RelRoot rootQueryRel) throws ValidationException
+  {
+    return handlerContext.engine().buildQueryMakerForSelect(
+        rootQueryRel,
+        handlerContext.plannerContext());
+  }
+}
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/SqlResourceCollectorShuttle.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/SqlResourceCollectorShuttle.java
index 850b472bc83c..c95c5573ab2b 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/SqlResourceCollectorShuttle.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/SqlResourceCollectorShuttle.java
@@ -20,10 +20,13 @@
 package org.apache.druid.sql.calcite.planner;
 
 import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlFunction;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.util.SqlShuttle;
 import org.apache.calcite.sql.validate.IdentifierNamespace;
+import org.apache.calcite.sql.validate.SqlUserDefinedTableMacro;
 import org.apache.calcite.sql.validate.SqlValidator;
 import org.apache.calcite.sql.validate.SqlValidatorNamespace;
 import org.apache.calcite.sql.validate.SqlValidatorTable;
@@ -42,7 +45,7 @@
  * Walks an {@link SqlNode} to collect a set of {@link Resource} for {@link ResourceType#DATASOURCE} and
  * {@link ResourceType#VIEW} to use for authorization during query planning.
  *
- * It works by looking for {@link SqlIdentifier} which corespond to a {@link IdentifierNamespace}, where
+ * It works by looking for {@link SqlIdentifier} which correspond to a {@link IdentifierNamespace}, where
  * {@link SqlValidatorNamespace} is calcite-speak for sources of data and {@link IdentifierNamespace} specifically are
  * namespaces which are identified by a single variable, e.g. table names.
  */
@@ -62,11 +65,19 @@ public SqlResourceCollectorShuttle(SqlValidator validator, PlannerContext planne
   @Override
   public SqlNode visit(SqlCall call)
   {
-    if (call.getOperator() instanceof AuthorizableOperator) {
-      resourceActions.addAll(((AuthorizableOperator) call.getOperator()).computeResources(
+    SqlOperator operator = call.getOperator();
+    if (operator instanceof AuthorizableOperator) {
+      resourceActions.addAll(((AuthorizableOperator) operator).computeResources(
           call,
           plannerContext.getPlannerToolbox().getAuthConfig().isEnableInputSourceSecurity()
       ));
+    } else if (operator instanceof SqlUserDefinedTableMacro) {
+      // This case is unfortunate: we have a table macro inside of a Calcite
+      // "user-defined" table macro. The Calcite object won't let us access the Druid
+      // object which could give us permissions. So, we have to reverse-engineer permissions
+      // from all we are allowed to see, which is the identifier.
+      final SqlIdentifier id = ((SqlFunction) operator).getSqlIdentifier();
+      visitIdentifier(id.names);
     }
 
     return super.visit(call);
@@ -82,24 +93,28 @@ public SqlNode visit(SqlIdentifier id)
       SqlValidatorTable validatorTable = namespace.getTable();
       // this should not probably be null if the namespace was not null,
       if (validatorTable != null) {
-        List<String> qualifiedNameParts = validatorTable.getQualifiedName();
-        // 'schema'.'identifier'
-        if (qualifiedNameParts.size() == 2) {
-          final String schema = qualifiedNameParts.get(0);
-          final String resourceName = qualifiedNameParts.get(1);
-          final String resourceType = plannerContext.getSchemaResourceType(schema, resourceName);
-          if (resourceType != null) {
-            resourceActions.add(new ResourceAction(new Resource(resourceName, resourceType), Action.READ));
-          }
-        } else if (qualifiedNameParts.size() > 2) {
-          // Don't expect to see more than 2 names (catalog?).
-          throw new ISE("Cannot analyze table idetifier %s", qualifiedNameParts);
-        }
+        visitIdentifier(validatorTable.getQualifiedName());
       }
     }
     return super.visit(id);
   }
 
+  private void visitIdentifier(List<String> qualifiedNameParts)
+  {
+    // 'schema'.'identifier'
+    if (qualifiedNameParts.size() == 2) {
+      final String schema = qualifiedNameParts.get(0);
+      final String resourceName = qualifiedNameParts.get(1);
+      final String resourceType = plannerContext.getSchemaResourceType(schema, resourceName);
+      if (resourceType != null) {
+        resourceActions.add(new ResourceAction(new Resource(resourceName, resourceType), Action.READ));
+      }
+    } else if (qualifiedNameParts.size() > 2) {
+      // Don't expect to see more than 2 names (catalog?).
+      throw new ISE("Cannot analyze table identifier %s", qualifiedNameParts);
+    }
+  }
+
   public Set<ResourceAction> getResourceActions()
   {
     return resourceActions;
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/planner/SqlStatementHandler.java b/sql/src/main/java/org/apache/druid/sql/calcite/planner/SqlStatementHandler.java
index 267feae1f0f3..ebf57f771b3a 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/planner/SqlStatementHandler.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/planner/SqlStatementHandler.java
@@ -55,6 +55,7 @@ interface HandlerContext
     SchemaPlus defaultSchema();
     ObjectMapper jsonMapper();
     DateTimeZone timeZone();
+    CatalogResolver catalog();
     PlannerHook hook();
   }
 
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidRel.java b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidRel.java
index ac6c08547494..eefe90ca8c6b 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidRel.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/rel/DruidRel.java
@@ -55,6 +55,11 @@ public QueryResponse<Object[]> runQuery()
     return getPlannerContext().getQueryMaker().runQuery(toDruidQuery(false));
   }
 
+  public Object dryRun()
+  {
+    return getPlannerContext().getQueryMaker().explain(toDruidQuery(false));
+  }
+
   public abstract T withPartialQuery(PartialDruidQuery newQueryBuilder);
 
   public boolean isValidDruidQuery()
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/run/NativeSqlEngine.java b/sql/src/main/java/org/apache/druid/sql/calcite/run/NativeSqlEngine.java
index 164e02a0ca8d..8f72619904ca 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/run/NativeSqlEngine.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/run/NativeSqlEngine.java
@@ -135,7 +135,8 @@ public QueryMaker buildQueryMakerForSelect(final RelRoot relRoot, final PlannerC
   public QueryMaker buildQueryMakerForInsert(
       final String targetDataSource,
       final RelRoot relRoot,
-      final PlannerContext plannerContext
+      final PlannerContext plannerContext,
+      final RelDataType targetType
   )
   {
     throw new UnsupportedOperationException();
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/run/QueryMaker.java b/sql/src/main/java/org/apache/druid/sql/calcite/run/QueryMaker.java
index 8039d60c3e75..2605eb0db5ec 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/run/QueryMaker.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/run/QueryMaker.java
@@ -19,6 +19,7 @@
 
 package org.apache.druid.sql.calcite.run;
 
+import com.google.common.collect.ImmutableMap;
 import org.apache.druid.server.QueryResponse;
 import org.apache.druid.sql.calcite.rel.DruidQuery;
 
@@ -28,6 +29,17 @@
  */
 public interface QueryMaker
 {
+  /**
+   * Do everything that would be done to run a query, don't actually run.
+   * Instead return what would have been sent to the execution engine.
+   * The result is a Jackson-serializable query plan.
+   */
+  default Object explain(DruidQuery druidQuery)
+  {
+    // Temporary to ensure extensions don't break.
+    return ImmutableMap.of("type", "unsupported");
+  }
+
   /**
    * Executes a given Druid query, which is expected to correspond to the SQL query that this QueryMaker was originally
    * created for. The returned arrays match the row type given by {@link SqlEngine#resultTypeForSelect} or
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/run/SqlEngine.java b/sql/src/main/java/org/apache/druid/sql/calcite/run/SqlEngine.java
index 678ded23e9da..a7ae6dcb516a 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/run/SqlEngine.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/run/SqlEngine.java
@@ -94,6 +94,7 @@ public interface SqlEngine
   QueryMaker buildQueryMakerForInsert(
       String targetDataSource,
       RelRoot relRoot,
-      PlannerContext plannerContext
+      PlannerContext plannerContext,
+      RelDataType targetType
   ) throws ValidationException;
 }
diff --git a/sql/src/main/java/org/apache/druid/sql/calcite/view/ViewSqlEngine.java b/sql/src/main/java/org/apache/druid/sql/calcite/view/ViewSqlEngine.java
index e2ce813a37f7..fb28ddb97553 100644
--- a/sql/src/main/java/org/apache/druid/sql/calcite/view/ViewSqlEngine.java
+++ b/sql/src/main/java/org/apache/druid/sql/calcite/view/ViewSqlEngine.java
@@ -109,7 +109,12 @@ public QueryMaker buildQueryMakerForSelect(RelRoot relRoot, PlannerContext plann
   }
 
   @Override
-  public QueryMaker buildQueryMakerForInsert(String targetDataSource, RelRoot relRoot, PlannerContext plannerContext)
+  public QueryMaker buildQueryMakerForInsert(
+      final String targetDataSource,
+      final RelRoot relRoot,
+      final PlannerContext plannerContext,
+      final RelDataType targetType
+  )
   {
     // Can't have views of INSERT or REPLACE statements.
     throw new UnsupportedOperationException();
diff --git a/sql/src/main/java/org/apache/druid/sql/guice/SqlModule.java b/sql/src/main/java/org/apache/druid/sql/guice/SqlModule.java
index 56d0d2d5d41f..9df1016e5f26 100644
--- a/sql/src/main/java/org/apache/druid/sql/guice/SqlModule.java
+++ b/sql/src/main/java/org/apache/druid/sql/guice/SqlModule.java
@@ -41,7 +41,6 @@
 import org.apache.druid.sql.calcite.aggregation.SqlAggregationModule;
 import org.apache.druid.sql.calcite.expression.builtin.QueryLookupOperatorConversion;
 import org.apache.druid.sql.calcite.planner.CalcitePlannerModule;
-import org.apache.druid.sql.calcite.planner.CatalogResolver;
 import org.apache.druid.sql.calcite.planner.PlannerFactory;
 import org.apache.druid.sql.calcite.run.NativeSqlEngine;
 import org.apache.druid.sql.calcite.schema.DruidCalciteSchemaModule;
@@ -121,9 +120,6 @@ public void configure(Binder binder)
     if (isAvaticaEnabled()) {
       binder.install(new AvaticaModule());
     }
-
-    // Default do-nothing catalog resolver
-    binder.bind(CatalogResolver.class).toInstance(CatalogResolver.NULL_RESOLVER);
   }
 
   private boolean isEnabled()
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java
index a34c93ce9c92..fc65ea0fb89a 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteInsertDmlTest.java
@@ -1638,7 +1638,6 @@ public void testErrorWithUnableToConstructColumnSignatureWithExtern()
                              + "partitioned by DAY\n"
                              + "clustered by channel";
     HashMap<String, Object> context = new HashMap<>(DEFAULT_CONTEXT);
-    context.put(PlannerContext.CTX_SQL_OUTER_LIMIT, 100);
     testIngestionQuery().context(context).sql(sqlString)
                         .expectValidationError(
                             new DruidExceptionMatcher(
@@ -1667,7 +1666,6 @@ public void testErrorWhenBothRowSignatureAndExtendsProvidedToExtern()
                              + "partitioned by DAY\n"
                              + "clustered by channel";
     HashMap<String, Object> context = new HashMap<>(DEFAULT_CONTEXT);
-    context.put(PlannerContext.CTX_SQL_OUTER_LIMIT, 100);
     testIngestionQuery().context(context).sql(sqlString)
                         .expectValidationError(
                             new DruidExceptionMatcher(
@@ -1695,7 +1693,6 @@ public void testErrorWhenNoneOfRowSignatureAndExtendsProvidedToExtern()
                              + "partitioned by DAY\n"
                              + "clustered by channel";
     HashMap<String, Object> context = new HashMap<>(DEFAULT_CONTEXT);
-    context.put(PlannerContext.CTX_SQL_OUTER_LIMIT, 100);
     testIngestionQuery().context(context).sql(sqlString)
                         .expectValidationError(
                             new DruidExceptionMatcher(
@@ -1724,7 +1721,6 @@ public void testErrorWhenInputSourceInvalid()
                              + "partitioned by DAY\n"
                              + "clustered by channel";
     HashMap<String, Object> context = new HashMap<>(DEFAULT_CONTEXT);
-    context.put(PlannerContext.CTX_SQL_OUTER_LIMIT, 100);
     testIngestionQuery().context(context).sql(sqlString)
                         .expectValidationError(
                             new DruidExceptionMatcher(
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteScanSignatureTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteScanSignatureTest.java
index 424a2e8895e9..b21730807ea6 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/CalciteScanSignatureTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/CalciteScanSignatureTest.java
@@ -168,7 +168,12 @@ public QueryMaker buildQueryMakerForSelect(RelRoot relRoot, PlannerContext plann
     }
 
     @Override
-    public QueryMaker buildQueryMakerForInsert(String targetDataSource, RelRoot relRoot, PlannerContext plannerContext)
+    public QueryMaker buildQueryMakerForInsert(
+        final String targetDataSource,
+        final RelRoot relRoot,
+        final PlannerContext plannerContext,
+        final RelDataType targetType
+    )
     {
       throw new UnsupportedOperationException();
     }
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/IngestionTestSqlEngine.java b/sql/src/test/java/org/apache/druid/sql/calcite/IngestionTestSqlEngine.java
index b0bf0bd7b29d..2b6fd9cfda18 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/IngestionTestSqlEngine.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/IngestionTestSqlEngine.java
@@ -104,7 +104,12 @@ public QueryMaker buildQueryMakerForSelect(RelRoot relRoot, PlannerContext plann
   }
 
   @Override
-  public QueryMaker buildQueryMakerForInsert(String targetDataSource, RelRoot relRoot, PlannerContext plannerContext)
+  public QueryMaker buildQueryMakerForInsert(
+      final String targetDataSource,
+      final RelRoot relRoot,
+      final PlannerContext plannerContext,
+      final RelDataType targetType
+  )
   {
     final RowSignature signature = RowSignatures.fromRelDataType(
         relRoot.validatedRowType.getFieldNames(),
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/parser/DruidSqlParserUtilsTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/parser/DruidSqlParserUtilsTest.java
index 5d9524b08eb7..718a5c0b5ce5 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/parser/DruidSqlParserUtilsTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/parser/DruidSqlParserUtilsTest.java
@@ -133,6 +133,26 @@ public void testGetGranularityFromFloor() throws ParseException
       Granularity actualGranularity = DruidSqlParserUtils.convertSqlNodeToGranularityThrowingParseExceptions(floorCall);
       Assert.assertEquals(expectedGranularity, actualGranularity);
     }
+
+    /**
+     * Tests clause like "PARTITIONED BY 'day'"
+     * @throws ParseException
+     */
+    @Test
+    public void testConvertSqlNodeToGranularityAsLiteral() throws ParseException
+    {
+      SqlNode sqlNode = SqlLiteral.createCharString(timeUnit.name(), SqlParserPos.ZERO);
+      Granularity actualGranularity = DruidSqlParserUtils.convertSqlNodeToGranularityThrowingParseExceptions(sqlNode);
+      Assert.assertEquals(expectedGranularity, actualGranularity);
+    }
+
+    @Test
+    public void testConvertSqlNodeToGranularityAsSymbol() throws ParseException
+    {
+      SqlNode sqlNode = SqlLiteral.createSymbol(DruidSqlParserUtils.GranularityGrain.fromValue(timeUnit.name()), SqlParserPos.ZERO);
+      Granularity actualGranularity = DruidSqlParserUtils.convertSqlNodeToGranularityThrowingParseExceptions(sqlNode);
+      Assert.assertEquals(expectedGranularity, actualGranularity);
+    }
   }
 
   /**
@@ -305,28 +325,6 @@ public void testClusteredByColumnsWithNegativeOrdinalThrowsException()
 
   public static class FloorToGranularityConversionErrorsTest
   {
-    /**
-     * Tests clause like "PARTITIONED BY 'day'"
-     */
-    @Test
-    public void testConvertSqlNodeToGranularityWithIncorrectNode()
-    {
-      SqlNode sqlNode = SqlLiteral.createCharString("day", SqlParserPos.ZERO);
-      DruidException e = Assert.assertThrows(
-          DruidException.class,
-          () -> DruidSqlParserUtils.convertSqlNodeToGranularityThrowingParseExceptions(sqlNode)
-      );
-      MatcherAssert.assertThat(
-          e,
-          DruidExceptionMatcher
-              .invalidSqlInput()
-              .expectMessageIs(
-                  "Invalid granularity ['day'] after PARTITIONED BY.  "
-                  + "Expected HOUR, DAY, MONTH, YEAR, ALL TIME, FLOOR() or TIME_FLOOR()"
-              )
-      );
-    }
-
     /**
      * Tests clause like "PARTITIONED BY CEIL(__time TO DAY)"
      */
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/planner/CalcitePlannerModuleTest.java b/sql/src/test/java/org/apache/druid/sql/calcite/planner/CalcitePlannerModuleTest.java
index 12db32d4f019..0c55d8bb72ba 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/planner/CalcitePlannerModuleTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/planner/CalcitePlannerModuleTest.java
@@ -131,7 +131,6 @@ public void onMatch(RelOptRuleCall call)
           binder.bind(Key.get(new TypeLiteral<Set<SqlOperatorConversion>>() {})).toInstance(operatorConversions);
           binder.bind(DruidSchemaCatalog.class).toInstance(rootSchema);
           binder.bind(JoinableFactoryWrapper.class).toInstance(joinableFactoryWrapper);
-          binder.bind(CatalogResolver.class).toInstance(CatalogResolver.NULL_RESOLVER);
         },
         target,
         binder -> {
diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerCaptureHook.java b/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerCaptureHook.java
index bf69e1bf162b..d806036eb80f 100644
--- a/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerCaptureHook.java
+++ b/sql/src/test/java/org/apache/druid/sql/calcite/planner/PlannerCaptureHook.java
@@ -19,6 +19,8 @@
 
 package org.apache.druid.sql.calcite.planner;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.calcite.interpreter.BindableRel;
 import org.apache.calcite.rel.RelRoot;
 import org.apache.calcite.rel.type.RelDataType;
@@ -33,6 +35,8 @@ public class PlannerCaptureHook implements PlannerHook
   private RelRoot relRoot;
   private SqlInsert insertNode;
   private SqlNode sqlNode;
+  private DruidRel<?> druidRel;
+  private Object execPlan;
 
   @Override
   public void captureSql(String sql)
@@ -55,7 +59,8 @@ public void captureQueryRel(RelRoot rootQueryRel)
   @Override
   public void captureDruidRel(DruidRel<?> druidRel)
   {
-    // Not used at present. Add a field to capture this if you need it.
+    this.druidRel = druidRel;
+    this.execPlan = druidRel.dryRun();
   }
 
   @Override
@@ -90,4 +95,30 @@ public SqlNode getSqlNode()
   {
     return sqlNode;
   }
+
+  public DruidRel<?> druidRel()
+  {
+    return druidRel;
+  }
+
+  public Object execPlan()
+  {
+    return execPlan;
+  }
+
+  public String visualizePlan(ObjectMapper objectMapper)
+  {
+    // Whatever the plan is: convert it to pretty-printed JSON.
+    if (execPlan == null) {
+      return "null";
+    }
+    try {
+      return objectMapper
+          .writerWithDefaultPrettyPrinter()
+          .writeValueAsString(execPlan);
+    }
+    catch (JsonProcessingException e) {
+      throw new RuntimeException("JSON conversion failed", e);
+    }
+  }
 }
diff --git a/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java b/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java
index dfd366250457..0d029e099d9a 100644
--- a/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java
+++ b/sql/src/test/java/org/apache/druid/sql/guice/SqlModuleTest.java
@@ -64,7 +64,6 @@
 import org.apache.druid.server.security.AuthorizerMapper;
 import org.apache.druid.server.security.Escalator;
 import org.apache.druid.server.security.NoopEscalator;
-import org.apache.druid.sql.calcite.planner.CatalogResolver;
 import org.apache.druid.sql.calcite.planner.PlannerFactory;
 import org.apache.druid.sql.calcite.util.CalciteTests;
 import org.apache.druid.sql.calcite.view.DruidViewMacro;
@@ -209,7 +208,6 @@ private Injector makeInjectorWithProperties(final Properties props)
                     .toProvider(QuerySchedulerProvider.class)
                     .in(LazySingleton.class);
               binder.bind(ResponseContextConfig.class).toInstance(SqlResourceTest.TEST_RESPONSE_CONTEXT_CONFIG);
-              binder.bind(CatalogResolver.class).toInstance(CatalogResolver.NULL_RESOLVER);
               binder.bind(OverlordClient.class).to(NoopOverlordClient.class);
               binder.bind(CoordinatorClient.class).to(NoopCoordinatorClient.class);
             },
diff --git a/sql/src/test/resources/calcite/expected/ingest/InsertFromExternalWithoutSecuritySupport-logicalPlan.txt b/sql/src/test/resources/calcite/expected/ingest/InsertFromExternalWithoutSecuritySupport-logicalPlan.txt
index cc477164c6d8..d392d8b3d01f 100644
--- a/sql/src/test/resources/calcite/expected/ingest/InsertFromExternalWithoutSecuritySupport-logicalPlan.txt
+++ b/sql/src/test/resources/calcite/expected/ingest/InsertFromExternalWithoutSecuritySupport-logicalPlan.txt
@@ -1,3 +1,3 @@
-LogicalInsert(target=[dst], partitionedBy=[AllGranularity], clusteredBy=[<none>])
+LogicalInsert(target=[dst], partitionedBy=[ALL TIME], clusteredBy=[<none>])
   LogicalProject(inputs=[0..2])
     ExternalTableScan(dataSource=[{"type":"external","inputSource":{"type":"CalciteIngestionDmlTest$TestFileInputSource","files":["/tmp/foo.csv"]},"inputFormat":{"type":"csv","columns":["x","y","z"]},"signature":[{"name":"x","type":"STRING"},{"name":"y","type":"STRING"},{"name":"z","type":"LONG"}]}])
diff --git a/sql/src/test/resources/calcite/expected/ingest/httpExtern-logicalPlan.txt b/sql/src/test/resources/calcite/expected/ingest/httpExtern-logicalPlan.txt
index 755d49878522..18dcef56af43 100644
--- a/sql/src/test/resources/calcite/expected/ingest/httpExtern-logicalPlan.txt
+++ b/sql/src/test/resources/calcite/expected/ingest/httpExtern-logicalPlan.txt
@@ -1,3 +1,3 @@
-LogicalInsert(target=[dst], partitionedBy=[AllGranularity], clusteredBy=[<none>])
+LogicalInsert(target=[dst], partitionedBy=[ALL TIME], clusteredBy=[<none>])
   LogicalProject(inputs=[0..2])
     ExternalTableScan(dataSource=[{"type":"external","inputSource":{"type":"http","uris":["http://foo.com/bar.csv"],"httpAuthenticationUsername":"bob","httpAuthenticationPassword":{"type":"default","password":"secret"}},"inputFormat":{"type":"csv","columns":["x","y","z"]},"signature":[{"name":"x","type":"STRING"},{"name":"y","type":"STRING"},{"name":"z","type":"LONG"}]}])
diff --git a/sql/src/test/resources/calcite/expected/ingest/insertFromExternal-logicalPlan.txt b/sql/src/test/resources/calcite/expected/ingest/insertFromExternal-logicalPlan.txt
index 125729918703..9f9d748dc8bb 100644
--- a/sql/src/test/resources/calcite/expected/ingest/insertFromExternal-logicalPlan.txt
+++ b/sql/src/test/resources/calcite/expected/ingest/insertFromExternal-logicalPlan.txt
@@ -1,3 +1,3 @@
-LogicalInsert(target=[dst], partitionedBy=[AllGranularity], clusteredBy=[<none>])
+LogicalInsert(target=[dst], partitionedBy=[ALL TIME], clusteredBy=[<none>])
   LogicalProject(inputs=[0..2])
     ExternalTableScan(dataSource=[{"type":"external","inputSource":{"type":"inline","data":"a,b,1\nc,d,2\n"},"inputFormat":{"type":"csv","columns":["x","y","z"]},"signature":[{"name":"x","type":"STRING"},{"name":"y","type":"STRING"},{"name":"z","type":"LONG"}]}])
diff --git a/sql/src/test/resources/calcite/expected/ingest/insertWithPartitionedBy-logicalPlan.txt b/sql/src/test/resources/calcite/expected/ingest/insertWithPartitionedBy-logicalPlan.txt
index 6ac42121d848..e8c15d84caa0 100644
--- a/sql/src/test/resources/calcite/expected/ingest/insertWithPartitionedBy-logicalPlan.txt
+++ b/sql/src/test/resources/calcite/expected/ingest/insertWithPartitionedBy-logicalPlan.txt
@@ -1,3 +1,3 @@
-LogicalInsert(target=[druid.dst], partitionedBy=[{type=period, period=PT1H, timeZone=UTC, origin=null}], clusteredBy=[<none>])
+LogicalInsert(target=[druid.dst], partitionedBy=[TIME_FLOOR(`__time`, 'PT1H')], clusteredBy=[<none>])
   LogicalProject(inputs=[0], exprs=[[FLOOR($5), $1]])
     LogicalTableScan(table=[[druid, foo]])
diff --git a/sql/src/test/resources/calcite/expected/ingest/localExtern-logicalPlan.txt b/sql/src/test/resources/calcite/expected/ingest/localExtern-logicalPlan.txt
index 3041c7eacea3..1f7bb5770940 100644
--- a/sql/src/test/resources/calcite/expected/ingest/localExtern-logicalPlan.txt
+++ b/sql/src/test/resources/calcite/expected/ingest/localExtern-logicalPlan.txt
@@ -1,3 +1,3 @@
-LogicalInsert(target=[dst], partitionedBy=[AllGranularity], clusteredBy=[<none>])
+LogicalInsert(target=[dst], partitionedBy=[ALL TIME], clusteredBy=[<none>])
   LogicalProject(inputs=[0..2])
     ExternalTableScan(dataSource=[{"type":"external","inputSource":{"type":"local","files":["/tmp/foo.csv","/tmp/bar.csv"]},"inputFormat":{"type":"csv","columns":["x","y","z"]},"signature":[{"name":"x","type":"STRING"},{"name":"y","type":"STRING"},{"name":"z","type":"LONG"}]}])