Incorporated Comments

go/vt/vtgate/planbuilder/insert.go

@@ -60,21 +60,26 @@ func buildInsertPlan(ins *sqlparser.Insert, vschema VSchema) (*engine.Route, err
 	indexColInsert := make([]bool, len(colVindexes))
 	autoIncValues := make([]interface{}, 0, len(values))
 	for rowNum := 0; rowNum < len(values); rowNum++ {
-		value := make([]interface{}, 0, len(colVindexes))
+		rowValue := make([]interface{}, 0, len(colVindexes))
 		for colNum, index := range colVindexes {
-			row, pos := findOrInsertPos(ins, index.Column, rowNum, &indexColInsert[colNum])
-			if err := buildIndexPlan(index, &value, rowNum, row, pos); err != nil {
-				return nil, err
-			}
-		}
-		if route.Table.AutoIncrement != nil {
-			autoIncVal, err := buildAutoIncrementPlan(ins, route.Table.AutoIncrement, route, &value, rowNum, &autoIncColInsert)
+			row, pos, colInsert := findOrInsertPos(ins, index.Column, rowNum, indexColInsert[colNum])
+			value, err := buildIndexPlan(index, rowNum, row, pos)
 			if err != nil {
 				return nil, err
 			}
+			indexColInsert[colNum] = colInsert
+			rowValue = append(rowValue, value)
+		}
+		if route.Table.AutoIncrement != nil {
+			autoIncVal, colInsert, value, err := buildAutoIncrementPlan(ins, route.Table.AutoIncrement, route, rowValue, rowNum, autoIncColInsert)
+			if err != nil {
+				return nil, err
+			}
+			rowValue = value
+			autoIncColInsert = colInsert
 			autoIncValues = append(autoIncValues, autoIncVal)
 		}
-		routeValues = append(routeValues, value)
+		routeValues = append(routeValues, rowValue)
 	}
 	if route.Table.AutoIncrement != nil {
 		route.Generate = &engine.Generate{
@@ -91,38 +96,38 @@ func buildInsertPlan(ins *sqlparser.Insert, vschema VSchema) (*engine.Route, err
 
 // buildIndexPlan adds the insert value to the Values field for the specified ColumnVindex.
 // This value will be used at the time of insert to validate the vindex value.
-func buildIndexPlan(colVindex *vindexes.ColumnVindex, value *[]interface{}, rowNum int, row sqlparser.ValTuple, pos int) error {
+func buildIndexPlan(colVindex *vindexes.ColumnVindex, rowNum int, row sqlparser.ValTuple, pos int) (interface{}, error) {
 	val, err := valConvert(row[pos])
 	if err != nil {
-		return fmt.Errorf("could not convert val: %s, pos: %d: %v", sqlparser.String(row[pos]), pos, err)
+		return val, fmt.Errorf("could not convert val: %s, pos: %d: %v", sqlparser.String(row[pos]), pos, err)
 	}
-	*value = append(*value, val)
 	row[pos] = sqlparser.ValArg([]byte(":_" + colVindex.Column.Original() + strconv.Itoa(rowNum)))
-	return nil
+	return val, nil
 }
 
-func buildAutoIncrementPlan(ins *sqlparser.Insert, autoinc *vindexes.AutoIncrement, route *engine.Route, value *[]interface{}, rowNum int, autoIncColInsert *bool) (interface{}, error) {
+func buildAutoIncrementPlan(ins *sqlparser.Insert, autoinc *vindexes.AutoIncrement, route *engine.Route, rowValue []interface{}, rowNum int, autoIncColInsert bool) (interface{}, bool, []interface{}, error) {
 	var autoIncVal interface{}
 	// If it's also a colvindex, we have to add a redirect from route.Values.
 	// Otherwise, we have to redirect from row[pos].
 	if autoinc.ColumnVindexNum >= 0 {
-		autoIncVal = (*value)[autoinc.ColumnVindexNum]
-		(*value)[autoinc.ColumnVindexNum] = ":" + engine.SeqVarName + strconv.Itoa(rowNum)
-		return autoIncVal, nil
+		autoIncVal = rowValue[autoinc.ColumnVindexNum]
+		rowValue[autoinc.ColumnVindexNum] = ":" + engine.SeqVarName + strconv.Itoa(rowNum)
+		return autoIncVal, autoIncColInsert, rowValue, nil
 	}
-	row, pos := findOrInsertPos(ins, autoinc.Column, rowNum, autoIncColInsert)
+	row, pos, colInsert := findOrInsertPos(ins, autoinc.Column, rowNum, autoIncColInsert)
+	autoIncColInsert = colInsert
 	val, err := valConvert(row[pos])
 	if err != nil {
-		return autoIncVal, fmt.Errorf("could not convert val: %s, pos: %d: %v", sqlparser.String(row[pos]), pos, err)
+		return autoIncVal, autoIncColInsert, rowValue, fmt.Errorf("could not convert val: %s, pos: %d: %v", sqlparser.String(row[pos]), pos, err)
 	}
 	autoIncVal = val
 	row[pos] = sqlparser.ValArg([]byte(":" + engine.SeqVarName + strconv.Itoa(rowNum)))
 
-	return autoIncVal, nil
+	return autoIncVal, autoIncColInsert, rowValue, nil
 }
 
-func findOrInsertPos(ins *sqlparser.Insert, col cistring.CIString, rowNum int, ColInsert *bool) (row sqlparser.ValTuple, pos int) {
-	pos = -1
+func findOrInsertPos(ins *sqlparser.Insert, col cistring.CIString, rowNum int, colInsert bool) (sqlparser.ValTuple, int, bool) {
+	pos := -1
 	for i, column := range ins.Columns {
 		if col.Equal(cistring.CIString(column)) {
 			pos = i
@@ -130,12 +135,12 @@ func findOrInsertPos(ins *sqlparser.Insert, col cistring.CIString, rowNum int, C
 		}
 	}
 	if pos == -1 {
-		*ColInsert = true
+		colInsert = true
 		pos = len(ins.Columns)
 		ins.Columns = append(ins.Columns, sqlparser.ColIdent(col))
 	}
-	if *ColInsert {
+	if colInsert {
 		ins.Rows.(sqlparser.Values)[rowNum] = append(ins.Rows.(sqlparser.Values)[rowNum].(sqlparser.ValTuple), &sqlparser.NullVal{})
 	}
-	return ins.Rows.(sqlparser.Values)[rowNum].(sqlparser.ValTuple), pos
+	return ins.Rows.(sqlparser.Values)[rowNum].(sqlparser.ValTuple), pos, colInsert
 }