sql: implement limited first version of UPSERT

Specifically, add a batching tableWriter for the "short form" described in
https://github.com/cockroachdb/cockroach/blob/master/docs/RFCS/upsert.md.

For cockroachdb#1962.

Notable followups:
- Add support for the full ON CONFLICT DO UPDATE form.
- Hook returningHelper up to the tableWriter abstraction.
- Using fetchRows to compute fetch selectors in Insert/Upsert/Delete.

sql/insert.go

@@ -49,14 +49,23 @@ type insertNode struct {
 }
 
 // Insert inserts rows into the database.
-// Privileges: INSERT on table
+// Privileges: INSERT on table. Also requires UPDATE on "ON DUPLICATE KEY UPDATE".
 //   Notes: postgres requires INSERT. No "on duplicate key update" option.
 //          mysql requires INSERT. Also requires UPDATE on "ON DUPLICATE KEY UPDATE".
 func (p *planner) Insert(n *parser.Insert, autoCommit bool) (planNode, *roachpb.Error) {
 	en, pErr := p.makeEditNode(n.Table, n.Returning, autoCommit, privilege.INSERT)
 	if pErr != nil {
 		return nil, pErr
 	}
+	if n.OnConflict != nil {
+		if err := p.checkPrivilege(en.tableDesc, privilege.UPDATE); err != nil {
+			return nil, roachpb.NewError(err)
+		}
+		// TODO(dan): Support RETURNING in UPSERTs.
+		if n.Returning != nil {
+			return nil, roachpb.NewErrorf("RETURNING is not supported with UPSERT")
+		}
+	}
 
 	var cols []ColumnDescriptor
 	// Determine which columns we're inserting into.
@@ -172,7 +181,41 @@ func (p *planner) Insert(n *parser.Insert, autoCommit bool) (planNode, *roachpb.
 	if err != nil {
 		return nil, roachpb.NewError(err)
 	}
-	tw := &tableInserter{ri: ri, autoCommit: autoCommit}
+
+	var tw tableWriter
+	if n.OnConflict == nil {
+		tw = &tableInserter{ri: ri, autoCommit: autoCommit}
+	} else {
+		// TODO(dan): These are both implied by the short form of UPSERT. When the
+		// INSERT INTO ON CONFLICT form is implemented, get these values from
+		// n.OnConfict.
+		upsertConflictIndex := en.tableDesc.PrimaryIndex
+		insertCols := ri.insertCols
+
+		indexColSet := make(map[ColumnID]struct{}, len(upsertConflictIndex.ColumnIDs))
+		for _, colID := range upsertConflictIndex.ColumnIDs {
+			indexColSet[colID] = struct{}{}
+		}
+
+		// updateCols contains the columns that will be updated when a conflict is
+		// found. For the UPSERT short form, it is the set of columns in insertCols
+		// minus any columns in the conflict index. Example:
+		// `UPSERT INTO abc VALUES (1, 2, 3)` is syntactic sugar for
+		// `INSERT INTO abc VALUES (1, 2, 3) ON CONFLICT a DO UPDATE SET b = 2, c = 3`.
+		updateCols := make([]ColumnDescriptor, 0, len(insertCols))
+		for _, c := range insertCols {
+			if _, ok := indexColSet[c.ID]; !ok {
+				updateCols = append(updateCols, c)
+			}
+		}
+		ru, err := makeRowUpdater(en.tableDesc, updateCols)
+		if err != nil {
+			return nil, roachpb.NewError(err)
+		}
+		// TODO(dan): Use ru.fetchCols to compute the fetch selectors.
+
+		tw = &tableUpserter{ri: ri, ru: ru, autoCommit: autoCommit}
+	}
 
 	in := &insertNode{
 		n:                     n,

sql/parser/insert.go

@@ -29,15 +29,21 @@ import (
 
 // Insert represents an INSERT statement.
 type Insert struct {
-	Table     TableExpr
-	Columns   QualifiedNames
-	Rows      *Select
-	Returning ReturningExprs
+	Table      TableExpr
+	Columns    QualifiedNames
+	Rows       *Select
+	Returning  ReturningExprs
+	OnConflict *OnConflict
 }
 
 func (node *Insert) String() string {
 	var buf bytes.Buffer
-	fmt.Fprintf(&buf, "INSERT INTO %s", node.Table)
+	if node.OnConflict != nil {
+		buf.WriteString("UPSERT")
+	} else {
+		buf.WriteString("INSERT")
+	}
+	fmt.Fprintf(&buf, " INTO %s", node.Table)
 	if node.Columns != nil {
 		fmt.Fprintf(&buf, "(%s)", node.Columns)
 	}
@@ -54,3 +60,12 @@ func (node *Insert) String() string {
 func (node *Insert) DefaultValues() bool {
 	return node.Rows.Select == nil
 }
+
+// OnConflict represents an `ON CONFLICT index DO UPDATE SET` statement.
+//
+// The zero value for OnConflict is used to signal the UPSERT short form, which
+// uses the primary key for Index and the values being inserted for Exprs.
+type OnConflict struct {
+	Index Name
+	Exprs UpdateExprs
+}

sql/parser/parse_test.go

@@ -172,6 +172,17 @@ func TestParse(t *testing.T) {
 		{`INSERT INTO a VALUES (1, 2) RETURNING 1, 2`},
 		{`INSERT INTO a VALUES (1, 2) RETURNING a + b, c`},
 
+		{`UPSERT INTO a VALUES (1)`},
+		{`UPSERT INTO a.b VALUES (1)`},
+		{`UPSERT INTO a VALUES (1, 2)`},
+		{`UPSERT INTO a VALUES (1, DEFAULT)`},
+		{`UPSERT INTO a VALUES (1, 2), (3, 4)`},
+		{`UPSERT INTO a VALUES (a + 1, 2 * 3)`},
+		{`UPSERT INTO a(a, b) VALUES (1, 2)`},
+		{`UPSERT INTO a(a, a.b) VALUES (1, 2)`},
+		{`UPSERT INTO a SELECT b, c FROM d`},
+		{`UPSERT INTO a DEFAULT VALUES`},
+
 		{`SELECT 1 + 1`},
 		{`SELECT - 1`},
 		{`SELECT + 1`},