diff --git a/expression/integration_test.go b/expression/integration_test.go
index 7e7a51412df8a..fb2bbfbbf1cca 100755
--- a/expression/integration_test.go
+++ b/expression/integration_test.go
@@ -7200,6 +7200,20 @@ func (s *testIntegrationSerialSuite) TestIssue18702(c *C) {
 	tk.MustQuery("SELECT * FROM t FORCE INDEX(idx_bc);").Check(testkit.Rows("1 A 10 1", "2 B 20 1"))
 }
 
+func (s *testIntegrationSerialSuite) TestCollationText(c *C) {
+	collate.SetNewCollationEnabledForTest(true)
+	defer collate.SetNewCollationEnabledForTest(false)
+
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(a TINYTEXT collate UTF8MB4_GENERAL_CI, UNIQUE KEY `a`(`a`(10)));")
+	tk.MustExec("insert into t (a) values ('A');")
+	tk.MustQuery("select * from t t1 inner join t t2 on t1.a = t2.a where t1.a = 'A';").Check(testkit.Rows("A A"))
+	tk.MustExec("update t set a = 'B';")
+	tk.MustExec("admin check table t;")
+}
+
 func (s *testIntegrationSuite) TestIssue18850(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use test")
diff --git a/session/bootstrap_test.go b/session/bootstrap_test.go
index e587aadbc8f74..2a59620a98d28 100644
--- a/session/bootstrap_test.go
+++ b/session/bootstrap_test.go
@@ -54,7 +54,7 @@ func (s *testBootstrapSuite) TestBootstrap(c *C) {
 	c.Assert(err, IsNil)
 	c.Assert(req.NumRows() == 0, IsFalse)
 	datums := statistics.RowToDatums(req.GetRow(0), r.Fields())
-	match(c, datums, `%`, "root", []byte(""), "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "N", "Y", "Y", "Y", "Y")
+	match(c, datums, `%`, "root", "", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "N", "Y", "Y", "Y", "Y")
 
 	c.Assert(se.Auth(&auth.UserIdentity{Username: "root", Hostname: "anyhost"}, []byte(""), []byte("")), IsTrue)
 	mustExecSQL(c, se, "USE test;")
@@ -159,7 +159,7 @@ func (s *testBootstrapSuite) TestBootstrapWithError(c *C) {
 	c.Assert(req.NumRows() == 0, IsFalse)
 	row := req.GetRow(0)
 	datums := statistics.RowToDatums(row, r.Fields())
-	match(c, datums, `%`, "root", []byte(""), "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "N", "Y", "Y", "Y", "Y")
+	match(c, datums, `%`, "root", "", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "Y", "N", "Y", "Y", "Y", "Y")
 	c.Assert(r.Close(), IsNil)
 
 	mustExecSQL(c, se, "USE test;")
diff --git a/table/column.go b/table/column.go
index d3fa703266285..29df9267b0cc0 100644
--- a/table/column.go
+++ b/table/column.go
@@ -604,10 +604,8 @@ func GetZeroValue(col *model.ColumnInfo) types.Datum {
 		} else {
 			d.SetString("", col.Collate)
 		}
-	case mysql.TypeVarString, mysql.TypeVarchar:
+	case mysql.TypeVarString, mysql.TypeVarchar, mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob:
 		d.SetString("", col.Collate)
-	case mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob:
-		d.SetBytes([]byte{})
 	case mysql.TypeDuration:
 		d.SetMysqlDuration(types.ZeroDuration)
 	case mysql.TypeDate:
diff --git a/table/column_test.go b/table/column_test.go
index feffb801ebf6a..826c9a8a4f11b 100644
--- a/table/column_test.go
+++ b/table/column_test.go
@@ -183,7 +183,7 @@ func (t *testTableSuite) TestGetZeroValue(c *C) {
 		},
 		{
 			types.NewFieldType(mysql.TypeBlob),
-			types.NewBytesDatum([]byte{}),
+			types.NewStringDatum(""),
 		},
 		{
 			types.NewFieldType(mysql.TypeDuration),
diff --git a/types/datum.go b/types/datum.go
index c48e5637ab460..08c989d366a74 100644
--- a/types/datum.go
+++ b/types/datum.go
@@ -2136,14 +2136,10 @@ func GetMaxValue(ft *FieldType) (max Datum) {
 		max.SetFloat32(float32(GetMaxFloat(ft.Flen, ft.Decimal)))
 	case mysql.TypeDouble:
 		max.SetFloat64(GetMaxFloat(ft.Flen, ft.Decimal))
-	case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar:
+	case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar, mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob:
 		// codec.Encode KindMaxValue, to avoid import circle
 		bytes := []byte{250}
 		max.SetString(string(bytes), ft.Collate)
-	case mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob:
-		// codec.Encode KindMaxValue, to avoid import circle
-		bytes := []byte{250}
-		max.SetBytes(bytes)
 	case mysql.TypeNewDecimal:
 		max.SetMysqlDecimal(NewMaxOrMinDec(false, ft.Flen, ft.Decimal))
 	case mysql.TypeDuration:
@@ -2171,14 +2167,10 @@ func GetMinValue(ft *FieldType) (min Datum) {
 		min.SetFloat32(float32(-GetMaxFloat(ft.Flen, ft.Decimal)))
 	case mysql.TypeDouble:
 		min.SetFloat64(-GetMaxFloat(ft.Flen, ft.Decimal))
-	case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar:
+	case mysql.TypeString, mysql.TypeVarString, mysql.TypeVarchar, mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob:
 		// codec.Encode KindMinNotNull, to avoid import circle
 		bytes := []byte{1}
 		min.SetString(string(bytes), ft.Collate)
-	case mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob:
-		// codec.Encode KindMinNotNull, to avoid import circle
-		bytes := []byte{1}
-		min.SetBytes(bytes)
 	case mysql.TypeNewDecimal:
 		min.SetMysqlDecimal(NewMaxOrMinDec(true, ft.Flen, ft.Decimal))
 	case mysql.TypeDuration:
diff --git a/util/chunk/row.go b/util/chunk/row.go
index 993ec9b58b9d1..0951de6803900 100644
--- a/util/chunk/row.go
+++ b/util/chunk/row.go
@@ -148,14 +148,10 @@ func (r Row) GetDatum(colIdx int, tp *types.FieldType) types.Datum {
 		if !r.IsNull(colIdx) {
 			d.SetFloat64(r.GetFloat64(colIdx))
 		}
-	case mysql.TypeVarchar, mysql.TypeVarString, mysql.TypeString:
+	case mysql.TypeVarchar, mysql.TypeVarString, mysql.TypeString, mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob:
 		if !r.IsNull(colIdx) {
 			d.SetString(r.GetString(colIdx), tp.Collate)
 		}
-	case mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob:
-		if !r.IsNull(colIdx) {
-			d.SetBytes(r.GetBytes(colIdx))
-		}
 	case mysql.TypeDate, mysql.TypeDatetime, mysql.TypeTimestamp:
 		if !r.IsNull(colIdx) {
 			d.SetMysqlTime(r.GetTime(colIdx))
diff --git a/util/rowcodec/decoder.go b/util/rowcodec/decoder.go
index 8352fc8fa2efc..cd8f7438578de 100644
--- a/util/rowcodec/decoder.go
+++ b/util/rowcodec/decoder.go
@@ -133,10 +133,8 @@ func (decoder *DatumMapDecoder) decodeColDatum(col *ColInfo, colData []byte) (ty
 			return d, err
 		}
 		d.SetFloat64(fVal)
-	case mysql.TypeVarString, mysql.TypeVarchar, mysql.TypeString:
+	case mysql.TypeVarString, mysql.TypeVarchar, mysql.TypeString, mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob:
 		d.SetString(string(colData), col.Collate)
-	case mysql.TypeBlob, mysql.TypeTinyBlob, mysql.TypeMediumBlob, mysql.TypeLongBlob:
-		d.SetBytes(colData)
 	case mysql.TypeNewDecimal:
 		_, dec, precision, frac, err := codec.DecodeDecimal(colData)
 		if err != nil {
diff --git a/util/rowcodec/rowcodec_test.go b/util/rowcodec/rowcodec_test.go
index 04b1c3455f5e1..dbe0c3568b8db 100644
--- a/util/rowcodec/rowcodec_test.go
+++ b/util/rowcodec/rowcodec_test.go
@@ -308,6 +308,8 @@ func (s *testSuite) TestTypesNewRowCodec(c *C) {
 			c.Assert(len(remain), Equals, 0)
 			if d.Kind() == types.KindMysqlDecimal {
 				c.Assert(d.GetMysqlDecimal(), DeepEquals, t.bt.GetMysqlDecimal())
+			} else if d.Kind() == types.KindBytes {
+				c.Assert(d.GetBytes(), DeepEquals, t.bt.GetBytes())
 			} else {
 				c.Assert(d, DeepEquals, t.bt)
 			}
@@ -341,9 +343,9 @@ func (s *testSuite) TestTypesNewRowCodec(c *C) {
 		},
 		{
 			24,
-			types.NewFieldType(mysql.TypeBlob),
-			types.NewBytesDatum([]byte("abc")),
-			types.NewBytesDatum([]byte("abc")),
+			types.NewFieldTypeWithCollation(mysql.TypeBlob, mysql.DefaultCollationName, types.UnspecifiedLength),
+			types.NewStringDatum("abc"),
+			types.NewStringDatum("abc"),
 			nil,
 			false,
 		},
@@ -470,8 +472,8 @@ func (s *testSuite) TestTypesNewRowCodec(c *C) {
 	testData[0].id = 1
 
 	// test large data
-	testData[3].dt = types.NewBytesDatum([]byte(strings.Repeat("a", math.MaxUint16+1)))
-	testData[3].bt = types.NewBytesDatum([]byte(strings.Repeat("a", math.MaxUint16+1)))
+	testData[3].dt = types.NewStringDatum(strings.Repeat("a", math.MaxUint16+1))
+	testData[3].bt = types.NewStringDatum(strings.Repeat("a", math.MaxUint16+1))
 	encodeAndDecode(c, testData)
 }