SQL INSERT planner support.

benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlBenchmark.java

@@ -425,7 +425,7 @@ public void setup()
         CalciteTests.createMockRootSchema(conglomerate, walker, plannerConfig, AuthTestUtils.TEST_AUTHORIZER_MAPPER);
     plannerFactory = new PlannerFactory(
         rootSchema,
-        CalciteTests.createMockQueryLifecycleFactory(walker, conglomerate),
+        CalciteTests.createMockQueryMakerFactory(walker, conglomerate),
         createOperatorTable(),
         CalciteTests.createExprMacroTable(),
         plannerConfig,
@@ -467,7 +467,7 @@ public void querySql(Blackhole blackhole) throws Exception
     );
     final String sql = QUERIES.get(Integer.parseInt(query));
     try (final DruidPlanner planner = plannerFactory.createPlannerForTesting(context, sql)) {
-      final PlannerResult plannerResult = planner.plan(sql);
+      final PlannerResult plannerResult = planner.plan();
       final Sequence<Object[]> resultSequence = plannerResult.run();
       final Object[] lastRow = resultSequence.accumulate(null, (accumulated, in) -> in);
       blackhole.consume(lastRow);
@@ -485,7 +485,7 @@ public void planSql(Blackhole blackhole) throws Exception
     );
     final String sql = QUERIES.get(Integer.parseInt(query));
     try (final DruidPlanner planner = plannerFactory.createPlannerForTesting(context, sql)) {
-      final PlannerResult plannerResult = planner.plan(sql);
+      final PlannerResult plannerResult = planner.plan();
       blackhole.consume(plannerResult);
     }
   }

benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlExpressionBenchmark.java

@@ -268,7 +268,7 @@ public void setup()
         CalciteTests.createMockRootSchema(conglomerate, walker, plannerConfig, AuthTestUtils.TEST_AUTHORIZER_MAPPER);
     plannerFactory = new PlannerFactory(
         rootSchema,
-        CalciteTests.createMockQueryLifecycleFactory(walker, conglomerate),
+        CalciteTests.createMockQueryMakerFactory(walker, conglomerate),
         CalciteTests.createOperatorTable(),
         CalciteTests.createExprMacroTable(),
         plannerConfig,
@@ -305,7 +305,7 @@ public void querySql(Blackhole blackhole) throws Exception
     );
     final String sql = QUERIES.get(Integer.parseInt(query));
     try (final DruidPlanner planner = plannerFactory.createPlannerForTesting(context, sql)) {
-      final PlannerResult plannerResult = planner.plan(sql);
+      final PlannerResult plannerResult = planner.plan();
       final Sequence<Object[]> resultSequence = plannerResult.run();
       final Object[] lastRow = resultSequence.accumulate(null, (accumulated, in) -> in);
       blackhole.consume(lastRow);

benchmarks/src/test/java/org/apache/druid/benchmark/query/SqlVsNativeBenchmark.java

@@ -115,7 +115,7 @@ public void setup()
         CalciteTests.createMockRootSchema(conglomerate, walker, plannerConfig, AuthTestUtils.TEST_AUTHORIZER_MAPPER);
     plannerFactory = new PlannerFactory(
         rootSchema,
-        CalciteTests.createMockQueryLifecycleFactory(walker, conglomerate),
+        CalciteTests.createMockQueryMakerFactory(walker, conglomerate),
         CalciteTests.createOperatorTable(),
         CalciteTests.createExprMacroTable(),
         plannerConfig,
@@ -162,7 +162,7 @@ public void queryNative(Blackhole blackhole)
   public void queryPlanner(Blackhole blackhole) throws Exception
   {
     try (final DruidPlanner planner = plannerFactory.createPlannerForTesting(null, sqlQuery)) {
-      final PlannerResult plannerResult = planner.plan(sqlQuery);
+      final PlannerResult plannerResult = planner.plan();
       final Sequence<Object[]> resultSequence = plannerResult.run();
       final Object[] lastRow = resultSequence.accumulate(null, (accumulated, in) -> in);
       blackhole.consume(lastRow);

core/src/main/java/org/apache/druid/data/input/impl/InlineInputSource.java

@@ -30,6 +30,7 @@
 
 import javax.annotation.Nullable;
 import java.io.File;
+import java.util.Objects;
 import java.util.stream.Stream;
 
 public class InlineInputSource extends AbstractInputSource
@@ -75,4 +76,31 @@ protected InputSourceReader formattableReader(
         temporaryDirectory
     );
   }
+
+  @Override
+  public boolean equals(Object o)
+  {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    InlineInputSource that = (InlineInputSource) o;
+    return Objects.equals(data, that.data);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hash(data);
+  }
+
+  @Override
+  public String toString()
+  {
+    return "InlineInputSource{" +
+           "data='" + data + '\'' +
+           '}';
+  }
 }

processing/src/main/java/org/apache/druid/segment/column/ColumnSignature.java

@@ -0,0 +1,79 @@
+/*
+ * 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.segment.column;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.druid.java.util.common.IAE;
+
+import javax.annotation.Nullable;
+
+/**
+ * Class used by {@link RowSignature} for serialization.
+ *
+ * Package-private since it is not intended to be used outside that narrow use case. In other cases where passing
+ * around information about column types is important, use {@link ColumnType} instead.
+ */
+class ColumnSignature
+{
+  private final String name;
+
+  @Nullable
+  private final ColumnType type;
+
+  @JsonCreator
+  ColumnSignature(
+      @JsonProperty("name") String name,
+      @JsonProperty("type") @Nullable ColumnType type
+  )
+  {
+    this.name = name;
+    this.type = type;
+
+    // Name must be nonnull, but type can be null (if the type is unknown)
+    if (name == null || name.isEmpty()) {
+      throw new IAE(name, "Column name must be non-empty");
+    }
+  }
+
+  @JsonProperty("name")
+  String name()
+  {
+    return name;
+  }
+
+  @Nullable
+  @JsonProperty("type")
+  @JsonInclude(JsonInclude.Include.NON_NULL)
+  ColumnType type()
+  {
+    return type;
+  }
+
+  @Override
+  public String toString()
+  {
+    return "ColumnSignature{" +
+           "name='" + name + '\'' +
+           ", type=" + type +
+           '}';
+  }
+}

processing/src/main/java/org/apache/druid/segment/column/RowSignature.java

@@ -19,12 +19,13 @@
 
 package org.apache.druid.segment.column;
 
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import it.unimi.dsi.fastutil.objects.Object2IntMap;
 import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
 import org.apache.druid.java.util.common.IAE;
-import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.query.aggregation.AggregatorFactory;
 import org.apache.druid.query.aggregation.PostAggregator;
 import org.apache.druid.query.dimension.DimensionSpec;
@@ -40,9 +41,7 @@
 import java.util.Optional;
 
 /**
- * Type signature for a row in a Druid datasource or query result. Rows have an ordering and every
- * column has a defined type. This is a little bit of a fiction in the Druid world (where rows do not _actually_ have
- * well defined types) but we do impose types for the SQL layer.
+ * Type signature for a row in a Druid datasource or query result.
  *
  * @see org.apache.druid.query.QueryToolChest#resultArraySignature which returns signatures for query results
  * @see org.apache.druid.query.InlineDataSource#getRowSignature which returns signatures for inline datasources
@@ -55,30 +54,42 @@ public class RowSignature implements ColumnInspector
   private final Object2IntMap<String> columnPositions = new Object2IntOpenHashMap<>();
   private final List<String> columnNames;
 
-  private RowSignature(final List<Pair<String, ColumnType>> columnTypeList)
+  private RowSignature(final List<ColumnSignature> columnTypeList)
   {
     this.columnPositions.defaultReturnValue(-1);
 
     final ImmutableList.Builder<String> columnNamesBuilder = ImmutableList.builder();
 
     for (int i = 0; i < columnTypeList.size(); i++) {
-      final Pair<String, ColumnType> pair = columnTypeList.get(i);
-      final ColumnType existingType = columnTypes.get(pair.lhs);
+      final ColumnSignature sig = columnTypeList.get(i);
+      final ColumnType existingType = columnTypes.get(sig.name());
 
-      if (columnTypes.containsKey(pair.lhs) && existingType != pair.rhs) {
+      if (columnTypes.containsKey(sig.name()) && !Objects.equals(existingType, sig.type())) {
         // It's ok to add the same column twice as long as the type is consistent.
         // Note: we need the containsKey because the existingType might be present, but null.
-        throw new IAE("Column[%s] has conflicting types [%s] and [%s]", pair.lhs, existingType, pair.rhs);
+        throw new IAE("Column[%s] has conflicting types [%s] and [%s]", sig.name(), existingType, sig.type());
       }
 
-      columnTypes.put(pair.lhs, pair.rhs);
-      columnPositions.put(pair.lhs, i);
-      columnNamesBuilder.add(pair.lhs);
+      columnTypes.put(sig.name(), sig.type());
+      columnPositions.put(sig.name(), i);
+      columnNamesBuilder.add(sig.name());
     }
 
     this.columnNames = columnNamesBuilder.build();
   }
 
+  @JsonCreator
+  static RowSignature fromColumnSignatures(final List<ColumnSignature> columnSignatures)
+  {
+    final Builder builder = builder();
+
+    for (final ColumnSignature columnSignature : columnSignatures) {
+      builder.add(columnSignature.name(), columnSignature.type());
+    }
+
+    return builder.build();
+  }
+
   public static Builder builder()
   {
     return new Builder();
@@ -158,6 +169,19 @@ public int indexOf(final String columnName)
     return columnPositions.applyAsInt(columnName);
   }
 
+  @JsonValue
+  private List<ColumnSignature> asColumnSignatures()
+  {
+    final List<ColumnSignature> retVal = new ArrayList<>();
+
+    for (String columnName : columnNames) {
+      final ColumnType type = columnTypes.get(columnName);
+      retVal.add(new ColumnSignature(columnName, type));
+    }
+
+    return retVal;
+  }
+
   @Override
   public boolean equals(Object o)
   {
@@ -207,7 +231,7 @@ public ColumnCapabilities getColumnCapabilities(String column)
 
   public static class Builder
   {
-    private final List<Pair<String, ColumnType>> columnTypeList;
+    private final List<ColumnSignature> columnTypeList;
 
     private Builder()
     {
@@ -216,21 +240,21 @@ private Builder()
 
     /**
      * Add a column to this signature.
-     *  @param columnName name, must be nonnull
+     *
+     * @param columnName name, must be nonnull
      * @param columnType type, may be null if unknown
      */
     public Builder add(final String columnName, @Nullable final ColumnType columnType)
     {
-      // Name must be nonnull, but type can be null (if the type is unknown)
-      Preconditions.checkNotNull(columnName, "'columnName' must be non-null");
-      columnTypeList.add(Pair.of(columnName, columnType));
+      columnTypeList.add(new ColumnSignature(columnName, columnType));
       return this;
     }
 
     public Builder addAll(final RowSignature other)
     {
-      for (String columnName : other.getColumnNames()) {
-        add(columnName, other.getColumnType(columnName).orElse(null));
+      final List<String> names = other.getColumnNames();
+      for (int i = 0; i < names.size(); i++) {
+        add(names.get(i), other.getColumnType(i).orElse(null));
       }
 
       return this;