Fixes for column duplicates and existence

Signed-off-by: Daylon Wilkins <[email protected]>

Author: Hydrocharged

engine_test.go

@@ -1908,11 +1908,21 @@ func TestInsertIntoErrors(t *testing.T) {
 			"too many values no columns specified",
 			"INSERT INTO mytable VALUES (999, 'x', 'y');",
 		},
+		{
+			"non-existent column",
+			"INSERT INTO mytable (i, s, z) VALUES (999, 'x', 999);",
+		},
+		{
+			"duplicate column",
+			"INSERT INTO mytable (i, s, s) VALUES (999, 'x', 'x');",
+		},
 	}
 
 	for _, expectedFailure := range expectedFailures {
-		_, _, err := newEngine(t).Query(newCtx(), expectedFailure.query)
-		require.Error(t, err)
+		t.Run(expectedFailure.name, func(t *testing.T) {
+			_, _, err := newEngine(t).Query(newCtx(), expectedFailure.query)
+			require.Error(t, err)
+		})
 	}
 }
 

sql/plan/insert.go

@@ -14,6 +14,8 @@ var ErrInsertIntoNotSupported = errors.NewKind("table doesn't support INSERT INT
 var ErrInsertIntoMismatchValueCount =
 	errors.NewKind("number of values does not match number of columns provided")
 var ErrInsertIntoUnsupportedValues = errors.NewKind("%T is unsupported for inserts")
+var ErrInsertIntoDuplicateColumn = errors.NewKind("duplicate column name %v")
+var ErrInsertIntoNonexistentColumn = errors.NewKind("invalid column name %v")
 
 // InsertInto is a node describing the insertion into some table.
 type InsertInto struct {
@@ -77,6 +79,11 @@ func (p *InsertInto) Execute(ctx *sql.Context) (int, error) {
 		for i, f := range dstSchema {
 			p.Columns[i] = f.Name
 		}
+	} else {
+		err = p.validateColumns(ctx, dstSchema)
+		if err != nil {
+			return 0, err
+		}
 	}
 
 	err = p.validateValueCount(ctx)
@@ -168,3 +175,22 @@ func (p *InsertInto) validateValueCount(ctx *sql.Context) error {
 	}
 	return nil
 }
+
+func (p *InsertInto) validateColumns(ctx *sql.Context, dstSchema sql.Schema) error {
+	dstColNames := make(map[string]struct{})
+	for _, dstCol := range dstSchema {
+		dstColNames[dstCol.Name] = struct{}{}
+	}
+	columnNames := make(map[string]struct{})
+	for _, columnName := range p.Columns {
+		if _, exists := dstColNames[columnName]; !exists {
+			return ErrInsertIntoNonexistentColumn.New(columnName)
+		}
+		if _, exists := columnNames[columnName]; !exists {
+			columnNames[columnName] = struct{}{}
+		} else {
+			return ErrInsertIntoDuplicateColumn.New(columnName)
+		}
+	}
+	return nil
+}