feat: optimise outer joins

go/test/endtoend/vtgate/queries/misc/misc_test.go

@@ -365,12 +365,12 @@ func TestAliasesInOuterJoinQueries(t *testing.T) {
 	mcmp.Exec("insert into t1(id1, id2) values (1,2), (42,5), (5, 42)")
 	mcmp.Exec("insert into tbl(id, unq_col, nonunq_col) values (1,2,3), (2,5,3), (3, 42, 2)")
 
-	// Check that the select query works as intended and then run it again verifying the column names as well.
-	mcmp.AssertMatches("select t1.id1 as t0, t1.id1 as t1, tbl.unq_col as col from t1 left outer join tbl on t1.id2 = tbl.nonunq_col", `[[INT64(1) INT64(1) INT64(42)] [INT64(5) INT64(5) NULL] [INT64(42) INT64(42) NULL]]`)
+	// Check that the select query works as intended and verifying the column names as well.
 	mcmp.ExecWithColumnCompare("select t1.id1 as t0, t1.id1 as t1, tbl.unq_col as col from t1 left outer join tbl on t1.id2 = tbl.nonunq_col")
-
-	mcmp.AssertMatches("select t1.id1 as t0, t1.id1 as t1, tbl.unq_col as col from t1 left outer join tbl on t1.id2 = tbl.nonunq_col order by t1.id2 limit 2", `[[INT64(1) INT64(1) INT64(42)] [INT64(42) INT64(42) NULL]]`)
 	mcmp.ExecWithColumnCompare("select t1.id1 as t0, t1.id1 as t1, tbl.unq_col as col from t1 left outer join tbl on t1.id2 = tbl.nonunq_col order by t1.id2 limit 2")
+	mcmp.ExecWithColumnCompare("select t1.id1 as t0, t1.id1 as t1, tbl.unq_col as col from t1 left outer join tbl on t1.id2 = tbl.nonunq_col order by t1.id2 limit 2 offset 2")
+	mcmp.ExecWithColumnCompare("select t1.id1 as t0, t1.id1 as t1, count(*) as leCount from t1 left outer join tbl on t1.id2 = tbl.nonunq_col group by 1, t1")
+	mcmp.ExecWithColumnCompare("select t.id1, t.id2, derived.unq_col from t1 t join (select id, unq_col, nonunq_col from tbl) as derived on t.id2 = derived.nonunq_col")
 }
 
 func TestAlterTableWithView(t *testing.T) {

go/test/endtoend/vtgate/queries/misc/schema.sql

@@ -1,7 +1,8 @@
-create table if not exists t1(
-                   id1 bigint,
-                   id2 bigint,
-                   primary key(id1)
+create table t1
+(
+    id1 bigint,
+    id2 bigint,
+    primary key (id1)
 ) Engine=InnoDB;
 
 create table unq_idx
@@ -30,8 +31,8 @@ create table tbl
 
 create table tbl_enum_set
 (
-    id         bigint,
-    enum_col   enum('xsmall', 'small', 'medium', 'large', 'xlarge'),
-    set_col    set('a', 'b', 'c', 'd', 'e', 'f', 'g'),
+    id       bigint,
+    enum_col enum('xsmall', 'small', 'medium', 'large', 'xlarge'),
+    set_col set('a', 'b', 'c', 'd', 'e', 'f', 'g'),
     primary key (id)
 ) Engine = InnoDB;

go/vt/vtgate/engine/limit.go

@@ -40,6 +40,8 @@ type Limit struct {
 	Input  Primitive
 }
 
+var UpperLimitStr = "__upper_limit"
+
 // RouteType returns a description of the query routing type used by the primitive
 func (l *Limit) RouteType() string {
 	return l.Input.RouteType()
@@ -63,7 +65,8 @@ func (l *Limit) TryExecute(ctx context.Context, vcursor VCursor, bindVars map[st
 	}
 	// When offset is present, we hijack the limit value so we can calculate
 	// the offset in memory from the result of the scatter query with count + offset.
-	bindVars["__upper_limit"] = sqltypes.Int64BindVariable(int64(count + offset))
+
+	bindVars[UpperLimitStr] = sqltypes.Int64BindVariable(int64(count + offset))
 
 	result, err := vcursor.ExecutePrimitive(ctx, l.Input, bindVars, wantfields)
 	if err != nil {
@@ -96,7 +99,7 @@ func (l *Limit) TryStreamExecute(ctx context.Context, vcursor VCursor, bindVars
 
 	// When offset is present, we hijack the limit value so we can calculate
 	// the offset in memory from the result of the scatter query with count + offset.
-	bindVars["__upper_limit"] = sqltypes.Int64BindVariable(int64(count + offset))
+	bindVars[UpperLimitStr] = sqltypes.Int64BindVariable(int64(count + offset))
 
 	var mu sync.Mutex
 	err = vcursor.StreamExecutePrimitive(ctx, l.Input, bindVars, wantfields, func(qr *sqltypes.Result) error {

go/vt/vtgate/engine/simple_projection.go

@@ -18,9 +18,13 @@ package engine
 
 import (
 	"context"
+	"fmt"
+	"strconv"
+	"strings"
 
 	"google.golang.org/protobuf/proto"
 
+	"vitess.io/vitess/go/slice"
 	"vitess.io/vitess/go/sqltypes"
 	querypb "vitess.io/vitess/go/vt/proto/query"
 )
@@ -90,6 +94,10 @@ func (sc *SimpleProjection) Inputs() ([]Primitive, []map[string]any) {
 // buildResult builds a new result by pulling the necessary columns from
 // the input in the requested order.
 func (sc *SimpleProjection) buildResult(inner *sqltypes.Result) *sqltypes.Result {
+	if sc.namesOnly() {
+		sc.renameFields(inner.Fields)
+		return inner
+	}
 	result := &sqltypes.Result{Fields: sc.buildFields(inner)}
 	result.Rows = make([][]sqltypes.Value, 0, len(inner.Rows))
 	for _, innerRow := range inner.Rows {
@@ -103,6 +111,10 @@ func (sc *SimpleProjection) buildResult(inner *sqltypes.Result) *sqltypes.Result
 	return result
 }
 
+func (sc *SimpleProjection) namesOnly() bool {
+	return sc.Cols == nil
+}
+
 func (sc *SimpleProjection) buildFields(inner *sqltypes.Result) []*querypb.Field {
 	if len(inner.Fields) == 0 {
 		return nil
@@ -119,20 +131,33 @@ func (sc *SimpleProjection) buildFields(inner *sqltypes.Result) []*querypb.Field
 	return fields
 }
 
+func (sc *SimpleProjection) renameFields(fields []*querypb.Field) {
+	if len(fields) == 0 {
+		return
+	}
+	for idx, name := range sc.ColNames {
+		if sc.ColNames[idx] != "" {
+			fields[idx].Name = name
+		}
+	}
+}
+
 func (sc *SimpleProjection) description() PrimitiveDescription {
-	other := map[string]any{
-		"Columns": sc.Cols,
+	other := map[string]any{}
+	if !sc.namesOnly() {
+		other["Columns"] = strings.Join(slice.Map(sc.Cols, strconv.Itoa), ",")
 	}
-	emptyColNames := true
-	for _, cName := range sc.ColNames {
+
+	var colNames []string
+	for idx, cName := range sc.ColNames {
 		if cName != "" {
-			emptyColNames = false
-			break
+			colNames = append(colNames, fmt.Sprintf("%d:%s", idx, cName))
 		}
 	}
-	if !emptyColNames {
-		other["ColumnNames"] = sc.ColNames
+	if colNames != nil {
+		other["ColumnNames"] = colNames
 	}
+
 	return PrimitiveDescription{
 		OperatorType: "SimpleProjection",
 		Other:        other,

go/vt/vtgate/executor_select_test.go

@@ -2329,7 +2329,7 @@ func TestSelectScatterLimit(t *testing.T) {
 	require.NoError(t, err)
 
 	wantQueries := []*querypb.BoundQuery{{
-		Sql:           "select col1, col2, weight_string(col2) from `user` order by `user`.col2 desc limit :__upper_limit",
+		Sql:           "select col1, col2, weight_string(col2) from `user` order by `user`.col2 desc limit 3",
 		BindVariables: map[string]*querypb.BindVariable{"__upper_limit": sqltypes.Int64BindVariable(3)},
 	}}
 	for _, conn := range conns {
@@ -2401,7 +2401,7 @@ func TestStreamSelectScatterLimit(t *testing.T) {
 	require.NoError(t, err)
 
 	wantQueries := []*querypb.BoundQuery{{
-		Sql:           "select col1, col2, weight_string(col2) from `user` order by `user`.col2 desc limit :__upper_limit",
+		Sql:           "select col1, col2, weight_string(col2) from `user` order by `user`.col2 desc limit 3",
 		BindVariables: map[string]*querypb.BindVariable{"__upper_limit": sqltypes.Int64BindVariable(3)},
 	}}
 	for _, conn := range conns {
@@ -2863,11 +2863,11 @@ func TestEmptyJoinRecursiveStream(t *testing.T) {
 	}
 }
 
-func TestCrossShardSubquery(t *testing.T) {
+func TestCrossShardDerivedTable(t *testing.T) {
 	executor, sbc1, sbc2, _, ctx := createExecutorEnv(t)
 	result1 := []*sqltypes.Result{{
 		Fields: []*querypb.Field{
-			{Name: "id", Type: sqltypes.Int32},
+			{Name: "id1", Type: sqltypes.Int32},
 			{Name: "col", Type: sqltypes.Int32},
 		},
 		InsertID: 0,
@@ -2894,10 +2894,8 @@ func TestCrossShardSubquery(t *testing.T) {
 	}}
 	utils.MustMatch(t, wantQueries, sbc2.Queries)
 
-	wantResult := sqltypes.MakeTestResult(sqltypes.MakeTestFields("id", "int32"), "1")
-	if !result.Equal(wantResult) {
-		t.Errorf("result: %+v, want %+v", result, wantResult)
-	}
+	wantResult := sqltypes.MakeTestResult(sqltypes.MakeTestFields("id1", "int32"), "1")
+	assert.Equal(t, wantResult, result)
 }
 
 func TestSubQueryAndQueryWithLimit(t *testing.T) {
@@ -2946,7 +2944,7 @@ func TestCrossShardSubqueryStream(t *testing.T) {
 	executor, sbc1, sbc2, _, ctx := createExecutorEnv(t)
 	result1 := []*sqltypes.Result{{
 		Fields: []*querypb.Field{
-			{Name: "id", Type: sqltypes.Int32, Charset: collations.CollationBinaryID, Flags: uint32(querypb.MySqlFlag_NUM_FLAG)},
+			{Name: "id1", Type: sqltypes.Int32, Charset: collations.CollationBinaryID, Flags: uint32(querypb.MySqlFlag_NUM_FLAG)},
 			{Name: "col", Type: sqltypes.Int32, Charset: collations.CollationBinaryID, Flags: uint32(querypb.MySqlFlag_NUM_FLAG)},
 		},
 		InsertID: 0,
@@ -2971,18 +2969,16 @@ func TestCrossShardSubqueryStream(t *testing.T) {
 
 	wantResult := &sqltypes.Result{
 		Fields: []*querypb.Field{
-			{Name: "id", Type: sqltypes.Int32, Charset: collations.CollationBinaryID, Flags: uint32(querypb.MySqlFlag_NUM_FLAG)},
+			{Name: "id1", Type: sqltypes.Int32, Charset: collations.CollationBinaryID, Flags: uint32(querypb.MySqlFlag_NUM_FLAG)},
 		},
 		Rows: [][]sqltypes.Value{{
 			sqltypes.NewInt32(1),
 		}},
 	}
-	if !result.Equal(wantResult) {
-		t.Errorf("result: %+v, want %+v", result, wantResult)
-	}
+	assert.Equal(t, wantResult, result)
 }
 
-func TestCrossShardSubqueryGetFields(t *testing.T) {
+func TestCrossShardDerivedTableGetFields(t *testing.T) {
 	executor, sbc1, _, sbclookup, ctx := createExecutorEnv(t)
 	sbclookup.SetResults([]*sqltypes.Result{{
 		Fields: []*querypb.Field{
@@ -2991,7 +2987,7 @@ func TestCrossShardSubqueryGetFields(t *testing.T) {
 	}})
 	result1 := []*sqltypes.Result{{
 		Fields: []*querypb.Field{
-			{Name: "id", Type: sqltypes.Int32, Charset: collations.CollationBinaryID, Flags: uint32(querypb.MySqlFlag_NUM_FLAG)},
+			{Name: "id1", Type: sqltypes.Int32, Charset: collations.CollationBinaryID, Flags: uint32(querypb.MySqlFlag_NUM_FLAG)},
 			{Name: "col", Type: sqltypes.Int32, Charset: collations.CollationBinaryID, Flags: uint32(querypb.MySqlFlag_NUM_FLAG)},
 		},
 	}}
@@ -3015,12 +3011,10 @@ func TestCrossShardSubqueryGetFields(t *testing.T) {
 	wantResult := &sqltypes.Result{
 		Fields: []*querypb.Field{
 			{Name: "col", Type: sqltypes.Int32, Charset: collations.CollationBinaryID, Flags: uint32(querypb.MySqlFlag_NUM_FLAG)},
-			{Name: "id", Type: sqltypes.Int32, Charset: collations.CollationBinaryID, Flags: uint32(querypb.MySqlFlag_NUM_FLAG)},
+			{Name: "id1", Type: sqltypes.Int32, Charset: collations.CollationBinaryID, Flags: uint32(querypb.MySqlFlag_NUM_FLAG)},
 		},
 	}
-	if !result.Equal(wantResult) {
-		t.Errorf("result: %+v, want %+v", result, wantResult)
-	}
+	assert.Equal(t, wantResult, result)
 }
 
 func TestSelectBindvarswithPrepare(t *testing.T) {
@@ -3042,9 +3036,7 @@ func TestSelectBindvarswithPrepare(t *testing.T) {
 		BindVariables: map[string]*querypb.BindVariable{"id": sqltypes.Int64BindVariable(1)},
 	}}
 	utils.MustMatch(t, wantQueries, sbc1.Queries)
-	if sbc2.Queries != nil {
-		t.Errorf("sbc2.Queries: %+v, want nil\n", sbc2.Queries)
-	}
+	assert.Empty(t, sbc2.Queries)
 }
 
 func TestSelectDatabasePrepare(t *testing.T) {
@@ -3908,14 +3900,14 @@ func TestSelectAggregationNoData(t *testing.T) {
 		{
 			sql:         `select count(*) from (select col1, col2 from user limit 2) x`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|1", "int64|int64|int64")),
-			expSandboxQ: "select x.col1, x.col2, 1 from (select col1, col2 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.col1, x.col2, 1 from (select col1, col2 from `user`) as x limit 2",
 			expField:    `[name:"count(*)" type:INT64]`,
 			expRow:      `[[INT64(0)]]`,
 		},
 		{
 			sql:         `select col2, count(*) from (select col1, col2 from user limit 2) x group by col2`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|1|weight_string(col2)", "int64|int64|int64|varbinary")),
-			expSandboxQ: "select x.col1, x.col2, 1, weight_string(x.col2) from (select col1, col2 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.col1, x.col2, 1, weight_string(x.col2) from (select col1, col2 from `user`) as x limit 2",
 			expField:    `[name:"col2" type:INT64 name:"count(*)" type:INT64]`,
 			expRow:      `[]`,
 		},
@@ -4000,70 +3992,70 @@ func TestSelectAggregationData(t *testing.T) {
 		{
 			sql:         `select count(*) from (select col1, col2 from user limit 2) x`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|1", "int64|int64|int64"), "100|200|1", "200|300|1"),
-			expSandboxQ: "select x.col1, x.col2, 1 from (select col1, col2 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.col1, x.col2, 1 from (select col1, col2 from `user`) as x limit 2",
 			expField:    `[name:"count(*)" type:INT64]`,
 			expRow:      `[[INT64(2)]]`,
 		},
 		{
 			sql:         `select col2, count(*) from (select col1, col2 from user limit 9) x group by col2`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|1|weight_string(col2)", "int64|int64|int64|varbinary"), "100|3|1|NULL", "200|2|1|NULL"),
-			expSandboxQ: "select x.col1, x.col2, 1, weight_string(x.col2) from (select col1, col2 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.col1, x.col2, 1, weight_string(x.col2) from (select col1, col2 from `user`) as x limit 9",
 			expField:    `[name:"col2" type:INT64 name:"count(*)" type:INT64]`,
 			expRow:      `[[INT64(2) INT64(4)] [INT64(3) INT64(5)]]`,
 		},
 		{
 			sql:         `select count(col1) from (select id, col1 from user limit 2) x`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("id|col1", "int64|varchar"), "1|a", "2|b"),
-			expSandboxQ: "select x.id, x.col1 from (select id, col1 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.id, x.col1 from (select id, col1 from `user`) as x limit 2",
 			expField:    `[name:"count(col1)" type:INT64]`,
 			expRow:      `[[INT64(2)]]`,
 		},
 		{
 			sql:         `select count(col1), col2 from (select col2, col1 from user limit 9) x group by col2`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col2|col1|weight_string(col2)", "int64|varchar|varbinary"), "3|a|NULL", "2|b|NULL"),
-			expSandboxQ: "select x.col2, x.col1, weight_string(x.col2) from (select col2, col1 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.col2, x.col1, weight_string(x.col2) from (select col2, col1 from `user`) as x limit 9",
 			expField:    `[name:"count(col1)" type:INT64 name:"col2" type:INT64]`,
 			expRow:      `[[INT64(4) INT64(2)] [INT64(5) INT64(3)]]`,
 		},
 		{
 			sql:         `select col1, count(col2) from (select col1, col2 from user limit 9) x group by col1`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|int64|varbinary"), "a|1|a", "b|null|b"),
-			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit 9",
 			expField:    `[name:"col1" type:VARCHAR name:"count(col2)" type:INT64]`,
 			expRow:      `[[VARCHAR("a") INT64(5)] [VARCHAR("b") INT64(0)]]`,
 		},
 		{
 			sql:         `select col1, count(col2) from (select col1, col2 from user limit 32) x group by col1`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|int64|varbinary"), "null|1|null", "null|null|null", "a|1|a", "b|null|b"),
-			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit 32",
 			expField:    `[name:"col1" type:VARCHAR name:"count(col2)" type:INT64]`,
 			expRow:      `[[NULL INT64(8)] [VARCHAR("a") INT64(8)] [VARCHAR("b") INT64(0)]]`,
 		},
 		{
 			sql:         `select col1, sum(col2) from (select col1, col2 from user limit 4) x group by col1`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|int64|varbinary"), "a|3|a"),
-			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit 4",
 			expField:    `[name:"col1" type:VARCHAR name:"sum(col2)" type:DECIMAL]`,
 			expRow:      `[[VARCHAR("a") DECIMAL(12)]]`,
 		},
 		{
 			sql:         `select col1, sum(col2) from (select col1, col2 from user limit 4) x group by col1`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|varchar|varbinary"), "a|2|a"),
-			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit 4",
 			expField:    `[name:"col1" type:VARCHAR name:"sum(col2)" type:FLOAT64]`,
 			expRow:      `[[VARCHAR("a") FLOAT64(8)]]`,
 		},
 		{
 			sql:         `select col1, sum(col2) from (select col1, col2 from user limit 4) x group by col1`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|varchar|varbinary"), "a|x|a"),
-			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit 4",
 			expField:    `[name:"col1" type:VARCHAR name:"sum(col2)" type:FLOAT64]`,
 			expRow:      `[[VARCHAR("a") FLOAT64(0)]]`,
 		},
 		{
 			sql:         `select col1, sum(col2) from (select col1, col2 from user limit 4) x group by col1`,
 			sandboxRes:  sqltypes.MakeTestResult(sqltypes.MakeTestFields("col1|col2|weight_string(col1)", "varchar|varchar|varbinary"), "a|null|a"),
-			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit :__upper_limit",
+			expSandboxQ: "select x.col1, x.col2, weight_string(x.col1) from (select col1, col2 from `user`) as x limit 4",
 			expField:    `[name:"col1" type:VARCHAR name:"sum(col2)" type:FLOAT64]`,
 			expRow:      `[[VARCHAR("a") NULL]]`,
 		},

go/vt/vtgate/planbuilder/operator_transformers.go

@@ -363,7 +363,10 @@ func transformProjection(ctx *plancontext.PlanningContext, op *operators.Project
 	if cols, colNames := op.AllOffsets(); cols != nil {
 		// if all this op is doing is passing through columns from the input, we
 		// can use the faster SimpleProjection
-		return useSimpleProjection(cols, colNames, src)
+		if len(op.Source.GetColumns(ctx)) == len(cols) && offsetInInputOrder(cols) {
+			cols = nil
+		}
+		return newSimpleProjection(cols, colNames, src)
 	}
 
 	ap, err := op.GetAliasedProjections()
@@ -393,6 +396,16 @@ func transformProjection(ctx *plancontext.PlanningContext, op *operators.Project
 	}, nil
 }
 
+// offsetInInputOrder returns true if the columns are in the same order as the input
+func offsetInInputOrder(cols []int) bool {
+	for i, c := range cols {
+		if c != i {
+			return false
+		}
+	}
+	return true
+}
+
 func getEvalEngingeExpr(ctx *plancontext.PlanningContext, pe *operators.ProjExpr) (evalengine.Expr, error) {
 	switch e := pe.Info.(type) {
 	case *operators.EvalEngine:
@@ -406,9 +419,8 @@ func getEvalEngingeExpr(ctx *plancontext.PlanningContext, pe *operators.ProjExpr
 
 }
 
-// useSimpleProjection uses nothing at all if the output is already correct,
-// or SimpleProjection when we have to reorder or truncate the columns
-func useSimpleProjection(cols []int, colNames []string, src logicalPlan) (logicalPlan, error) {
+// newSimpleProjection creates a simple projections
+func newSimpleProjection(cols []int, colNames []string, src logicalPlan) (logicalPlan, error) {
 	return &simpleProjection{
 		logicalPlanCommon: newBuilderCommon(src),
 		eSimpleProj: &engine.SimpleProjection{