Merge pull request #3111 from dolthub/angela/compare_type

Enum and set conversions in comparison.Compare

Author: angelamayxie

enginetest/queries/script_queries.go

@@ -9320,6 +9320,12 @@ where
 					{"hijkl", []uint8("hijkl")},
 				},
 			},
+			{
+				Query: "select e from t where e like 'a%'",
+				Expected: []sql.Row{
+					{"abc"},
+				},
+			},
 		},
 	},
 	{
@@ -9764,9 +9770,6 @@ where
 				},
 			},
 			{
-				// this is failing due to a type coercion bug in comparison.Compare
-				// https://github.com/dolthub/dolt/issues/9510
-				Skip:  true,
 				Query: "select i, s + 0, s from t where s = '';",
 				Expected: []sql.Row{
 					{0, float64(0), ""},
@@ -9820,8 +9823,6 @@ where
 				},
 			},
 			{
-				// https://github.com/dolthub/dolt/issues/9510
-				Skip:  true,
 				Query: "select s from t where s like 'a%' order by s;",
 				Expected: []sql.Row{
 					{"abc"},

sql/expression/comparison.go

@@ -141,50 +141,20 @@ func (c *comparison) Compare(ctx *sql.Context, row sql.Row) (int, error) {
 		return c.Left().Type().Compare(ctx, left, right)
 	}
 
-	// ENUM, SET, and TIME must be excluded when doing comparisons, as they're too restrictive to use as a comparison
-	// base.
-	//
-	// The best overall method would be to assign type priority. For example, INT would have a higher priority than
-	// TINYINT. This could then be combined with the origin of the value (table column, procedure param, etc.) to
-	// determine the best type for any comparison (tie-breakers can be simple rules such as the current left preference).
-	var compareType sql.Type
-	collationPreference := sql.Collation_Default
-	switch c.Left().(type) {
-	case *GetField, *UserVar, *SystemVar, *ProcedureParam:
-		compareType = c.Left().Type()
-		if twc, ok := compareType.(sql.TypeWithCollation); ok {
-			collationPreference = twc.Collation()
-		}
-	default:
-		switch c.Right().(type) {
-		case *GetField, *UserVar, *SystemVar, *ProcedureParam:
-			compareType = c.Right().Type()
-			if twc, ok := compareType.(sql.TypeWithCollation); ok {
-				collationPreference = twc.Collation()
-			}
-		}
-	}
-	if compareType != nil {
-		_, isEnum := compareType.(sql.EnumType)
-		_, isSet := compareType.(sql.SetType)
-		_, isTime := compareType.(types.TimeType)
-		if !isEnum && !isSet && !isTime {
-			compareType = nil
-		}
+	l, r, compareType, err := c.castLeftAndRight(ctx, left, right)
+	if err != nil {
+		return 0, err
 	}
-	if compareType == nil {
-		left, right, compareType, err = c.castLeftAndRight(ctx, left, right)
-		if err != nil {
-			return 0, err
-		}
+
+	// Set comparison relies on empty strings not being converted yet
+	if types.IsSet(compareType) {
+		return compareType.Compare(ctx, left, right)
 	}
-	if _, isSet := compareType.(sql.SetType); !isSet && types.IsTextOnly(compareType) {
-		collationPreference, _ = c.CollationCoercibility(ctx)
-		stringCompareType := compareType.(sql.StringType)
+	collationPreference, _ := c.CollationCoercibility(ctx)
+	if stringCompareType, ok := compareType.(sql.StringType); ok && types.IsTextOnly(stringCompareType) {
 		compareType = types.MustCreateString(stringCompareType.Type(), stringCompareType.Length(), collationPreference)
 	}
-
-	return compareType.Compare(ctx, left, right)
+	return compareType.Compare(ctx, l, r)
 }
 
 func (c *comparison) evalLeftAndRight(ctx *sql.Context, row sql.Row) (interface{}, interface{}, error) {
@@ -204,6 +174,49 @@ func (c *comparison) evalLeftAndRight(ctx *sql.Context, row sql.Row) (interface{
 func (c *comparison) castLeftAndRight(ctx *sql.Context, left, right interface{}) (interface{}, interface{}, sql.Type, error) {
 	leftType := c.Left().Type()
 	rightType := c.Right().Type()
+
+	leftIsEnumOrSet := types.IsEnum(leftType) || types.IsSet(leftType)
+	rightIsEnumOrSet := types.IsEnum(rightType) || types.IsSet(rightType)
+	// Only convert if same Enum or Set
+	if leftIsEnumOrSet && rightIsEnumOrSet {
+		if types.TypesEqual(leftType, rightType) {
+			return left, right, leftType, nil
+		}
+	} else {
+		// If right side is convertible to enum/set, convert. Otherwise, convert left side
+		if leftIsEnumOrSet && (types.IsText(rightType) || types.IsNumber(rightType)) {
+			if r, inRange, err := leftType.Convert(ctx, right); inRange && err == nil {
+				return left, r, leftType, nil
+			} else {
+				l, err := types.TypeAwareConversion(ctx, left, leftType, rightType)
+				if err != nil {
+					return nil, nil, nil, err
+				}
+				return l, right, rightType, nil
+			}
+		}
+		// If left side is convertible to enum/set, convert. Otherwise, convert right side
+		if rightIsEnumOrSet && (types.IsText(leftType) || types.IsNumber(leftType)) {
+			if l, inRange, err := rightType.Convert(ctx, left); inRange && err == nil {
+				return l, right, rightType, nil
+			} else {
+				r, err := types.TypeAwareConversion(ctx, right, rightType, leftType)
+				if err != nil {
+					return nil, nil, nil, err
+				}
+				return left, r, leftType, nil
+			}
+		}
+	}
+
+	if types.IsTimespan(leftType) || types.IsTimespan(rightType) {
+		if l, err := types.Time.ConvertToTimespan(left); err == nil {
+			if r, err := types.Time.ConvertToTimespan(right); err == nil {
+				return l, r, types.Time, nil
+			}
+		}
+	}
+
 	if types.IsTuple(leftType) && types.IsTuple(rightType) {
 		return left, right, c.Left().Type(), nil
 	}

sql/types/enum.go

@@ -218,7 +218,7 @@ func (t EnumType) Convert(ctx context.Context, v interface{}) (interface{}, sql.
 		return t.Convert(ctx, string(value))
 	}
 
-	return nil, sql.InRange, ErrConvertingToEnum.New(v)
+	return nil, sql.OutOfRange, ErrConvertingToEnum.New(v)
 }
 
 // Equals implements the Type interface.

sql/types/set.go

@@ -129,6 +129,24 @@ func (t SetType) Compare(ctx context.Context, a interface{}, b interface{}) (int
 	au := ai.(uint64)
 	bu := bi.(uint64)
 
+	// If there's an empty string in the set, empty strings should match both 0 and an empty string bit field
+	if emptyStringBitField, ok := t.emptyStringBitField(); ok {
+		aIsEmptyString := isEmptyString(a)
+		bIsEmptyString := isEmptyString(b)
+		if aIsEmptyString {
+			if bu == 0 || bu == emptyStringBitField {
+				return 0, nil
+			}
+			return -1, nil
+		}
+		if bIsEmptyString {
+			if au == 0 || au == emptyStringBitField {
+				return 0, nil
+			}
+			return 1, nil
+		}
+	}
+
 	if au < bu {
 		return -1, nil
 	} else if au > bu {
@@ -180,7 +198,7 @@ func (t SetType) Convert(ctx context.Context, v interface{}) (interface{}, sql.C
 		return t.Convert(ctx, value.Decimal.BigInt().Uint64())
 	case string:
 		ret, err := t.convertStringToBitField(value)
-		return ret, sql.InRange, err
+		return ret, err == nil, err
 	case []byte:
 		return t.Convert(ctx, string(value))
 	}
@@ -364,7 +382,7 @@ func (t SetType) convertStringToBitField(str string) (uint64, error) {
 		return 0, nil
 	}
 	var bitField uint64
-	_, allowEmptyString := t.valToBit[""]
+	_, allowEmptyString := t.emptyStringBitField()
 	lastI := 0
 	var val string
 	for i := 0; i < len(str)+1; i++ {
@@ -410,3 +428,16 @@ func (t SetType) convertStringToBitField(str string) (uint64, error) {
 	}
 	return bitField, nil
 }
+
+func (t SetType) emptyStringBitField() (bitField uint64, ok bool) {
+	bitField, ok = t.valToBit[""]
+	return bitField, ok
+}
+
+func isEmptyString(val interface{}) bool {
+	switch v := val.(type) {
+	case string:
+		return v == ""
+	}
+	return false
+}