Repository: vitessio/vitess
Pull request #9874 (2022-03-11 18:27:30)

feat: add weightstring for distinct
## Description
When the planner is missing schema information for tables, we might be missing collation information for columns. To be able to correctly hash these, we fall back on using `weightstring()` for the columns that are missing collation information.
This avoids the error "text type with an unknown/unsupported collation cannot be hashed" that would sometimes show up.

## Checklist
- [x] Should this PR be backported?
- [x] Tests were added or are not required
- [x] Documentation was added or is not required


Patch:
diff --git a/go/vt/vtgate/engine/cached_size.go b/go/vt/vtgate/engine/cached_size.go
index 51fc58bb76..58cda04172 100644
--- a/go/vt/vtgate/engine/cached_size.go
+++ b/go/vt/vtgate/engine/cached_size.go
@@ -59,7 +59,7 @@ func (cached *AlterVSchema) CachedSize(alloc bool) int64 {
 	size += cached.AlterVschemaDDL.CachedSize(true)
 	return size
 }
-func (cached *Concatenate) CachedSize(alloc bool) int64 {
+func (cached *CheckCol) CachedSize(alloc bool) int64 {
 	if cached == nil {
 		return int64(0)
 	}
@@ -67,6 +67,20 @@ func (cached *Concatenate) CachedSize(alloc bool) int64 {
 	if alloc {
 		size += int64(24)
 	}
+	// field WsCol *int
+	size += hack.RuntimeAllocSize(int64(8))
+	return size
+}
+
+//go:nocheckptr
+func (cached *Concatenate) CachedSize(alloc bool) int64 {
+	if cached == nil {
+		return int64(0)
+	}
+	size := int64(0)
+	if alloc {
+		size += int64(32)
+	}
 	// field Sources []vitess.io/vitess/go/vt/vtgate/engine.Primitive
 	{
 		size += hack.RuntimeAllocSize(int64(cap(cached.Sources)) * int64(16))
@@ -76,6 +90,17 @@ func (cached *Concatenate) CachedSize(alloc bool) int64 {
 			}
 		}
 	}
+	// field NoNeedToTypeCheck map[int]any
+	if cached.NoNeedToTypeCheck != nil {
+		size += int64(48)
+		hmap := reflect.ValueOf(cached.NoNeedToTypeCheck)
+		numBuckets := int(math.Pow(2, float64((*(*uint8)(unsafe.Pointer(hmap.Pointer() + uintptr(9)))))))
+		numOldBuckets := (*(*uint16)(unsafe.Pointer(hmap.Pointer() + uintptr(10))))
+		size += hack.RuntimeAllocSize(int64(numOldBuckets * 208))
+		if len(cached.NoNeedToTypeCheck) > 0 || numBuckets > 1 {
+			size += hack.RuntimeAllocSize(int64(numBuckets * 208))
+		}
+	}
 	return size
 }
 func (cached *DBDDL) CachedSize(alloc bool) int64 {
@@ -158,9 +183,26 @@ func (cached *Distinct) CachedSize(alloc bool) int64 {
 	if cc, ok := cached.Source.(cachedObject); ok {
 		size += cc.CachedSize(true)
 	}
-	// field ColCollations []vitess.io/vitess/go/mysql/collations.ID
+	// field CheckCols []vitess.io/vitess/go/vt/vtgate/engine.CheckCol
 	{
-		size += hack.RuntimeAllocSize(int64(cap(cached.ColCollations)) * int64(2))
+		size += hack.RuntimeAllocSize(int64(cap(cached.CheckCols)) * int64(18))
+		for _, elem := range cached.CheckCols {
+			size += elem.CachedSize(false)
+		}
+	}
+	return size
+}
+func (cached *DistinctV3) CachedSize(alloc bool) int64 {
+	if cached == nil {
+		return int64(0)
+	}
+	size := int64(0)
+	if alloc {
+		size += int64(16)
+	}
+	// field Source vitess.io/vitess/go/vt/vtgate/engine.Primitive
+	if cc, ok := cached.Source.(cachedObject); ok {
+		size += cc.CachedSize(true)
 	}
 	return size
 }
diff --git a/go/vt/vtgate/engine/concatenate.go b/go/vt/vtgate/engine/concatenate.go
index 1e8eb5b3f1..a2f8dcc7ae 100644
--- a/go/vt/vtgate/engine/concatenate.go
+++ b/go/vt/vtgate/engine/concatenate.go
@@ -31,6 +31,24 @@ var _ Primitive = (*Concatenate)(nil)
 // Concatenate specified the parameter for concatenate primitive
 type Concatenate struct {
 	Sources []Primitive
+
+	// These column offsets do not need to be typed checked - they usually contain weight_string()
+	// columns that are not going to be returned to the user
+	NoNeedToTypeCheck map[int]any
+}
+
+// NewConcatenate creates a Concatenate primitive. The ignoreCols slice contains the offsets that
+// don't need to have the same type between sources -
+// weight_string() sometimes returns VARBINARY and sometimes VARCHAR
+func NewConcatenate(Sources []Primitive, ignoreCols []int) *Concatenate {
+	ignoreTypes := map[int]any{}
+	for _, i := range ignoreCols {
+		ignoreTypes[i] = nil
+	}
+	return &Concatenate{
+		Sources:           Sources,
+		NoNeedToTypeCheck: ignoreTypes,
+	}
 }
 
 // RouteType returns a description of the query routing type used by the primitive
@@ -101,23 +119,28 @@ func (c *Concatenate) TryExecute(vcursor VCursor, bindVars map[string]*querypb.B
 }
 
 func (c *Concatenate) getFields(res []*sqltypes.Result) ([]*querypb.Field, error) {
-	var resFields []*querypb.Field
+	if len(res) == 0 {
+		return nil, nil
+	}
+
+	var fields []*querypb.Field
 	for _, r := range res {
-		fields := r.Fields
-		if fields == nil {
+		if r.Fields == nil {
 			continue
 		}
-		if resFields == nil {
-			resFields = fields
+		if fields == nil {
+			fields = r.Fields
 			continue
 		}
-		err := compareFields(fields, resFields)
+
+		err := c.compareFields(fields, r.Fields)
 		if err != nil {
 			return nil, err
 		}
 	}
-	return resFields, nil
+	return fields, nil
 }
+
 func (c *Concatenate) execSources(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) ([]*sqltypes.Result, error) {
 	results := make([]*sqltypes.Result, len(c.Sources))
 	g, restoreCtx := vcursor.ErrorGroupCancellableContext()
@@ -172,7 +195,7 @@ func (c *Concatenate) TryStreamExecute(vcursor VCursor, bindVars map[string]*que
 				}
 				wg.Wait()
 				if resultChunk.Fields != nil {
-					err := compareFields(seenFields, resultChunk.Fields)
+					err := c.compareFields(seenFields, resultChunk.Fields)
 					if err != nil {
 						return err
 					}
@@ -209,6 +232,7 @@ func (c *Concatenate) TryStreamExecute(vcursor VCursor, bindVars map[string]*que
 
 // GetFields fetches the field info.
 func (c *Concatenate) GetFields(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
+	// TODO: type coercions
 	res, err := c.Sources[0].GetFields(vcursor, bindVars)
 	if err != nil {
 		return nil, err
@@ -219,11 +243,12 @@ func (c *Concatenate) GetFields(vcursor VCursor, bindVars map[string]*querypb.Bi
 		if err != nil {
 			return nil, err
 		}
-		err = compareFields(result.Fields, res.Fields)
+		err = c.compareFields(res.Fields, result.Fields)
 		if err != nil {
 			return nil, err
 		}
 	}
+
 	return res, nil
 }
 
@@ -246,12 +271,15 @@ func (c *Concatenate) description() PrimitiveDescription {
 	return PrimitiveDescription{OperatorType: c.RouteType()}
 }
 
-func compareFields(fields1 []*querypb.Field, fields2 []*querypb.Field) error {
+func (c *Concatenate) compareFields(fields1 []*querypb.Field, fields2 []*querypb.Field) error {
 	if len(fields1) != len(fields2) {
 		return ErrWrongNumberOfColumnsInSelect
 	}
-	for i, field2 := range fields2 {
-		field1 := fields1[i]
+	for i, field1 := range fields1 {
+		if _, found := c.NoNeedToTypeCheck[i]; found {
+			continue
+		}
+		field2 := fields2[i]
 		if field1.Type != field2.Type {
 			return vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "merging field of different types is not supported, name: (%v, %v) types: (%v, %v)", field1.Name, field2.Name, field1.Type, field2.Type)
 		}
diff --git a/go/vt/vtgate/engine/distinct.go b/go/vt/vtgate/engine/distinct.go
index 4ed9f34384..026b359eb8 100644
--- a/go/vt/vtgate/engine/distinct.go
+++ b/go/vt/vtgate/engine/distinct.go
@@ -17,32 +17,40 @@ limitations under the License.
 package engine
 
 import (
+	"fmt"
+
 	"vitess.io/vitess/go/mysql/collations"
 	"vitess.io/vitess/go/sqltypes"
 	querypb "vitess.io/vitess/go/vt/proto/query"
+	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
+	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vtgate/evalengine"
 )
 
 // Distinct Primitive is used to uniqueify results
 var _ Primitive = (*Distinct)(nil)
 
-// Distinct Primitive is used to uniqueify results
-type Distinct struct {
-	Source        Primitive
-	ColCollations []collations.ID
-}
-
-type row = []sqltypes.Value
-
-type probeTable struct {
-	seenRows      map[evalengine.HashCode][]row
-	colCollations []collations.ID
-}
+type (
+	// Distinct Primitive is used to uniqueify results
+	Distinct struct {
+		Source    Primitive
+		CheckCols []CheckCol
+		Truncate  bool
+	}
+	CheckCol struct {
+		Col       int
+		WsCol     *int
+		Collation collations.ID
+	}
+	probeTable struct {
+		seenRows  map[evalengine.HashCode][]sqltypes.Row
+		checkCols []CheckCol
+	}
+)
 
-func (pt *probeTable) exists(inputRow row) (bool, error) {
+func (pt *probeTable) exists(inputRow sqltypes.Row) (bool, error) {
 	// the two prime numbers used here (17 and 31) are used to
-
-	// calculate hashcode from all column values in the input row
+	// calculate hashcode from all column values in the input sqltypes.Row
 	code, err := pt.hashCodeForRow(inputRow)
 	if err != nil {
 		return false, err
@@ -50,15 +58,15 @@ func (pt *probeTable) exists(inputRow row) (bool, error) {
 
 	existingRows, found := pt.seenRows[code]
 	if !found {
-		// nothing with this hash code found, we can be sure it's a not seen row
-		pt.seenRows[code] = []row{inputRow}
+		// nothing with this hash code found, we can be sure it's a not seen sqltypes.Row
+		pt.seenRows[code] = []sqltypes.Row{inputRow}
 		return false, nil
 	}
 
 	// we found something in the map - still need to check all individual values
 	// so we don't just fall for a hash collision
 	for _, existingRow := range existingRows {
-		exists, err := equal(existingRow, inputRow, pt.colCollations)
+		exists, err := pt.equal(existingRow, inputRow)
 		if err != nil {
 			return false, err
 		}
@@ -72,7 +80,7 @@ func (pt *probeTable) exists(inputRow row) (bool, error) {
 	return false, nil
 }
 
-func (pt *probeTable) hashCodeForRow(inputRow row) (evalengine.HashCode, error) {
+func (pt *probeTable) hashCodeForRow(inputRow sqltypes.Row) (evalengine.HashCode, error) {
 	// Why use 17 and 31 in this method?
 	// Copied from an old usenet discussion on the topic:
 	// https://groups.google.com/g/comp.programming/c/HSurZEyrZ1E?pli=1#d887b5bdb2dac99d
@@ -105,32 +113,42 @@ func (pt *probeTable) hashCodeForRow(inputRow row) (evalengine.HashCode, error)
 	// > So you want an odd MULT that has plenty of one-bits.
 
 	code := evalengine.HashCode(17)
-	for idx, value := range inputRow {
-		// We use unknown collations when we do not have collation information
-		// This is safe for types which do not require collation information like
-		// numeric types. It will fail at runtime for text types.
-		collation := collations.Unknown
-		if len(pt.colCollations) > idx {
-			collation = pt.colCollations[idx]
+	for i, checkCol := range pt.checkCols {
+		if i >= len(inputRow) {
+			return 0, vterrors.New(vtrpcpb.Code_INTERNAL, "distinct check colls is larger than its input row")
 		}
-		hashcode, err := evalengine.NullsafeHashcode(value, collation, value.Type())
+		col := inputRow[checkCol.Col]
+		hashcode, err := evalengine.NullsafeHashcode(col, checkCol.Collation, col.Type())
 		if err != nil {
-			return 0, err
+			if err != evalengine.UnsupportedCollationHashError || checkCol.WsCol == nil {
+				return 0, err
+			}
+			checkCol = checkCol.SwitchToWeightString()
+			pt.checkCols[i] = checkCol
+			hashcode, err = evalengine.NullsafeHashcode(inputRow[checkCol.Col], checkCol.Collation, col.Type())
+			if err != nil {
+				return 0, err
+			}
 		}
 		code = code*31 + hashcode
 	}
 	return code, nil
 }
 
-func equal(a, b []sqltypes.Value, colCollations []collations.ID) (bool, error) {
-	for i, aVal := range a {
-		collation := collations.Unknown
-		if len(colCollations) > i {
-			collation = colCollations[i]
-		}
-		cmp, err := evalengine.NullsafeCompare(aVal, b[i], collation)
+func (pt *probeTable) equal(a, b sqltypes.Row) (bool, error) {
+	for i, checkCol := range pt.checkCols {
+		cmp, err := evalengine.NullsafeCompare(a[i], b[i], checkCol.Collation)
 		if err != nil {
-			return false, err
+			_, isComparisonErr := err.(evalengine.UnsupportedComparisonError)
+			if !isComparisonErr || checkCol.WsCol == nil {
+				return false, err
+			}
+			checkCol = checkCol.SwitchToWeightString()
+			pt.checkCols[i] = checkCol
+			cmp, err = evalengine.NullsafeCompare(a[i], b[i], checkCol.Collation)
+			if err != nil {
+				return false, err
+			}
 		}
 		if cmp != 0 {
 			return false, nil
@@ -139,10 +157,12 @@ func equal(a, b []sqltypes.Value, colCollations []collations.ID) (bool, error) {
 	return true, nil
 }
 
-func newProbeTable(colCollations []collations.ID) *probeTable {
+func newProbeTable(checkCols []CheckCol) *probeTable {
+	cols := make([]CheckCol, len(checkCols))
+	copy(cols, checkCols)
 	return &probeTable{
-		seenRows:      map[uintptr][]row{},
-		colCollations: colCollations,
+		seenRows:  map[uintptr][]sqltypes.Row{},
+		checkCols: cols,
 	}
 }
 
@@ -158,7 +178,7 @@ func (d *Distinct) TryExecute(vcursor VCursor, bindVars map[string]*querypb.Bind
 		InsertID: input.InsertID,
 	}
 
-	pt := newProbeTable(d.ColCollations)
+	pt := newProbeTable(d.CheckCols)
 
 	for _, row := range input.Rows {
 		exists, err := pt.exists(row)
@@ -169,13 +189,15 @@ func (d *Distinct) TryExecute(vcursor VCursor, bindVars map[string]*querypb.Bind
 			result.Rows = append(result.Rows, row)
 		}
 	}
-
+	if d.Truncate {
+		return result.Truncate(len(d.CheckCols)), nil
+	}
 	return result, err
 }
 
 // TryStreamExecute implements the Primitive interface
 func (d *Distinct) TryStreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
-	pt := newProbeTable(d.ColCollations)
+	pt := newProbeTable(d.CheckCols)
 
 	err := vcursor.StreamExecutePrimitive(d.Source, bindVars, wantfields, func(input *sqltypes.Result) error {
 		result := &sqltypes.Result{
@@ -191,7 +213,7 @@ func (d *Distinct) TryStreamExecute(vcursor VCursor, bindVars map[string]*queryp
 				result.Rows = append(result.Rows, row)
 			}
 		}
-		return callback(result)
+		return callback(result.Truncate(len(d.CheckCols)))
 	})
 
 	return err
@@ -228,26 +250,46 @@ func (d *Distinct) Inputs() []Primitive {
 }
 
 func (d *Distinct) description() PrimitiveDescription {
-	var other map[string]any
-	if d.ColCollations != nil {
-		allUnknown := true
-		other = map[string]any{}
-		var colls []string
-		for _, collation := range d.ColCollations {
-			coll := collations.Local().LookupByID(collation)
-			if coll == nil {
-				colls = append(colls, "UNKNOWN")
-			} else {
-				colls = append(colls, coll.Name())
-				allUnknown = false
-			}
-		}
-		if !allUnknown {
-			other["Collations"] = colls
-		}
+	other := map[string]any{}
+
+	var colls []string
+	for _, checkCol := range d.CheckCols {
+		colls = append(colls, checkCol.String())
+	}
+	if colls != nil {
+		other["Collations"] = colls
+	}
+
+	if d.Truncate {
+		other["ResultColumns"] = len(d.CheckCols)
 	}
 	return PrimitiveDescription{
 		Other:        other,
 		OperatorType: "Distinct",
 	}
 }
+
+// SwitchToWeightString returns a new CheckCol that works on the weight string column instead
+func (cc CheckCol) SwitchToWeightString() CheckCol {
+	return CheckCol{
+		Col:       *cc.WsCol,
+		WsCol:     nil,
+		Collation: collations.CollationBinaryID,
+	}
+}
+
+func (cc CheckCol) String() string {
+	coll := collations.Local().LookupByID(cc.Collation)
+	var collation string
+	if coll != nil {
+		collation = ": " + coll.Name()
+	}
+
+	var column string
+	if cc.WsCol == nil {
+		column = fmt.Sprintf("%d", cc.Col)
+	} else {
+		column = fmt.Sprintf("(%d:%d)", cc.Col, *cc.WsCol)
+	}
+	return column + collation
+}
diff --git a/go/vt/vtgate/engine/distinctV3.go b/go/vt/vtgate/engine/distinctV3.go
new file mode 100644
index 0000000000..9992c4def7
--- /dev/null
+++ b/go/vt/vtgate/engine/distinctV3.go
@@ -0,0 +1,178 @@
+/*
+Copyright 2020 The Vitess Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package engine
+
+import (
+	"vitess.io/vitess/go/mysql/collations"
+	"vitess.io/vitess/go/sqltypes"
+	querypb "vitess.io/vitess/go/vt/proto/query"
+	"vitess.io/vitess/go/vt/vtgate/evalengine"
+)
+
+// DistinctV3 Primitive is used to uniqueify results
+// It does not always work, and should be removed once the V3 planner has been removed
+var _ Primitive = (*DistinctV3)(nil)
+
+// Distinct Primitive is used to uniqueify results
+type DistinctV3 struct {
+	Source Primitive
+}
+
+type row = []sqltypes.Value
+
+type probeTableV3 struct {
+	m map[evalengine.HashCode][]row
+}
+
+func (pt *probeTableV3) exists(inputRow row) (bool, error) {
+	// calculate hashcode from all column values in the input row
+	code := evalengine.HashCode(17)
+	for _, value := range inputRow {
+		hashcode, err := evalengine.NullsafeHashcode(value, collations.Unknown, value.Type())
+		if err != nil {
+			return false, err
+		}
+		code = code*31 + hashcode
+	}
+
+	existingRows, found := pt.m[code]
+	if !found {
+		// nothing with this hash code found, we can be sure it's a not seen row
+		pt.m[code] = []row{inputRow}
+		return false, nil
+	}
+
+	// we found something in the map - still need to check all individual values
+	// so we don't just fall for a hash collision
+	for _, existingRow := range existingRows {
+		exists, err := equalV3(existingRow, inputRow)
+		if err != nil {
+			return false, err
+		}
+		if exists {
+			return true, nil
+		}
+	}
+
+	pt.m[code] = append(existingRows, inputRow)
+
+	return false, nil
+}
+
+func equalV3(a, b []sqltypes.Value) (bool, error) {
+	for i, aVal := range a {
+		cmp, err := evalengine.NullsafeCompare(aVal, b[i], collations.Unknown)
+		if err != nil {
+			return false, err
+		}
+		if cmp != 0 {
+			return false, nil
+		}
+	}
+	return true, nil
+}
+
+func newProbeTableV3() *probeTableV3 {
+	return &probeTableV3{m: map[evalengine.HashCode][]row{}}
+}
+
+// TryExecute implements the Primitive interface
+func (d *DistinctV3) TryExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool) (*sqltypes.Result, error) {
+	input, err := vcursor.ExecutePrimitive(d.Source, bindVars, wantfields)
+	if err != nil {
+		return nil, err
+	}
+
+	result := &sqltypes.Result{
+		Fields:   input.Fields,
+		InsertID: input.InsertID,
+	}
+
+	pt := newProbeTableV3()
+
+	for _, row := range input.Rows {
+		exists, err := pt.exists(row)
+		if err != nil {
+			return nil, err
+		}
+		if !exists {
+			result.Rows = append(result.Rows, row)
+		}
+	}
+
+	return result, err
+}
+
+// TryStreamExecute implements the Primitive interface
+func (d *DistinctV3) TryStreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
+	pt := newProbeTableV3()
+
+	err := vcursor.StreamExecutePrimitive(d.Source, bindVars, wantfields, func(input *sqltypes.Result) error {
+		result := &sqltypes.Result{
+			Fields:   input.Fields,
+			InsertID: input.InsertID,
+		}
+		for _, row := range input.Rows {
+			exists, err := pt.exists(row)
+			if err != nil {
+				return err
+			}
+			if !exists {
+				result.Rows = append(result.Rows, row)
+			}
+		}
+		return callback(result)
+	})
+
+	return err
+}
+
+// RouteType implements the Primitive interface
+func (d *DistinctV3) RouteType() string {
+	return d.Source.RouteType()
+}
+
+// GetKeyspaceName implements the Primitive interface
+func (d *DistinctV3) GetKeyspaceName() string {
+	return d.Source.GetKeyspaceName()
+}
+
+// GetTableName implements the Primitive interface
+func (d *DistinctV3) GetTableName() string {
+	return d.Source.GetTableName()
+}
+
+// GetFields implements the Primitive interface
+func (d *DistinctV3) GetFields(vcursor VCursor, bindVars map[string]*querypb.BindVariable) (*sqltypes.Result, error) {
+	return d.Source.GetFields(vcursor, bindVars)
+}
+
+// NeedsTransaction implements the Primitive interface
+func (d *DistinctV3) NeedsTransaction() bool {
+	return d.Source.NeedsTransaction()
+}
+
+// Inputs implements the Primitive interface
+func (d *DistinctV3) Inputs() []Primitive {
+	return []Primitive{d.Source}
+}
+
+func (d *DistinctV3) description() PrimitiveDescription {
+	return PrimitiveDescription{
+		OperatorType: "Distinct",
+	}
+}
diff --git a/go/vt/vtgate/engine/hash_join.go b/go/vt/vtgate/engine/hash_join.go
index b97f7a995a..3e06a53625 100644
--- a/go/vt/vtgate/engine/hash_join.go
+++ b/go/vt/vtgate/engine/hash_join.go
@@ -112,8 +112,8 @@ func (hj *HashJoin) TryExecute(vcursor VCursor, bindVars map[string]*querypb.Bin
 	return result, nil
 }
 
-func (hj *HashJoin) buildProbeTable(lresult *sqltypes.Result) (map[evalengine.HashCode][]row, error) {
-	probeTable := map[evalengine.HashCode][]row{}
+func (hj *HashJoin) buildProbeTable(lresult *sqltypes.Result) (map[evalengine.HashCode][]sqltypes.Row, error) {
+	probeTable := map[evalengine.HashCode][]sqltypes.Row{}
 	for _, current := range lresult.Rows {
 		joinVal := current[hj.LHSKey]
 		if joinVal.IsNull() {
@@ -131,7 +131,7 @@ func (hj *HashJoin) buildProbeTable(lresult *sqltypes.Result) (map[evalengine.Ha
 // TryStreamExecute implements the Primitive interface
 func (hj *HashJoin) TryStreamExecute(vcursor VCursor, bindVars map[string]*querypb.BindVariable, wantfields bool, callback func(*sqltypes.Result) error) error {
 	// build the probe table from the LHS result
-	probeTable := map[evalengine.HashCode][]row{}
+	probeTable := map[evalengine.HashCode][]sqltypes.Row{}
 	var lfields []*querypb.Field
 	err := vcursor.StreamExecutePrimitive(hj.Left, bindVars, wantfields, func(result *sqltypes.Result) error {
 		if len(lfields) == 0 && len(result.Fields) != 0 {
diff --git a/go/vt/vtgate/evalengine/comparisons.go b/go/vt/vtgate/evalengine/comparisons.go
index aa3510d9bb..5372dfc1b2 100644
--- a/go/vt/vtgate/evalengine/comparisons.go
+++ b/go/vt/vtgate/evalengine/comparisons.go
@@ -146,7 +146,7 @@ func evalResultsAreDateAndNumeric(l, r *EvalResult) bool {
 func compareAsTuples(lVal, rVal *EvalResult) bool {
 	left := lVal.typeof() == sqltypes.Tuple
 	right := rVal.typeof() == sqltypes.Tuple
-	if left && right {
+	if right && left {
 		return true
 	}
 	if left || right {
@@ -408,6 +408,9 @@ func (err UnsupportedCollationError) Error() string {
 	return fmt.Sprintf("cannot compare strings, collation is unknown or unsupported (collation ID: %d)", err.ID)
 }
 
+// UnsupportedCollationHashError is returned when we try to get the hash value and are missing the collation to use
+var UnsupportedCollationHashError = vterrors.Errorf(vtrpcpb.Code_INTERNAL, "text type with an unknown/unsupported collation cannot be hashed")
+
 // NullsafeCompare returns 0 if v1==v2, -1 if v1<v2, and 1 if v1>v2.
 // NULL is the lowest value. If any value is
 // numeric, then a numeric comparison is performed after
diff --git a/go/vt/vtgate/evalengine/eval_result.go b/go/vt/vtgate/evalengine/eval_result.go
index 4361e2aa5a..2e7c5d29ee 100644
--- a/go/vt/vtgate/evalengine/eval_result.go
+++ b/go/vt/vtgate/evalengine/eval_result.go
@@ -465,7 +465,7 @@ func (er *EvalResult) nullSafeHashcode() (HashCode, error) {
 	case er.isTextual():
 		coll := collations.Local().LookupByID(er.collation().Collation)
 		if coll == nil {
-			return 0, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "text type with an unknown/unsupported collation cannot be hashed")
+			return 0, UnsupportedCollationHashError
 		}
 		return coll.Hash(er.bytes(), 0), nil
 	case sqltypes.IsDate(er.typeof()):
diff --git a/go/vt/vtgate/planbuilder/builder.go b/go/vt/vtgate/planbuilder/builder.go
index 7dbf06e9d0..f97ac6b7ef 100644
--- a/go/vt/vtgate/planbuilder/builder.go
+++ b/go/vt/vtgate/planbuilder/builder.go
@@ -393,19 +393,16 @@ func buildFlushTables(stmt *sqlparser.Flush, vschema plancontext.VSchema) (engin
 		return keys[i].ks.Name < keys[j].ks.Name
 	})
 
-	finalPlan := &engine.Concatenate{
-		Sources: nil,
-	}
+	var sources []engine.Primitive
 	for _, sendDest := range keys {
 		plan := &engine.Send{
 			Keyspace:          sendDest.ks,
 			TargetDestination: sendDest.dest,
 			Query:             sqlparser.String(newFlushStmt(stmt, tablesMap[sendDest])),
 		}
-		finalPlan.Sources = append(finalPlan.Sources, plan)
+		sources = append(sources, plan)
 	}
-
-	return finalPlan, nil
+	return engine.NewConcatenate(sources, nil), nil
 }
 
 func newFlushStmt(stmt *sqlparser.Flush, tables sqlparser.TableNames) *sqlparser.Flush {
diff --git a/go/vt/vtgate/planbuilder/concatenate.go b/go/vt/vtgate/planbuilder/concatenate.go
index 8c6d43a8d2..92e31d3de7 100644
--- a/go/vt/vtgate/planbuilder/concatenate.go
+++ b/go/vt/vtgate/planbuilder/concatenate.go
@@ -70,9 +70,7 @@ func (c *concatenate) Primitive() engine.Primitive {
 	lhs := c.lhs.Primitive()
 	rhs := c.rhs.Primitive()
 
-	return &engine.Concatenate{
-		Sources: []engine.Primitive{lhs, rhs},
-	}
+	return engine.NewConcatenate([]engine.Primitive{lhs, rhs}, nil)
 }
 
 // Rewrite implements the logicalPlan interface
diff --git a/go/vt/vtgate/planbuilder/concatenateGen4.go b/go/vt/vtgate/planbuilder/concatenateGen4.go
index 29617b2fb8..8c14b1ba0f 100644
--- a/go/vt/vtgate/planbuilder/concatenateGen4.go
+++ b/go/vt/vtgate/planbuilder/concatenateGen4.go
@@ -26,6 +26,10 @@ import (
 
 type concatenateGen4 struct {
 	sources []logicalPlan
+
+	// These column offsets do not need to be typed checked - they usually contain weight_string()
+	// columns that are not going to be returned to the user
+	noNeedToTypeCheck []int
 }
 
 var _ logicalPlan = (*concatenateGen4)(nil)
@@ -82,9 +86,8 @@ func (c *concatenateGen4) Primitive() engine.Primitive {
 	for _, source := range c.sources {
 		sources = append(sources, source.Primitive())
 	}
-	return &engine.Concatenate{
-		Sources: sources,
-	}
+
+	return engine.NewConcatenate(sources, c.noNeedToTypeCheck)
 }
 
 // Rewrite implements the logicalPlan interface
diff --git a/go/vt/vtgate/planbuilder/distinct.go b/go/vt/vtgate/planbuilder/distinct.go
index 40cb3d6030..695cc43905 100644
--- a/go/vt/vtgate/planbuilder/distinct.go
+++ b/go/vt/vtgate/planbuilder/distinct.go
@@ -17,7 +17,6 @@ limitations under the License.
 package planbuilder
 
 import (
-	"vitess.io/vitess/go/mysql/collations"
 	vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
 	"vitess.io/vitess/go/vt/vterrors"
 	"vitess.io/vitess/go/vt/vtgate/engine"
@@ -28,20 +27,40 @@ var _ logicalPlan = (*distinct)(nil)
 // distinct is the logicalPlan for engine.Distinct.
 type distinct struct {
 	logicalPlanCommon
-	ColCollations []collations.ID
+	checkCols      []engine.CheckCol
+	needToTruncate bool
 }
 
-func newDistinct(source logicalPlan, colCollations []collations.ID) logicalPlan {
+func newDistinct(source logicalPlan, checkCols []engine.CheckCol, needToTruncate bool) logicalPlan {
 	return &distinct{
 		logicalPlanCommon: newBuilderCommon(source),
-		ColCollations:     colCollations,
+		checkCols:         checkCols,
+		needToTruncate:    needToTruncate,
 	}
 }
 
+func newDistinctV3(source logicalPlan) logicalPlan {
+	return &distinct{logicalPlanCommon: newBuilderCommon(source)}
+}
+
 func (d *distinct) Primitive() engine.Primitive {
+	if d.checkCols == nil {
+		// If we are missing the checkCols information, we are on the V3 planner and should produce a V3 Distinct
+		return &engine.DistinctV3{Source: d.input.Primitive()}
+	}
+	truncate := false
+	if d.needToTruncate {
+		for _, col := range d.checkCols {
+			if col.WsCol != nil {
+				truncate = true
+				break
+			}
+		}
+	}
 	return &engine.Distinct{
-		Source:        d.input.Primitive(),
-		ColCollations: d.ColCollations,
+		Source:    d.input.Primitive(),
+		CheckCols: d.checkCols,
+		Truncate:  truncate,
 	}
 }
 
diff --git a/go/vt/vtgate/planbuilder/gen4_planner.go b/go/vt/vtgate/planbuilder/gen4_planner.go
index 46946e08c8..54808a8a1c 100644
--- a/go/vt/vtgate/planbuilder/gen4_planner.go
+++ b/go/vt/vtgate/planbuilder/gen4_planner.go
@@ -173,12 +173,12 @@ func newBuildSelectPlan(
 		return nil, err
 	}
 
-	plan, err := transformToLogicalPlan(ctx, physOp)
+	plan, err := transformToLogicalPlan(ctx, physOp, true)
 	if err != nil {
 		return nil, err
 	}
 
-	plan, err = planHorizon(ctx, plan, selStmt)
+	plan, err = planHorizon(ctx, plan, selStmt, true)
 	if err != nil {
 		return nil, err
 	}
@@ -225,7 +225,7 @@ func planLimit(limit *sqlparser.Limit, plan logicalPlan) (logicalPlan, error) {
 	return lPlan, nil
 }
 
-func planHorizon(ctx *plancontext.PlanningContext, plan logicalPlan, in sqlparser.SelectStatement) (logicalPlan, error) {
+func planHorizon(ctx *plancontext.PlanningContext, plan logicalPlan, in sqlparser.SelectStatement, truncateColumns bool) (logicalPlan, error) {
 	switch node := in.(type) {
 	case *sqlparser.Select:
 		hp := horizonPlanning{
@@ -234,7 +234,7 @@ func planHorizon(ctx *plancontext.PlanningContext, plan logicalPlan, in sqlparse
 
 		replaceSubQuery(ctx, node)
 		var err error
-		plan, err = hp.planHorizon(ctx, plan)
+		plan, err = hp.planHorizon(ctx, plan, truncateColumns)
 		if err != nil {
 			return nil, err
 		}
diff --git a/go/vt/vtgate/planbuilder/grouping.go b/go/vt/vtgate/planbuilder/grouping.go
index 2bf58dcfa4..d96534f5ce 100644
--- a/go/vt/vtgate/planbuilder/grouping.go
+++ b/go/vt/vtgate/planbuilder/grouping.go
@@ -108,7 +108,7 @@ func planDistinct(input logicalPlan) (logicalPlan, error) {
 			// So, the distinct 'operator' cannot be pushed down into the
 			// route.
 			if rc.column.Origin() == node {
-				return newDistinct(node, nil), nil
+				return newDistinctV3(node), nil
 			}
 			node.groupByKeys = append(node.groupByKeys, &engine.GroupByParams{KeyCol: i, WeightStringCol: -1, FromGroupBy: false})
 		}
diff --git a/go/vt/vtgate/planbuilder/horizon_planning.go b/go/vt/vtgate/planbuilder/horizon_planning.go
index bb0267aadd..401de2ada5 100644
--- a/go/vt/vtgate/planbuilder/horizon_planning.go
+++ b/go/vt/vtgate/planbuilder/horizon_planning.go
@@ -37,7 +37,7 @@ type horizonPlanning struct {
 	qp  *abstract.QueryProjection
 }
 
-func (hp *horizonPlanning) planHorizon(ctx *plancontext.PlanningContext, plan logicalPlan) (logicalPlan, error) {
+func (hp *horizonPlanning) planHorizon(ctx *plancontext.PlanningContext, plan logicalPlan, truncateColumns bool) (logicalPlan, error) {
 	rb, isRoute := plan.(*routeGen4)
 	if !isRoute && ctx.SemTable.NotSingleRouteErr != nil {
 		// If we got here, we don't have a single shard plan
@@ -63,24 +63,23 @@ func (hp *horizonPlanning) planHorizon(ctx *plancontext.PlanningContext, plan lo
 
 	// If we still have a HAVING clause, it's because it could not be pushed to the WHERE,
 	// so it probably has aggregations
-	if hp.qp.NeedsAggregation() || hp.sel.Having != nil {
+	switch {
+	case hp.qp.NeedsAggregation() || hp.sel.Having != nil:
 		plan, err = hp.planAggregations(ctx, plan)
 		if err != nil {
 			return nil, err
 		}
 		// if we already did sorting, we don't need to do it again
 		needsOrdering = needsOrdering && !hp.qp.CanPushDownSorting
-	} else {
-		if canShortcut {
-			err = planSingleShardRoutePlan(hp.sel, rb)
-			if err != nil {
-				return nil, err
-			}
-		} else {
-			err = pushProjections(ctx, plan, hp.qp.SelectExprs)
-			if err != nil {
-				return nil, err
-			}
+	case canShortcut:
+		err = planSingleShardRoutePlan(hp.sel, rb)
+		if err != nil {
+			return nil, err
+		}
+	default:
+		err = pushProjections(ctx, plan, hp.qp.SelectExprs)
+		if err != nil {
+			return nil, err
 		}
 	}
 
@@ -97,6 +96,10 @@ func (hp *horizonPlanning) planHorizon(ctx *plancontext.PlanningContext, plan lo
 		return nil, err
 	}
 
+	if !truncateColumns {
+		return plan, nil
+	}
+
 	plan, err = hp.truncateColumnsIfNeeded(ctx, plan)
 	if err != nil {
 		return nil, err
@@ -1219,7 +1222,7 @@ func (hp *horizonPlanning) addDistinct(ctx *plancontext.PlanningContext, plan lo
 			inner = sqlparser.NewColName(aliasExpr.As.String())
 			ctx.SemTable.CopyDependencies(aliasExpr.Expr, inner)
 		}
-		grpParam := &engine.GroupByParams{KeyCol: index, WeightStringCol: -1, CollationID: ctx.SemTable.CollationForExpr(inner)}
+		grpParam := &engine.GroupByParams{KeyCol: index, WeightStringCol: -1, CollationID: ctx.SemTable.CollationForExpr(inner), Expr: inner}
 		_, wOffset, err := wrapAndPushExpr(ctx, aliasExpr.Expr, aliasExpr.Expr, plan)
 		if err != nil {
 			return nil, err
diff --git a/go/vt/vtgate/planbuilder/operator_transformers.go b/go/vt/vtgate/planbuilder/operator_transformers.go
index 724d65c8af..ef7553bcb9 100644
--- a/go/vt/vtgate/planbuilder/operator_transformers.go
+++ b/go/vt/vtgate/planbuilder/operator_transformers.go
@@ -21,6 +21,8 @@ import (
 	"strconv"
 	"strings"
 
+	"vitess.io/vitess/go/sqltypes"
+
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/plancontext"
 
 	"vitess.io/vitess/go/vt/vtgate/planbuilder/physical"
@@ -39,14 +41,14 @@ import (
 	"vitess.io/vitess/go/vt/vterrors"
 )
 
-func transformToLogicalPlan(ctx *plancontext.PlanningContext, op abstract.PhysicalOperator) (logicalPlan, error) {
+func transformToLogicalPlan(ctx *plancontext.PlanningContext, op abstract.PhysicalOperator, isRoot bool) (logicalPlan, error) {
 	switch op := op.(type) {
 	case *physical.Route:
 		return transformRoutePlan(ctx, op)
 	case *physical.ApplyJoin:
 		return transformApplyJoinPlan(ctx, op)
 	case *physical.Union:
-		return transformUnionPlan(ctx, op)
+		return transformUnionPlan(ctx, op, isRoot)
 	case *physical.Vindex:
 		return transformVindexPlan(ctx, op)
 	case *physical.SubQueryOp:
@@ -56,7 +58,7 @@ func transformToLogicalPlan(ctx *plancontext.PlanningContext, op abstract.Physic
 	case *physical.Derived:
 		return transformDerivedPlan(ctx, op)
 	case *physical.Filter:
-		plan, err := transformToLogicalPlan(ctx, op.Source)
+		plan, err := transformToLogicalPlan(ctx, op.Source, false)
 		if err != nil {
 			return nil, err
 		}
@@ -91,11 +93,11 @@ func transformApplyJoinPlan(ctx *plancontext.PlanningContext, n *physical.ApplyJ
 	//	return nil, err
 	// }
 
-	lhs, err := transformToLogicalPlan(ctx, n.LHS)
+	lhs, err := transformToLogicalPlan(ctx, n.LHS, false)
 	if err != nil {
 		return nil, err
 	}
-	rhs, err := transformToLogicalPlan(ctx, n.RHS)
+	rhs, err := transformToLogicalPlan(ctx, n.RHS, false)
 	if err != nil {
 		return nil, err
 	}
@@ -237,7 +239,7 @@ func getAllTableNames(op *physical.Route) ([]string, error) {
 	return tableNames, nil
 }
 
-func transformUnionPlan(ctx *plancontext.PlanningContext, op *physical.Union) (logicalPlan, error) {
+func transformUnionPlan(ctx *plancontext.PlanningContext, op *physical.Union, isRoot bool) (logicalPlan, error) {
 	var sources []logicalPlan
 	var err error
 	if op.Distinct {
@@ -272,12 +274,98 @@ func transformUnionPlan(ctx *plancontext.PlanningContext, op *physical.Union) (l
 		result = &concatenateGen4{sources: sources}
 	}
 	if op.Distinct {
-		return newDistinct(result, getCollationsFor(ctx, op)), nil
+		colls := getCollationsFor(ctx, op)
+		checkCols, err := getCheckColsForUnion(ctx, result, colls)
+		if err != nil {
+			return nil, err
+		}
+		return newDistinct(result, checkCols, isRoot), nil
 	}
 	return result, nil
 
 }
 
+func getWeightStringForSelectExpr(selectExpr sqlparser.SelectExpr) (*sqlparser.AliasedExpr, error) {
+	expr, isAliased := selectExpr.(*sqlparser.AliasedExpr)
+	if !isAliased {
+		return nil, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "cannot convert select expression to an aliased expression")
+	}
+	return &sqlparser.AliasedExpr{Expr: weightStringFor(expr.Expr)}, nil
+}
+
+func getCheckColsForUnion(ctx *plancontext.PlanningContext, result logicalPlan, colls []collations.ID) ([]engine.CheckCol, error) {
+	checkCols := make([]engine.CheckCol, 0, len(colls))
+	for i, coll := range colls {
+		checkCol := engine.CheckCol{Col: i, Collation: coll}
+		if coll != collations.Unknown {
+			checkCols = append(checkCols, checkCol)
+			continue
+		}
+		// We might need a weight string - let's push one
+		// `might` because we just don't know what type we are dealing with.
+		// If we encounter a numerical value, we don't need any weight_string values
+		newOffset, err := pushWeightStringForDistinct(ctx, result, i)
+		if err != nil {
+			return nil, err
+		}
+		checkCol.WsCol = &newOffset
+		checkCols = append(checkCols, checkCol)
+	}
+	return checkCols, nil
+}
+
+// pushWeightStringForDistinct adds a weight_string projection
+func pushWeightStringForDistinct(ctx *plancontext.PlanningContext, plan logicalPlan, offset int) (newOffset int, err error) {
+	switch node := plan.(type) {
+	case *routeGen4:
+		allSelects := sqlparser.GetAllSelects(node.Select)
+		for _, sel := range allSelects {
+			expr, err := getWeightStringForSelectExpr(sel.SelectExprs[offset])
+			if err != nil {
+				return 0, err
+			}
+			if i := checkIfAlreadyExists(expr, sel, ctx.SemTable); i != -1 {
+				return i, nil
+			}
+			sel.SelectExprs = append(sel.SelectExprs, expr)
+			newOffset = len(sel.SelectExprs) - 1
+		}
+		// we leave the responsibility of truncating to distinct
+		node.eroute.TruncateColumnCount = 0
+	case *concatenateGen4:
+		for _, source := range node.sources {
+			newOffset, err = pushWeightStringForDistinct(ctx, source, offset)
+			if err != nil {
+				return 0, err
+			}
+		}
+		node.noNeedToTypeCheck = append(node.noNeedToTypeCheck, newOffset)
+	case *joinGen4:
+		lhsSolves := node.Left.ContainsTables()
+		rhsSolves := node.Right.ContainsTables()
+		expr := node.OutputColumns()[offset]
+		aliasedExpr, isAliased := expr.(*sqlparser.AliasedExpr)
+		if !isAliased {
+			return 0, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "cannot convert select expression to an aliased expression")
+		}
+		deps := ctx.SemTable.RecursiveDeps(aliasedExpr.Expr)
+		switch {
+		case deps.IsSolvedBy(lhsSolves):
+			offset, err = pushWeightStringForDistinct(ctx, node.Left, offset)
+			node.Cols = append(node.Cols, -(offset + 1))
+		case deps.IsSolvedBy(rhsSolves):
+			offset, err = pushWeightStringForDistinct(ctx, node.Right, offset)
+			node.Cols = append(node.Cols, offset+1)
+		default:
+			return 0, vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "cannot push distinct weight string to both sides of the join")
+		}
+		newOffset = len(node.Cols) - 1
+	default:
+		return 0, vterrors.Errorf(vtrpcpb.Code_UNIMPLEMENTED, "bug: not supported pushWeightStringForDistinct on %T", plan)
+	}
+	return
+}
+
 func transformAndMerge(ctx *plancontext.PlanningContext, op *physical.Union) (sources []logicalPlan, err error) {
 	for i, source := range op.Sources {
 		// first we go over all the operator inputs and turn them into logical plans,
@@ -351,12 +439,12 @@ func transformAndMergeInOrder(ctx *plancontext.PlanningContext, op *physical.Uni
 }
 
 func createLogicalPlan(ctx *plancontext.PlanningContext, source abstract.PhysicalOperator, selStmt *sqlparser.Select) (logicalPlan, error) {
-	plan, err := transformToLogicalPlan(ctx, source)
+	plan, err := transformToLogicalPlan(ctx, source, false)
 	if err != nil {
 		return nil, err
 	}
 	if selStmt != nil {
-		plan, err = planHorizon(ctx, plan, selStmt)
+		plan, err = planHorizon(ctx, plan, selStmt, true)
 		if err != nil {
 			return nil, err
 		}
@@ -368,6 +456,7 @@ func createLogicalPlan(ctx *plancontext.PlanningContext, source abstract.Physica
 }
 
 func getCollationsFor(ctx *plancontext.PlanningContext, n *physical.Union) []collations.ID {
+	// TODO: coerce selects' select expressions' collations
 	var colls []collations.ID
 	for _, expr := range n.SelectStmts[0].SelectExprs {
 		aliasedE, ok := expr.(*sqlparser.AliasedExpr)
@@ -375,6 +464,11 @@ func getCollationsFor(ctx *plancontext.PlanningContext, n *physical.Union) []col
 			return nil
 		}
 		typ := ctx.SemTable.CollationForExpr(aliasedE.Expr)
+		if typ == collations.Unknown {
+			if t, hasT := ctx.SemTable.ExprTypes[aliasedE.Expr]; hasT && sqltypes.IsNumber(t.Type) {
+				typ = collations.CollationBinaryID
+			}
+		}
 		colls = append(colls, typ)
 	}
 	return colls
@@ -387,12 +481,12 @@ func transformDerivedPlan(ctx *plancontext.PlanningContext, op *physical.Derived
 	// expression containing our derived table's inner select and the derived
 	// table's alias.
 
-	plan, err := transformToLogicalPlan(ctx, op.Source)
+	plan, err := transformToLogicalPlan(ctx, op.Source, false)
 	if err != nil {
 		return nil, err
 	}
 
-	plan, err = planHorizon(ctx, plan, op.Query)
+	plan, err = planHorizon(ctx, plan, op.Query, false)
 	if err != nil {
 		return nil, err
 	}
diff --git a/go/vt/vtgate/planbuilder/ordered_aggregate.go b/go/vt/vtgate/planbuilder/ordered_aggregate.go
index 7506ccde07..4c091055dd 100644
--- a/go/vt/vtgate/planbuilder/ordered_aggregate.go
+++ b/go/vt/vtgate/planbuilder/ordered_aggregate.go
@@ -104,7 +104,7 @@ func (pb *primitiveBuilder) checkAggregates(sel *sqlparser.Select) error {
 		if hasAggregates {
 			return vterrors.New(vtrpcpb.Code_UNIMPLEMENTED, "unsupported: cross-shard query with aggregates")
 		}
-		pb.plan = newDistinct(pb.plan, nil)
+		pb.plan = newDistinctV3(pb.plan)
 		return nil
 	}
 
diff --git a/go/vt/vtgate/planbuilder/subquery_op.go b/go/vt/vtgate/planbuilder/subquery_op.go
index a1abc30942..6191e5228c 100644
--- a/go/vt/vtgate/planbuilder/subquery_op.go
+++ b/go/vt/vtgate/planbuilder/subquery_op.go
@@ -25,18 +25,18 @@ import (
 )
 
 func transformSubQueryPlan(ctx *plancontext.PlanningContext, op *physical.SubQueryOp) (logicalPlan, error) {
-	innerPlan, err := transformToLogicalPlan(ctx, op.Inner)
+	innerPlan, err := transformToLogicalPlan(ctx, op.Inner, false)
 	if err != nil {
 		return nil, err
 	}
-	innerPlan, err = planHorizon(ctx, innerPlan, op.Extracted.Subquery.Select)
+	innerPlan, err = planHorizon(ctx, innerPlan, op.Extracted.Subquery.Select, true)
 	if err != nil {
 		return nil, err
 	}
 
 	argName := op.Extracted.GetArgName()
 	hasValuesArg := op.Extracted.GetHasValuesArg()
-	outerPlan, err := transformToLogicalPlan(ctx, op.Outer)
+	outerPlan, err := transformToLogicalPlan(ctx, op.Outer, false)
 
 	merged := mergeSubQueryOpPlan(ctx, innerPlan, outerPlan, op)
 	if merged != nil {
@@ -51,11 +51,11 @@ func transformSubQueryPlan(ctx *plancontext.PlanningContext, op *physical.SubQue
 }
 
 func transformCorrelatedSubQueryPlan(ctx *plancontext.PlanningContext, op *physical.CorrelatedSubQueryOp) (logicalPlan, error) {
-	outer, err := transformToLogicalPlan(ctx, op.Outer)
+	outer, err := transformToLogicalPlan(ctx, op.Outer, false)
 	if err != nil {
 		return nil, err
 	}
-	inner, err := transformToLogicalPlan(ctx, op.Inner)
+	inner, err := transformToLogicalPlan(ctx, op.Inner, false)
 	if err != nil {
 		return nil, err
 	}
diff --git a/go/vt/vtgate/planbuilder/union.go b/go/vt/vtgate/planbuilder/union.go
index c0305cdbc4..32dbf5ad21 100644
--- a/go/vt/vtgate/planbuilder/union.go
+++ b/go/vt/vtgate/planbuilder/union.go
@@ -78,7 +78,7 @@ func (pb *primitiveBuilder) processUnion(union *sqlparser.Union, reservedVars *s
 		}
 
 		if union.Distinct {
-			pb.plan = newDistinct(pb.plan, nil)
+			pb.plan = newDistinctV3(pb.plan)
 		}
 	}
 	pb.st.Outer = outer


Test patch:
diff --git a/go/vt/vtgate/engine/concatenate_test.go b/go/vt/vtgate/engine/concatenate_test.go
index 0acb217f27..bf1b54e061 100644
--- a/go/vt/vtgate/engine/concatenate_test.go
+++ b/go/vt/vtgate/engine/concatenate_test.go
@@ -39,8 +39,14 @@ func TestConcatenate_NoErrors(t *testing.T) {
 		inputs         []*sqltypes.Result
 		expectedResult *sqltypes.Result
 		expectedError  string
+		ignoreTypes    []int
 	}
 
+	r1 := r("id|col1|col2", "int64|varbinary|varbinary", "1|a1|b1", "2|a2|b2")
+	r2 := r("id|col1|col2", "int32|varbinary|varbinary", "1|a1|b1", "2|a2|b2")
+	combinedResult := r1
+	combinedResult.Rows = append(combinedResult.Rows, r2.Rows...)
+
 	testCases := []*testCase{{
 		testName: "empty results",
 		inputs: []*sqltypes.Result{
@@ -66,6 +72,14 @@ func TestConcatenate_NoErrors(t *testing.T) {
 			r("id|col3|col4", "int64|varchar|varbinary", "1|a1|b1", "2|a2|b2"),
 		},
 		expectedError: "merging field of different types is not supported",
+	}, {
+		testName: "ignored field types - ignored",
+		inputs: []*sqltypes.Result{
+			r("id|col1|col2", "int64|varbinary|varbinary", "1|a1|b1", "2|a2|b2"),
+			r("id|col1|col2", "int32|varbinary|varbinary", "1|a1|b1", "2|a2|b2"),
+		},
+		expectedResult: combinedResult,
+		ignoreTypes:    []int{0},
 	}, {
 		testName: "input source has different column count",
 		inputs: []*sqltypes.Result{
@@ -89,9 +103,8 @@ func TestConcatenate_NoErrors(t *testing.T) {
 			// input is added twice, since the first one is used by execute and the next by stream execute
 			sources = append(sources, &fakePrimitive{results: []*sqltypes.Result{input, input}})
 		}
-		concatenate := &Concatenate{
-			Sources: sources,
-		}
+
+		concatenate := NewConcatenate(sources, tc.ignoreTypes)
 
 		t.Run(tc.testName+"-Execute", func(t *testing.T) {
 			qr, err := concatenate.TryExecute(&noopVCursor{ctx: context.Background()}, nil, true)
@@ -121,13 +134,13 @@ func TestConcatenate_WithErrors(t *testing.T) {
 	strFailed := "failed"
 
 	fake := r("id|col1|col2", "int64|varchar|varbinary", "1|a1|b1", "2|a2|b2")
-	concatenate := &Concatenate{
-		Sources: []Primitive{
+	concatenate := NewConcatenate(
+		[]Primitive{
 			&fakePrimitive{results: []*sqltypes.Result{fake, fake}},
 			&fakePrimitive{results: []*sqltypes.Result{nil, nil}, sendErr: errors.New(strFailed)},
 			&fakePrimitive{results: []*sqltypes.Result{fake, fake}},
-		},
-	}
+		}, nil,
+	)
 	ctx := context.Background()
 	_, err := concatenate.TryExecute(&noopVCursor{ctx: ctx}, nil, true)
 	require.EqualError(t, err, strFailed)
@@ -135,13 +148,13 @@ func TestConcatenate_WithErrors(t *testing.T) {
 	_, err = wrapStreamExecute(concatenate, &noopVCursor{ctx: ctx}, nil, true)
 	require.EqualError(t, err, strFailed)
 
-	concatenate = &Concatenate{
-		Sources: []Primitive{
+	concatenate = NewConcatenate(
+		[]Primitive{
 			&fakePrimitive{results: []*sqltypes.Result{fake, fake}},
 			&fakePrimitive{results: []*sqltypes.Result{nil, nil}, sendErr: errors.New(strFailed)},
 			&fakePrimitive{results: []*sqltypes.Result{fake, fake}},
-		},
-	}
+		}, nil)
+
 	_, err = concatenate.TryExecute(&noopVCursor{ctx: ctx}, nil, true)
 	require.EqualError(t, err, strFailed)
 	_, err = wrapStreamExecute(concatenate, &noopVCursor{ctx: ctx}, nil, true)
diff --git a/go/vt/vtgate/engine/distinct_test.go b/go/vt/vtgate/engine/distinct_test.go
index 86652b7bf6..d5be097dd1 100644
--- a/go/vt/vtgate/engine/distinct_test.go
+++ b/go/vt/vtgate/engine/distinct_test.go
@@ -76,10 +76,27 @@ func TestDistinct(t *testing.T) {
 	}}
 
 	for _, tc := range testCases {
+		var checkCols []CheckCol
+		if len(tc.inputs.Rows) > 0 {
+			for i := range tc.inputs.Rows[0] {
+				collID := collations.Unknown
+				if tc.collations != nil {
+					collID = tc.collations[i]
+				}
+				if sqltypes.IsNumber(tc.inputs.Fields[i].Type) {
+					collID = collations.CollationBinaryID
+				}
+				checkCols = append(checkCols, CheckCol{
+					Col:       i,
+					Collation: collID,
+				})
+			}
+		}
 		t.Run(tc.testName+"-Execute", func(t *testing.T) {
 			distinct := &Distinct{
-				Source:        &fakePrimitive{results: []*sqltypes.Result{tc.inputs}},
-				ColCollations: tc.collations,
+				Source:    &fakePrimitive{results: []*sqltypes.Result{tc.inputs}},
+				CheckCols: checkCols,
+				Truncate:  false,
 			}
 
 			qr, err := distinct.TryExecute(&noopVCursor{ctx: context.Background()}, nil, true)
@@ -94,8 +111,8 @@ func TestDistinct(t *testing.T) {
 		})
 		t.Run(tc.testName+"-StreamExecute", func(t *testing.T) {
 			distinct := &Distinct{
-				Source:        &fakePrimitive{results: []*sqltypes.Result{tc.inputs}},
-				ColCollations: tc.collations,
+				Source:    &fakePrimitive{results: []*sqltypes.Result{tc.inputs}},
+				CheckCols: checkCols,
 			}
 
 			result, err := wrapStreamExecute(distinct, &noopVCursor{ctx: context.Background()}, nil, true)
@@ -111,3 +128,37 @@ func TestDistinct(t *testing.T) {
 		})
 	}
 }
+
+func TestWeightStringFallBack(t *testing.T) {
+	offsetOne := 1
+	checkCols := []CheckCol{{
+		Col:       0,
+		WsCol:     &offsetOne,
+		Collation: collations.Unknown,
+	}}
+	input := r("myid|weightstring(myid)",
+		"varchar|varbinary",
+		"monkey|monkey",
+		"horse|horse",
+		"horse|horse")
+
+	distinct := &Distinct{
+		Source:    &fakePrimitive{results: []*sqltypes.Result{input}},
+		CheckCols: checkCols,
+		Truncate:  true,
+	}
+
+	qr, err := distinct.TryExecute(&noopVCursor{ctx: context.Background()}, nil, true)
+	require.NoError(t, err)
+
+	got := fmt.Sprintf("%v", qr.Rows)
+	expected := fmt.Sprintf("%v", r("myid", "varchar", "monkey", "horse").Rows)
+	utils.MustMatch(t, expected, got)
+
+	// the primitive must not change just because one run needed weight strings
+	utils.MustMatch(t, []CheckCol{{
+		Col:       0,
+		WsCol:     &offsetOne,
+		Collation: collations.Unknown,
+	}}, distinct.CheckCols, "checkCols should not be updated")
+}
diff --git a/go/vt/vtgate/executor_select_test.go b/go/vt/vtgate/executor_select_test.go
index b379650bd6..e09212c4c4 100644
--- a/go/vt/vtgate/executor_select_test.go
+++ b/go/vt/vtgate/executor_select_test.go
@@ -881,12 +881,33 @@ func testRowCount(t *testing.T, executor *Executor, wantRowCount int64) {
 }
 
 func TestSelectLastInsertIdInUnion(t *testing.T) {
-	executor, _, _, _ := createExecutorEnv()
+	executor, sbc1, _, _ := createExecutorEnv()
 	executor.normalize = true
-	sql := "select last_insert_id() as id union select id from user"
-	_, err := executorExec(executor, sql, map[string]*querypb.BindVariable{})
-	require.Error(t, err)
-	assert.Contains(t, err.Error(), "text type with an unknown/unsupported collation cannot be hashed")
+	primarySession.LastInsertId = 52
+
+	result1 := []*sqltypes.Result{{
+		Fields: []*querypb.Field{
+			{Name: "id", Type: sqltypes.Int32},
+		},
+		InsertID: 0,
+		Rows: [][]sqltypes.Value{{
+			sqltypes.NewInt32(52),
+		}},
+	}}
+	sbc1.SetResults(result1)
+
+	sql := "select last_insert_id() as id union select last_insert_id() as id"
+	got, err := executorExec(executor, sql, map[string]*querypb.BindVariable{})
+	require.NoError(t, err)
+	wantResult := &sqltypes.Result{
+		Fields: []*querypb.Field{
+			{Name: "id", Type: sqltypes.Int32},
+		},
+		Rows: [][]sqltypes.Value{{
+			sqltypes.NewInt32(52),
+		}},
+	}
+	utils.MustMatch(t, wantResult, got, "mismatch")
 }
 
 func TestSelectLastInsertIdInWhere(t *testing.T) {
diff --git a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
index 853b18d915..0e342f6d61 100644
--- a/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/aggr_cases.txt
@@ -2146,7 +2146,6 @@ Gen4 plan same as above
         "FieldQuery": "select count(*) from (select `user`.col, user_extra.extra, weight_string(user_extra.extra) from `user`, user_extra where 1 != 1) as a where 1 != 1",
         "OrderBy": "(1|2) ASC",
         "Query": "select count(*) from (select `user`.col, user_extra.extra, weight_string(user_extra.extra) from `user`, user_extra where `user`.id = user_extra.user_id order by user_extra.extra asc) as a",
-        "ResultColumns": 2,
         "Table": "`user`, user_extra"
       }
     ]
@@ -2175,7 +2174,6 @@ Gen4 plan same as above
         "FieldQuery": "select col, count(*) from (select `user`.col, user_extra.extra, weight_string(user_extra.extra) from `user`, user_extra where 1 != 1) as a where 1 != 1 group by col",
         "OrderBy": "(1|2) ASC, 0 ASC",
         "Query": "select col, count(*) from (select `user`.col, user_extra.extra, weight_string(user_extra.extra) from `user`, user_extra where `user`.id = user_extra.user_id order by user_extra.extra asc) as a group by col order by col asc",
-        "ResultColumns": 2,
         "Table": "`user`, user_extra"
       }
     ]
diff --git a/go/vt/vtgate/planbuilder/testdata/select_cases.txt b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
index c1245a8c17..652ad343b9 100644
--- a/go/vt/vtgate/planbuilder/testdata/select_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/select_cases.txt
@@ -3021,3 +3021,92 @@ Gen4 plan same as above
   }
 }
 Gen4 plan same as above
+
+"select (select id from user order by id limit 1) from user_extra"
+{
+  "QueryType": "SELECT",
+  "Original": "select (select id from user order by id limit 1) from user_extra",
+  "Instructions": {
+    "OperatorType": "Subquery",
+    "Variant": "PulloutValue",
+    "PulloutVars": [
+      "__sq_has_values1",
+      "__sq1"
+    ],
+    "Inputs": [
+      {
+        "OperatorType": "Limit",
+        "Count": "INT64(1)",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "Scatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1",
+            "OrderBy": "(0|1) ASC",
+            "Query": "select id, weight_string(id) from `user` order by id asc limit :__upper_limit",
+            "ResultColumns": 1,
+            "Table": "`user`"
+          }
+        ]
+      },
+      {
+        "OperatorType": "Route",
+        "Variant": "Scatter",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select :__sq1 from user_extra where 1 != 1",
+        "Query": "select :__sq1 from user_extra",
+        "Table": "user_extra"
+      }
+    ]
+  }
+}
+{
+  "QueryType": "SELECT",
+  "Original": "select (select id from user order by id limit 1) from user_extra",
+  "Instructions": {
+    "OperatorType": "Subquery",
+    "Variant": "PulloutValue",
+    "PulloutVars": [
+      "__sq1"
+    ],
+    "Inputs": [
+      {
+        "OperatorType": "Limit",
+        "Count": "INT64(1)",
+        "Inputs": [
+          {
+            "OperatorType": "Route",
+            "Variant": "Scatter",
+            "Keyspace": {
+              "Name": "user",
+              "Sharded": true
+            },
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1",
+            "OrderBy": "(0|1) ASC",
+            "Query": "select id, weight_string(id) from `user` order by id asc limit :__upper_limit",
+            "ResultColumns": 1,
+            "Table": "`user`"
+          }
+        ]
+      },
+      {
+        "OperatorType": "Route",
+        "Variant": "Scatter",
+        "Keyspace": {
+          "Name": "user",
+          "Sharded": true
+        },
+        "FieldQuery": "select :__sq1 from user_extra where 1 != 1",
+        "Query": "select :__sq1 from user_extra",
+        "Table": "user_extra"
+      }
+    ]
+  }
+}
diff --git a/go/vt/vtgate/planbuilder/testdata/union_cases.txt b/go/vt/vtgate/planbuilder/testdata/union_cases.txt
index df3b63e565..60c1531f1e 100644
--- a/go/vt/vtgate/planbuilder/testdata/union_cases.txt
+++ b/go/vt/vtgate/planbuilder/testdata/union_cases.txt
@@ -60,6 +60,10 @@ Gen4 plan same as above
   "Original": "select id from user union select id from music",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:1)"
+    ],
+    "ResultColumns": 1,
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -68,8 +72,8 @@ Gen4 plan same as above
           "Name": "user",
           "Sharded": true
         },
-        "FieldQuery": "select id from `user` where 1 != 1 union select id from music where 1 != 1",
-        "Query": "select id from `user` union select id from music",
+        "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1 union select id, weight_string(id) from music where 1 != 1",
+        "Query": "select id, weight_string(id) from `user` union select id, weight_string(id) from music",
         "Table": "`user`"
       }
     ]
@@ -531,6 +535,10 @@ Gen4 plan same as above
   "Original": "(select id from user union select id from music) union select 1 from dual",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:1)"
+    ],
+    "ResultColumns": 1,
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -542,8 +550,8 @@ Gen4 plan same as above
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select id from `user` where 1 != 1 union select id from music where 1 != 1",
-            "Query": "select id from `user` union select id from music",
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1 union select id, weight_string(id) from music where 1 != 1",
+            "Query": "select id, weight_string(id) from `user` union select id, weight_string(id) from music",
             "Table": "`user`"
           },
           {
@@ -553,8 +561,8 @@ Gen4 plan same as above
               "Name": "main",
               "Sharded": false
             },
-            "FieldQuery": "select 1 from dual where 1 != 1",
-            "Query": "select distinct 1 from dual",
+            "FieldQuery": "select 1, weight_string(1) from dual where 1 != 1",
+            "Query": "select distinct 1, weight_string(1) from dual",
             "Table": "dual"
           }
         ]
@@ -822,6 +830,9 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "select 1 from music where id = 1 union select 1 from music where id = 2",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "0: binary"
+    ],
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -909,6 +920,10 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "(select id from user order by 1 desc) union (select id from user order by 1 asc)",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:1)"
+    ],
+    "ResultColumns": 1,
     "Inputs": [
       {
         "OperatorType": "Route",
@@ -970,6 +985,9 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "select 1 union select null union select 1.0 union select '1' union select 2 union select 2.0 from user",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "0: binary"
+    ],
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1064,6 +1082,11 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "(select user.id, user.name from user join user_extra where user_extra.extra = 'asdf') union select 'b','c' from user",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:2)",
+      "(1:3)"
+    ],
+    "ResultColumns": 2,
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1071,7 +1094,7 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
           {
             "OperatorType": "Join",
             "Variant": "Join",
-            "JoinColumnIndexes": "L:0,L:1",
+            "JoinColumnIndexes": "L:0,L:1,L:2,L:3",
             "TableName": "`user`_user_extra",
             "Inputs": [
               {
@@ -1081,8 +1104,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
                   "Name": "user",
                   "Sharded": true
                 },
-                "FieldQuery": "select `user`.id, `user`.`name` from `user` where 1 != 1",
-                "Query": "select `user`.id, `user`.`name` from `user`",
+                "FieldQuery": "select `user`.id, `user`.`name`, weight_string(`user`.id), weight_string(`user`.`name`) from `user` where 1 != 1",
+                "Query": "select `user`.id, `user`.`name`, weight_string(`user`.id), weight_string(`user`.`name`) from `user`",
                 "Table": "`user`"
               },
               {
@@ -1105,8 +1128,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select 'b', 'c' from `user` where 1 != 1",
-            "Query": "select distinct 'b', 'c' from `user`",
+            "FieldQuery": "select 'b', 'c', weight_string('b'), weight_string('c') from `user` where 1 != 1",
+            "Query": "select distinct 'b', 'c', weight_string('b'), weight_string('c') from `user`",
             "Table": "`user`"
           }
         ]
@@ -1177,6 +1200,11 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "select 'b','c' from user union (select user.id, user.name from user join user_extra where user_extra.extra = 'asdf')",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:2)",
+      "(1:3)"
+    ],
+    "ResultColumns": 2,
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1188,14 +1216,14 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select 'b', 'c' from `user` where 1 != 1",
-            "Query": "select distinct 'b', 'c' from `user`",
+            "FieldQuery": "select 'b', 'c', weight_string('b'), weight_string('c') from `user` where 1 != 1",
+            "Query": "select distinct 'b', 'c', weight_string('b'), weight_string('c') from `user`",
             "Table": "`user`"
           },
           {
             "OperatorType": "Join",
             "Variant": "Join",
-            "JoinColumnIndexes": "L:0,L:1",
+            "JoinColumnIndexes": "L:0,L:1,L:2,L:3",
             "TableName": "`user`_user_extra",
             "Inputs": [
               {
@@ -1205,8 +1233,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
                   "Name": "user",
                   "Sharded": true
                 },
-                "FieldQuery": "select `user`.id, `user`.`name` from `user` where 1 != 1",
-                "Query": "select `user`.id, `user`.`name` from `user`",
+                "FieldQuery": "select `user`.id, `user`.`name`, weight_string(`user`.id), weight_string(`user`.`name`) from `user` where 1 != 1",
+                "Query": "select `user`.id, `user`.`name`, weight_string(`user`.id), weight_string(`user`.`name`) from `user`",
                 "Table": "`user`"
               },
               {
@@ -1285,6 +1313,9 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "select count(*) as s from user union select count(*) as s from music",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "0: binary"
+    ],
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1409,6 +1440,9 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
     "Inputs": [
       {
         "OperatorType": "Distinct",
+        "Collations": [
+          "(0:1)"
+        ],
         "Inputs": [
           {
             "OperatorType": "Route",
@@ -1417,8 +1451,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select id from `user` where 1 != 1 union all select id + 1 from `user` where 1 != 1 union select user_id from user_extra where 1 != 1",
-            "Query": "select id from `user` union all select id + 1 from `user` union select user_id from user_extra",
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1 union all select id + 1, weight_string(id + 1) from `user` where 1 != 1 union select user_id, weight_string(user_id) from user_extra where 1 != 1",
+            "Query": "select id, weight_string(id) from `user` union all select id + 1, weight_string(id + 1) from `user` union select user_id, weight_string(user_id) from user_extra",
             "Table": "`user`"
           }
         ]
@@ -1435,6 +1469,10 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
   "Original": "(select id from user union select id from music order by id) union select 1 from unsharded",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:1)"
+    ],
+    "ResultColumns": 1,
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1446,8 +1484,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select id from `user` where 1 != 1 union select id from music where 1 != 1",
-            "Query": "select id from `user` union select id from music",
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1 union select id, weight_string(id) from music where 1 != 1",
+            "Query": "select id, weight_string(id) from `user` union select id, weight_string(id) from music",
             "Table": "`user`"
           },
           {
@@ -1457,8 +1495,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
               "Name": "main",
               "Sharded": false
             },
-            "FieldQuery": "select 1 from unsharded where 1 != 1",
-            "Query": "select distinct 1 from unsharded",
+            "FieldQuery": "select 1, weight_string(1) from unsharded where 1 != 1",
+            "Query": "select distinct 1, weight_string(1) from unsharded",
             "Table": "unsharded"
           }
         ]
@@ -1520,6 +1558,10 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
     "Inputs": [
       {
         "OperatorType": "Distinct",
+        "Collations": [
+          "(0:1)"
+        ],
+        "ResultColumns": 1,
         "Inputs": [
           {
             "OperatorType": "Concatenate",
@@ -1531,8 +1573,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
                   "Name": "user",
                   "Sharded": true
                 },
-                "FieldQuery": "select id from `user` where 1 != 1",
-                "Query": "select distinct id from `user` limit :__upper_limit",
+                "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1",
+                "Query": "select distinct id, weight_string(id) from `user` limit :__upper_limit",
                 "Table": "`user`"
               },
               {
@@ -1542,8 +1584,8 @@ Gen4 error: nesting of unions at the right-hand side is not yet supported
                   "Name": "main",
                   "Sharded": false
                 },
-                "FieldQuery": "select 3 from dual where 1 != 1",
-                "Query": "select distinct 3 from dual limit :__upper_limit",
+                "FieldQuery": "select 3, weight_string(3) from dual where 1 != 1",
+                "Query": "select distinct 3, weight_string(3) from dual limit :__upper_limit",
                 "Table": "dual"
               }
             ]
@@ -1658,6 +1700,10 @@ Gen4 plan same as above
   "Original": "select col from unsharded union select id from user union select col2 from unsharded union select col from user_extra",
   "Instructions": {
     "OperatorType": "Distinct",
+    "Collations": [
+      "(0:1)"
+    ],
+    "ResultColumns": 1,
     "Inputs": [
       {
         "OperatorType": "Concatenate",
@@ -1669,8 +1715,8 @@ Gen4 plan same as above
               "Name": "main",
               "Sharded": false
             },
-            "FieldQuery": "select col from unsharded where 1 != 1 union select col2 from unsharded where 1 != 1",
-            "Query": "select col from unsharded union select col2 from unsharded",
+            "FieldQuery": "select col, weight_string(col) from unsharded where 1 != 1 union select col2, weight_string(col2) from unsharded where 1 != 1",
+            "Query": "select col, weight_string(col) from unsharded union select col2, weight_string(col2) from unsharded",
             "Table": "unsharded"
           },
           {
@@ -1680,8 +1726,8 @@ Gen4 plan same as above
               "Name": "user",
               "Sharded": true
             },
-            "FieldQuery": "select id from `user` where 1 != 1 union select col from user_extra where 1 != 1",
-            "Query": "select id from `user` union select col from user_extra",
+            "FieldQuery": "select id, weight_string(id) from `user` where 1 != 1 union select col, weight_string(col) from user_extra where 1 != 1",
+            "Query": "select id, weight_string(id) from `user` union select col, weight_string(col) from user_extra",
             "Table": "`user`"
           }
         ]
@@ -1808,11 +1854,14 @@ Gen4 plan same as above
   "Instructions": {
     "OperatorType": "SimpleProjection",
     "Columns": [
-      1
+      2
     ],
     "Inputs": [
       {
         "OperatorType": "Distinct",
+        "Collations": [
+          "(0:1)"
+        ],
         "Inputs": [
           {
             "OperatorType": "Concatenate",
@@ -1824,8 +1873,8 @@ Gen4 plan same as above
                   "Name": "user",
                   "Sharded": true
                 },
-                "FieldQuery": "select id + 42 as foo, 1 from `user` where 1 != 1",
-                "Query": "select distinct id + 42 as foo, 1 from `user`",
+                "FieldQuery": "select id + 42 as foo, weight_string(id + 42), 1 from `user` where 1 != 1",
+                "Query": "select distinct id + 42 as foo, weight_string(id + 42), 1 from `user`",
                 "Table": "`user`"
               },
               {
@@ -1835,8 +1884,8 @@ Gen4 plan same as above
                   "Name": "main",
                   "Sharded": false
                 },
-                "FieldQuery": "select 1 + id as foo, 1 from unsharded where 1 != 1",
-                "Query": "select distinct 1 + id as foo, 1 from unsharded",
+                "FieldQuery": "select 1 + id as foo, weight_string(1 + id), 1 from unsharded where 1 != 1",
+                "Query": "select distinct 1 + id as foo, weight_string(1 + id), 1 from unsharded",
                 "Table": "unsharded"
               }
             ]