project too

Author: James Cor

sql/expression/function/aggregation/group_concat.go

@@ -308,7 +308,7 @@ func (g *groupConcatBuffer) Update(ctx *sql.Context, originalRow sql.Row) error
 
 	// Append the current value to the end of the row. We want to preserve the row's original structure
 	// for sort ordering in the final step.
-	g.rows = append(g.rows, append(originalRow, vs))
+	g.rows = append(g.rows, append(originalRow.Copy(), vs))
 
 	return nil
 }

sql/rowexec/ddl_iters.go

@@ -918,7 +918,7 @@ func (i *loggingKeyValueIter) Close(ctx *sql.Context) error {
 // projectRowWithTypes projects the row given with the projections given and additionally converts them to the
 // corresponding types found in the schema given, using the standard type conversion logic.
 func projectRowWithTypes(ctx *sql.Context, oldSchema, newSchema sql.Schema, projections []sql.Expression, r sql.Row) (sql.Row, error) {
-	newRow, err := ProjectRow(ctx, projections, r)
+	newRow, err := ProjectRow(ctx, projections, r, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -1440,7 +1440,7 @@ func (i *addColumnIter) rewriteTable(ctx *sql.Context, rwt sql.RewritableTable)
 			return false, err
 		}
 
-		newRow, err := ProjectRow(ctx, projections, r)
+		newRow, err := ProjectRow(ctx, projections, r, nil)
 		if err != nil {
 			_ = inserter.DiscardChanges(ctx, err)
 			_ = inserter.Close(ctx)
@@ -1736,7 +1736,7 @@ func (i *dropColumnIter) rewriteTable(ctx *sql.Context, rwt sql.RewritableTable)
 			return false, err
 		}
 
-		newRow, err := ProjectRow(ctx, projections, r)
+		newRow, err := ProjectRow(ctx, projections, r, nil)
 		if err != nil {
 			_ = inserter.DiscardChanges(ctx, err)
 			_ = inserter.Close(ctx)
@@ -2240,7 +2240,7 @@ func buildIndex(ctx *sql.Context, n *plan.AlterIndex, ibt sql.IndexBuildingTable
 		}
 
 		if isVirtual {
-			r, err = ProjectRow(ctx, projections, r)
+			r, err = ProjectRow(ctx, projections, r, nil)
 			if err != nil {
 				return err
 			}
@@ -2326,7 +2326,7 @@ func rewriteTableForIndexCreate(ctx *sql.Context, n *plan.AlterIndex, table sql.
 		}
 
 		if isVirtual {
-			r, err = ProjectRow(ctx, projections, r)
+			r, err = ProjectRow(ctx, projections, r, nil)
 			if err != nil {
 				return err
 			}

sql/rowexec/insert.go

@@ -67,15 +67,16 @@ func getInsertExpressions(values sql.Node) []sql.Expression {
 }
 
 func (i *insertIter) Next(ctx *sql.Context) (returnRow sql.Row, returnErr error) {
-	row, err := i.rowSource.Next(ctx)
+	origRow, err := i.rowSource.Next(ctx)
 	if err == io.EOF {
 		return nil, err
 	}
-
 	if err != nil {
-		return nil, i.ignoreOrClose(ctx, row, err)
+		return nil, i.ignoreOrClose(ctx, origRow, err)
 	}
 
+	row := origRow.Copy()
+
 	// Increment row number for error reporting (MySQL starts at 1)
 	i.rowNumber++
 
@@ -107,9 +108,6 @@ func (i *insertIter) Next(ctx *sql.Context) (returnRow sql.Row, returnErr error)
 		return nil, i.ignoreOrClose(ctx, row, err)
 	}
 
-	origRow := make(sql.Row, len(row))
-	copy(origRow, row)
-
 	// Do any necessary type conversions to the target schema
 	for idx, col := range i.schema {
 		if row[idx] != nil {

sql/rowexec/join_iters.go

@@ -214,8 +214,8 @@ func (i *joinIter) buildRow(primary, secondary sql.Row) sql.Row {
 }
 
 func (i *joinIter) Close(ctx *sql.Context) (err error) {
-	i.rowBuffer.Reset()
-	sql.RowBufPool.Put(i.rowBuffer)
+	//i.rowBuffer.Reset()
+	//sql.RowBufPool.Put(i.rowBuffer)
 
 	if i.primary != nil {
 		if err = i.primary.Close(ctx); err != nil {

sql/rowexec/merge_join.go

@@ -320,7 +320,7 @@ func (i *mergeJoinIter) Next(ctx *sql.Context) (sql.Row, error) {
 }
 
 func (i *mergeJoinIter) copyReturnRow() sql.Row {
-	ret := i.rowBuffer.Get(len(i.fullRow))
+	ret := make(sql.Row, len(i.fullRow))
 	copy(ret, i.fullRow)
 	return ret
 }
@@ -580,6 +580,9 @@ func (i *mergeJoinIter) removeParentRow(r sql.Row) sql.Row {
 }
 
 func (i *mergeJoinIter) Close(ctx *sql.Context) (err error) {
+	i.rowBuffer.Reset()
+	sql.RowBufPool.Put(i.rowBuffer)
+
 	if i.left != nil {
 		err = i.left.Close(ctx)
 	}

sql/rowexec/rel.go

@@ -97,7 +97,7 @@ func (b *BaseBuilder) buildValues(ctx *sql.Context, n *plan.Values, row sql.Row)
 			// For the values node, the relevant values to evaluate are the tuple itself. We may need to project
 			// DEFAULT values onto them, which ProjectRow handles correctly (could require multiple passes)
 			var err error
-			vals, err = ProjectRow(ctx, et, vals)
+			vals, err = ProjectRow(ctx, et, vals, nil)
 			if err != nil {
 				return nil, err
 			}
@@ -315,6 +315,7 @@ func (b *BaseBuilder) buildProject(ctx *sql.Context, n *plan.Project, row sql.Ro
 		canDefer:       n.CanDefer,
 		hasNestedIters: n.IncludesNestedIters,
 		childIter:      i,
+		rowBuffer:      sql.RowBufPool.Get().(*sql.RowBuffer),
 	}), nil
 }
 
@@ -326,6 +327,7 @@ func (b *BaseBuilder) buildVirtualColumnTable(ctx *sql.Context, n *plan.VirtualC
 	return sql.NewSpanIter(span, &ProjectIter{
 		projs:     n.Projections,
 		childIter: tableIter,
+		rowBuffer: sql.RowBufPool.Get().(*sql.RowBuffer),
 	}), nil
 }
 

sql/rowexec/rel_iters.go

@@ -132,6 +132,8 @@ type ProjectIter struct {
 	hasNestedIters bool
 	nestedState    *nestedIterState
 	childIter      sql.RowIter
+
+	rowBuffer *sql.RowBuffer
 }
 
 type nestedIterState struct {
@@ -150,10 +152,12 @@ func (i *ProjectIter) Next(ctx *sql.Context) (sql.Row, error) {
 		return nil, err
 	}
 
-	return ProjectRow(ctx, i.projs, childRow)
+	return ProjectRow(ctx, i.projs, childRow, i.rowBuffer)
 }
 
 func (i *ProjectIter) Close(ctx *sql.Context) error {
+	i.rowBuffer.Reset()
+	sql.RowBufPool.Put(i.rowBuffer)
 	return i.childIter.Close(ctx)
 }
 
@@ -178,7 +182,7 @@ func (i *ProjectIter) ProjectRowWithNestedIters(
 	// Other iterator values will be NULL after they are depleted. All non-iterator fields for the row are returned
 	// identically for each row in the result set.
 	if i.nestedState != nil {
-		row, err := ProjectRow(ctx, i.nestedState.projections, i.nestedState.sourceRow)
+		row, err := ProjectRow(ctx, i.nestedState.projections, i.nestedState.sourceRow, nil)
 		if err != nil {
 			return nil, err
 		}
@@ -304,8 +308,14 @@ func ProjectRow(
 	ctx *sql.Context,
 	projections []sql.Expression,
 	row sql.Row,
+	rowBuffer *sql.RowBuffer,
 ) (sql.Row, error) {
-	var fields = make(sql.Row, len(projections))
+	var fields sql.Row
+	if rowBuffer != nil {
+		fields = rowBuffer.Get(len(projections))
+	} else {
+		fields = make(sql.Row, len(projections))
+	}
 	var secondPass []int
 	for i, expr := range projections {
 		// Default values that are expressions may reference other fields, thus they must evaluate after all other exprs.

sql/rows.go

@@ -226,7 +226,11 @@ func rowFromRow2(sch Schema, r Row2) Row {
 
 // RowsToRowIter creates a RowIter that iterates over the given rows.
 func RowsToRowIter(rows ...Row) RowIter {
-	return &sliceRowIter{rows: rows}
+	allRows := make([]Row, len(rows))
+	for i, row := range rows {
+		allRows[i] = row.Copy()
+	}
+	return &sliceRowIter{rows: allRows}
 }
 
 type sliceRowIter struct {
@@ -238,10 +242,9 @@ func (i *sliceRowIter) Next(*Context) (Row, error) {
 	if i.idx >= len(i.rows) {
 		return nil, io.EOF
 	}
-
 	r := i.rows[i.idx]
 	i.idx++
-	return r.Copy(), nil
+	return r, nil
 }
 
 func (i *sliceRowIter) Close(*Context) error {