Insert into a table with a list of columns

Now it is possible to specify list of columns with INSERT INTO e.g.
INSERT INTO orders (orderkey) VALUES(1);

When the list of column names is specified the columns produced by the
query must exactly match to them. Each column not being present in
the column list will be filled with a ``null`` value.
In case the list of column names is not given then the columns produced
by query must exactly match the columns in the table being inserted
into.

When the list of columns is present then there is added additional
ProjectNode to logical plan which represents mapping between query
columns and table columns. Also it takes care of table columns not
present in a list.

presto-docs/src/main/sphinx/sql/insert.rst

@@ -7,18 +7,19 @@ Synopsis
 
 .. code-block:: none
 
-    INSERT INTO table_name query
+    INSERT INTO table_name [ ( column [, ... ] ) ] query
 
 Description
 -----------
 
 Insert new rows into a table.
 
-.. note::
+If the list of column names is specified, they must exactly match the list
+of columns produced by the query. Each column in the table not present in the
+column list will be filled with a ``null`` value. Otherwise, if the list of
+columns is not specified, the columns produced by the query must exactly match
+the columns in the table being inserted into.
 
-    Currently, the list of column names cannot be specified. Thus,
-    the columns produced by the query must exactly match the columns
-    in the table being inserted into.
 
 Examples
 --------
@@ -35,3 +36,14 @@ Insert a single row into the ``cities`` table::
 Insert multiple rows into the ``cities`` table::
 
     INSERT INTO cities VALUES (2, 'San Jose'), (3, 'Oakland');
+
+Insert a single row into the ``nation`` table with the specified column list::
+
+    INSERT INTO nation (nationkey, name, regionkey, comment)
+    VALUES (26, 'POLAND', 3, 'no comment');
+
+Insert a row without specifying the ``comment`` column. 
+That column will be ``null``::
+
+    INSERT INTO nation (nationkey, name, regionkey) 
+    VALUES (26, 'POLAND', 3);

presto-hive/src/test/java/com/facebook/presto/hive/TestHiveDistributedQueries.java

@@ -28,13 +28,6 @@ public TestHiveDistributedQueries()
         super(createQueryRunner(getTables()), createSampledSession());
     }
 
-    @Override
-    public void testInsert()
-            throws Exception
-    {
-        // Hive connector currently does not support insert
-    }
-
     @Override
     public void testDelete()
             throws Exception

presto-hive/src/test/java/com/facebook/presto/hive/TestHiveIntegrationSmokeTest.java

@@ -266,7 +266,7 @@ public void insertTable(HiveStorageFormat storageFormat)
                 "(" +
                 "  _varchar VARCHAR," +
                 "  _bigint BIGINT," +
-                "  _doube DOUBLE," +
+                "  _double DOUBLE," +
                 "  _boolean BOOLEAN" +
                 ") " +
                 "WITH (format = '" + storageFormat + "') ";
@@ -286,6 +286,14 @@ public void insertTable(HiveStorageFormat storageFormat)
 
         assertQuery("SELECT * from test_insert_format_table", select);
 
+        assertQuery("INSERT INTO test_insert_format_table (_bigint, _double) SELECT 2, 14.3", "SELECT 1");
+
+        assertQuery("SELECT * from test_insert_format_table where _bigint = 2", "SELECT null, 2, 14.3, null");
+
+        assertQuery("INSERT INTO test_insert_format_table (_double, _bigint) SELECT 2.72, 3", "SELECT 1");
+
+        assertQuery("SELECT * from test_insert_format_table where _bigint = 3", "SELECT null, 3, 2.72, null");
+
         assertQueryTrue("DROP TABLE test_insert_format_table");
 
         assertFalse(queryRunner.tableExists(getSession(), "test_insert_format_table"));

presto-main/src/main/java/com/facebook/presto/metadata/TableMetadata.java

@@ -19,6 +19,7 @@
 
 import java.util.List;
 
+import static com.facebook.presto.util.ImmutableCollectors.toImmutableList;
 import static java.util.Objects.requireNonNull;
 
 public class TableMetadata
@@ -54,4 +55,27 @@ public List<ColumnMetadata> getColumns()
     {
         return metadata.getColumns();
     }
+
+    public List<String> getVisibleColumnNames()
+    {
+        return getColumns().stream()
+                .filter(column -> !column.isHidden())
+                .map(ColumnMetadata::getName)
+                .collect(toImmutableList());
+    }
+
+    public List<ColumnMetadata> getVisibleColumns()
+    {
+        return getColumns().stream()
+                .filter(column -> !column.isHidden())
+                .collect(toImmutableList());
+    }
+
+    public ColumnMetadata getColumn(String name)
+    {
+        return getColumns().stream()
+                .filter(columnMetadata -> columnMetadata.getName().equals(name))
+                .findFirst()
+                .orElseThrow(() -> new IllegalArgumentException(String.format("Invalid column name: %s", name)));
+    }
 }

presto-main/src/main/java/com/facebook/presto/sql/analyzer/Analysis.java

@@ -35,13 +35,16 @@
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.SetMultimap;
 
+import javax.annotation.concurrent.Immutable;
+
 import java.util.IdentityHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static java.util.Objects.requireNonNull;
 
 public class Analysis
@@ -83,8 +86,7 @@ public class Analysis
     private Map<String, Expression> createTableProperties = ImmutableMap.of();
     private boolean createTableAsSelectWithData = true;
 
-    // for insert
-    private Optional<TableHandle> insertTarget = Optional.empty();
+    private Optional<Insert> insert = Optional.empty();
 
     // for delete
     private Optional<Delete> delete = Optional.empty();
@@ -146,7 +148,7 @@ public IdentityHashMap<Expression, Type> getTypes()
 
     public Type getType(Expression expression)
     {
-        Preconditions.checkArgument(types.containsKey(expression), "Expression not analyzed: %s", expression);
+        checkArgument(types.containsKey(expression), "Expression not analyzed: %s", expression);
         return types.get(expression);
     }
 
@@ -356,14 +358,14 @@ public Map<String, Expression> getCreateTableProperties()
         return createTableProperties;
     }
 
-    public void setInsertTarget(TableHandle target)
+    public void setInsert(Insert insert)
     {
-        this.insertTarget = Optional.of(target);
+        this.insert = Optional.of(insert);
     }
 
-    public Optional<TableHandle> getInsertTarget()
+    public Optional<Insert> getInsert()
     {
-        return insertTarget;
+        return insert;
     }
 
     public void setDelete(Delete delete)
@@ -441,4 +443,28 @@ public boolean equals(Object obj)
                     Objects.equals(this.rightInPredicates, other.rightInPredicates);
         }
     }
+
+    @Immutable
+    public static final class Insert
+    {
+        private final TableHandle target;
+        private final List<ColumnHandle> columns;
+
+        public Insert(TableHandle target, List<ColumnHandle> columns)
+        {
+            this.target = requireNonNull(target, "target is null");
+            this.columns = requireNonNull(columns, "columns is null");
+            checkArgument(columns.size() > 0, "No columns given to insert");
+        }
+
+        public List<ColumnHandle> getColumns()
+        {
+            return columns;
+        }
+
+        public TableHandle getTarget()
+        {
+            return target;
+        }
+    }
 }

presto-main/src/main/java/com/facebook/presto/sql/analyzer/StatementAnalyzer.java

@@ -165,6 +165,7 @@
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.MISMATCHED_COLUMN_ALIASES;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.MISMATCHED_SET_COLUMN_TYPES;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.MISSING_CATALOG;
+import static com.facebook.presto.sql.analyzer.SemanticErrorCode.MISSING_COLUMN;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.MISSING_SCHEMA;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.MISSING_TABLE;
 import static com.facebook.presto.sql.analyzer.SemanticErrorCode.MUST_BE_AGGREGATE_OR_GROUP_BY;
@@ -510,23 +511,48 @@ protected RelationType visitInsert(Insert insert, AnalysisContext context)
         analysis.setUpdateType("INSERT");
 
         // analyze the query that creates the data
-        RelationType descriptor = process(insert.getQuery(), context);
+        RelationType queryDescriptor = process(insert.getQuery(), context);
 
         // verify the insert destination columns match the query
         Optional<TableHandle> targetTableHandle = metadata.getTableHandle(session, targetTable);
         if (!targetTableHandle.isPresent()) {
             throw new SemanticException(MISSING_TABLE, insert, "Table '%s' does not exist", targetTable);
         }
         accessControl.checkCanInsertIntoTable(session.getIdentity(), targetTable);
-        analysis.setInsertTarget(targetTableHandle.get());
 
-        List<ColumnMetadata> columns = metadata.getTableMetadata(session, targetTableHandle.get()).getColumns();
-        Iterable<Type> tableTypes = columns.stream()
-                .filter(column -> !column.isHidden())
-                .map(ColumnMetadata::getType)
+        TableMetadata tableMetadata = metadata.getTableMetadata(session, targetTableHandle.get());
+        List<String> tableColumns = tableMetadata.getVisibleColumnNames();
+
+        final List<String> insertColumns;
+        if (insert.getColumns().isPresent()) {
+            insertColumns = insert.getColumns().get().stream()
+                    .map(String::toLowerCase)
+                    .collect(toImmutableList());
+
+            Set<String> columnNames = new HashSet<>();
+            for (String insertColumn : insertColumns) {
+                if (!tableColumns.contains(insertColumn)) {
+                    throw new SemanticException(MISSING_COLUMN, insert, "Insert column name does not exist in target table: %s", insertColumn);
+                }
+                if (!columnNames.add(insertColumn)) {
+                    throw new SemanticException(DUPLICATE_COLUMN_NAME, insert, "Insert column name is specified more than once: %s", insertColumn);
+                }
+            }
+        }
+        else {
+            insertColumns = tableColumns;
+        }
+
+        Map<String, ColumnHandle> columnHandles = metadata.getColumnHandles(session, targetTableHandle.get());
+        analysis.setInsert(new Analysis.Insert(
+                targetTableHandle.get(),
+                insertColumns.stream().map(column -> columnHandles.get(column)).collect(toImmutableList())));
+
+        Iterable<Type> tableTypes = insertColumns.stream()
+                .map(insertColumn -> tableMetadata.getColumn(insertColumn).getType())
                 .collect(toImmutableList());
 
-        Iterable<Type> queryTypes = transform(descriptor.getVisibleFields(), Field::getType);
+        Iterable<Type> queryTypes = transform(queryDescriptor.getVisibleFields(), Field::getType);
 
         if (!elementsEqual(tableTypes, queryTypes)) {
             throw new SemanticException(MISMATCHED_SET_COLUMN_TYPES, insert, "Insert query has mismatched column types: " +

presto-main/src/main/java/com/facebook/presto/sql/planner/LogicalPlanner.java

@@ -16,8 +16,8 @@
 import com.facebook.presto.Session;
 import com.facebook.presto.metadata.Metadata;
 import com.facebook.presto.metadata.QualifiedTableName;
-import com.facebook.presto.metadata.TableHandle;
 import com.facebook.presto.metadata.TableMetadata;
+import com.facebook.presto.spi.ColumnHandle;
 import com.facebook.presto.spi.ColumnMetadata;
 import com.facebook.presto.spi.ConnectorTableMetadata;
 import com.facebook.presto.spi.PrestoException;
@@ -29,10 +29,13 @@
 import com.facebook.presto.sql.planner.plan.LimitNode;
 import com.facebook.presto.sql.planner.plan.OutputNode;
 import com.facebook.presto.sql.planner.plan.PlanNode;
+import com.facebook.presto.sql.planner.plan.ProjectNode;
 import com.facebook.presto.sql.planner.plan.TableCommitNode;
 import com.facebook.presto.sql.planner.plan.TableWriterNode;
 import com.facebook.presto.sql.tree.Expression;
+import com.facebook.presto.sql.tree.NullLiteral;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
 
 import java.util.List;
 import java.util.Map;
@@ -79,7 +82,7 @@ public Plan plan(Analysis analysis)
         if (analysis.getCreateTableDestination().isPresent()) {
             plan = createTableCreationPlan(analysis);
         }
-        else if (analysis.getInsertTarget().isPresent()) {
+        else if (analysis.getInsert().isPresent()) {
             plan = createInsertPlan(analysis);
         }
         else if (analysis.getDelete().isPresent()) {
@@ -113,26 +116,55 @@ private RelationPlan createTableCreationPlan(Analysis analysis)
 
         TableMetadata tableMetadata = createTableMetadata(destination, getOutputTableColumns(plan), analysis.getCreateTableProperties(), plan.getSampleWeight().isPresent());
         if (plan.getSampleWeight().isPresent() && !metadata.canCreateSampledTables(session, destination.getCatalogName())) {
-          throw new PrestoException(NOT_SUPPORTED, "Cannot write sampled data to a store that doesn't support sampling");
+            throw new PrestoException(NOT_SUPPORTED, "Cannot write sampled data to a store that doesn't support sampling");
         }
 
         return createTableWriterPlan(
                 analysis,
                 plan,
-                new CreateName(destination.getCatalogName(), tableMetadata), getVisibleColumnNames(tableMetadata));
+                new CreateName(destination.getCatalogName(), tableMetadata), tableMetadata.getVisibleColumnNames());
     }
 
     private RelationPlan createInsertPlan(Analysis analysis)
     {
-        TableHandle target = analysis.getInsertTarget().get();
+        Analysis.Insert insert = analysis.getInsert().get();
 
-        TableMetadata tableMetadata = metadata.getTableMetadata(session, target);
+        TableMetadata tableMetadata = metadata.getTableMetadata(session, insert.getTarget());
+
+        List<String> visibleTableColumnNames = tableMetadata.getVisibleColumnNames();
+        List<ColumnMetadata> visibleTableColumns = tableMetadata.getVisibleColumns();
+
+        RelationPlan plan = createRelationPlan(analysis);
+
+        Map<String, ColumnHandle> columns = metadata.getColumnHandles(session, insert.getTarget());
+        ImmutableMap.Builder<Symbol, Expression> assignments = ImmutableMap.builder();
+        for (ColumnMetadata column : tableMetadata.getVisibleColumns()) {
+            Symbol output = symbolAllocator.newSymbol(column.getName(), column.getType());
+            int index = insert.getColumns().indexOf(columns.get(column.getName()));
+            if (index < 0) {
+                assignments.put(output, new NullLiteral());
+            }
+            else {
+                assignments.put(output, plan.getSymbol(index).toQualifiedNameReference());
+            }
+        }
+        ProjectNode projectNode = new ProjectNode(idAllocator.getNextId(), plan.getRoot(), assignments.build());
+
+        RelationType tupleDescriptor = new RelationType(visibleTableColumns.stream()
+                .map(column -> Field.newUnqualified(column.getName(), column.getType()))
+                .collect(toImmutableList()));
+
+        plan = new RelationPlan(
+                projectNode,
+                tupleDescriptor,
+                projectNode.getOutputSymbols(),
+                plan.getSampleWeight());
 
         return createTableWriterPlan(
                 analysis,
-                createRelationPlan(analysis),
-                new InsertReference(target),
-                getVisibleColumnNames(tableMetadata));
+                plan,
+                new InsertReference(insert.getTarget()),
+                visibleTableColumnNames);
     }
 
     private RelationPlan createTableWriterPlan(Analysis analysis, RelationPlan plan, WriterTarget target, List<String> columnNames)
@@ -227,12 +259,4 @@ private static List<ColumnMetadata> getOutputTableColumns(RelationPlan plan)
         }
         return columns.build();
     }
-
-    private static List<String> getVisibleColumnNames(TableMetadata tableMetadata)
-    {
-        return tableMetadata.getColumns().stream()
-                .filter(column -> !column.isHidden())
-                .map(ColumnMetadata::getName)
-                .collect(toImmutableList());
-    }
 }