Repository: tikv/tikv (language: rust)

Issue:
`encoding failed` when processing queries with GBK/GB18030
## Bug Report

### What version of TiKV are you using?

master

### What operating system and CPU are you using?

not system related bug

### Steps to reproduce

1. start a cluster with pd, tidb and tikv
2. execute the following queries

```sql
CREATE TABLE t1 (c VARCHAR(4) CHARACTER SET gbk);
INSERT INTO t1 VALUES (0x8BF5819AEDC3); --'嬽仛砻', or others like (0x90459958) '怑橷'
SELECT ANY_VALUE(HEX(c)), COUNT(c) FROM t1 GROUP BY c COLLATE gbk_chinese_ci;
-- or SELECT ANY_VALUE(HEX(c)), COUNT(c) FROM t1 GROUP BY c COLLATE gbk_bin;
```

### What did you expect?

All statements succeed to execute (mysql-server and tidb with unistore can execute)

### What did happened?

Encounter `(1105, 'encoding failed')` for last statement


Discussion / hints:
/assign
Check the query plan:
```sql
TiDB root@127.0.0.1:test> explain SELECT ANY_VALUE(HEX(c)), COUNT(c) FROM t1 GROUP BY c COLLATE gbk_chinese_ci;
+-------------------------+----------+-----------+---------------+------------------------------------------------------------------------------------------------------------------------------------------------------+
| id                      | estRows  | task      | access object | operator info                                                                                                                                        |
+-------------------------+----------+-----------+---------------+------------------------------------------------------------------------------------------------------------------------------------------------------+
| Projection_4            | 8000.00  | root      |               | any_value(hex(to_binary(test.t1.c)))->Column#4, Column#3                                                                                             |
| └─HashAgg_9             | 8000.00  | root      |               | group by:Column#6, funcs:count(Column#7)->Column#3, funcs:firstrow(Column#8)->test.t1.c                                                              |
|   └─TableReader_10      | 8000.00  | root      |               | data:HashAgg_5                                                                                                                                       |
|     └─HashAgg_5         | 8000.00  | cop[tikv] |               | group by:cast(test.t1.c, varchar(4) CHARACTER SET gbk COLLATE gbk_chinese_ci), funcs:count(test.t1.c)->Column#7, funcs:firstrow(test.t1.c)->Column#8 |
|       └─TableFullScan_8 | 10000.00 | cop[tikv] | table:t1      | keep order:false, stats:pseudo                                                                                                                       |
+-------------------------+----------+-----------+---------------+------------------------------------------------------------------------------------------------------------------------------------------------------+
5 rows in set
Time: 0.013s
```

The error happens at https://github.com/tikv/tikv/blob/6f389be835d5b8c4362c4989dca925af077e63f7/components/tidb_query_datatype/src/codec/collation/mod.rs#L216


The cause is that, the length of `inner` is truncated to `4` bytes somewhere, which is not expected. `4` in the `varchar(4)` means '4 chars', not '4 bytes'
Reproduced in 6.5.11, 7.1.6 and 7.5.4
Not repro in 5.4 and 6.1, since the hash agg is not pushed down to tikv
```sql
TiDB root@127.0.0.1:test> explain SELECT ANY_VALUE(HEX(c)), COUNT(c) FROM
                       -> t1 GROUP BY c COLLATE gbk_chinese_ci;
+--------------------------+----------+-----------+---------------+-------------------------------------------------------------------------------------------------------+
| id                       | estRows  | task      | access object | operator info                                                                                         |
+--------------------------+----------+-----------+---------------+-------------------------------------------------------------------------------------------------------+
| Projection_4             | 8000.00  | root      |               | any_value(hex(to_binary(test.t1.c)))->Column#4, Column#3                                              |
| └─HashAgg_7              | 8000.00  | root      |               | group by:Column#11, funcs:count(Column#9)->Column#3, funcs:firstrow(Column#10)->test.t1.c             |
|   └─Projection_13        | 10000.00 | root      |               | test.t1.c, test.t1.c, cast(test.t1.c, varchar(4) CHARACTER SET gbk COLLATE gbk_chinese_ci)->Column#11 |
|     └─TableReader_12     | 10000.00 | root      |               | data:TableFullScan_11                                                                                 |
|       └─TableFullScan_11 | 10000.00 | cop[tikv] | table:t1      | keep order:false, stats:pseudo                                                                        |
+--------------------------+----------+-----------+---------------+-------------------------------------------------------------------------------------------------------+
```

Fix patch:
diff --git a/components/tidb_query_datatype/src/codec/convert.rs b/components/tidb_query_datatype/src/codec/convert.rs
index e4e2a3d2e8f..d5cc2381596 100644
--- a/components/tidb_query_datatype/src/codec/convert.rs
+++ b/components/tidb_query_datatype/src/codec/convert.rs
@@ -5,7 +5,7 @@ use std::{borrow::Cow, fmt::Display};
 use tipb::FieldType;
 
 use super::{
-    mysql::{RoundMode, DEFAULT_FSP},
+    mysql::{charset::MULTI_BYTES_CHARSETS, RoundMode, DEFAULT_FSP},
     Error, Result,
 };
 // use crate::{self, FieldTypeTp, UNSPECIFIED_LENGTH};
@@ -13,7 +13,7 @@ use crate::{
     codec::{
         data_type::*,
         error::ERR_DATA_OUT_OF_RANGE,
-        mysql::{charset, decimal::max_or_min_dec, Res},
+        mysql::{decimal::max_or_min_dec, Res},
     },
     expr::{EvalContext, Flag},
     Collation, FieldTypeAccessor, FieldTypeTp, UNSPECIFIED_LENGTH,
@@ -713,9 +713,10 @@ pub fn produce_str_with_specified_tp<'a>(
         return Ok(s);
     }
     let flen = flen as usize;
-    // flen is the char length, not byte length, for UTF8 charset, we need to
-    // calculate the char count and truncate to flen chars if it is too long.
-    if chs == charset::CHARSET_UTF8 || chs == charset::CHARSET_UTF8MB4 {
+    // flen is the char length, not byte length, for UTF8 and GBK charset,
+    // we need to calculate the char count and truncate to flen chars if it is
+    // too long.
+    if MULTI_BYTES_CHARSETS.contains(chs) {
         let (char_count, truncate_pos) = {
             let s = &String::from_utf8_lossy(&s);
             let mut truncate_pos = 0;
@@ -1142,7 +1143,7 @@ mod tests {
                 ERR_DATA_OUT_OF_RANGE, ERR_M_BIGGER_THAN_D, ERR_TRUNCATE_WRONG_VALUE,
                 WARN_DATA_TRUNCATED,
             },
-            mysql::{Res, UNSPECIFIED_FSP},
+            mysql::{charset, Res, UNSPECIFIED_FSP},
         },
         expr::{EvalConfig, EvalContext, Flag},
         Collation, FieldTypeFlag,
@@ -2221,6 +2222,13 @@ mod tests {
             ("世界，中国", 4, charset::CHARSET_ASCII),
             ("世界，中国", 5, charset::CHARSET_ASCII),
             ("世界，中国", 6, charset::CHARSET_ASCII),
+            // GBK
+            ("世界，中国", 1, charset::CHARSET_GBK),
+            ("世界，中国", 2, charset::CHARSET_GBK),
+            ("世界，中国", 3, charset::CHARSET_GBK),
+            ("世界，中国", 4, charset::CHARSET_GBK),
+            ("世界，中国", 5, charset::CHARSET_GBK),
+            ("世界，中国", 6, charset::CHARSET_GBK),
         ];
 
         let cfg = EvalConfig::from_flag(Flag::TRUNCATE_AS_WARNING);
@@ -2232,10 +2240,10 @@ mod tests {
             ft.set_flen(char_num);
             let bs = s.as_bytes();
             let r = produce_str_with_specified_tp(&mut ctx, Cow::Borrowed(bs), &ft, false);
-            assert!(r.is_ok(), "{}, {}, {}", s, char_num, cs);
+            assert!(r.is_ok(), "{}, {}, {}, {}", s, char_num, cs, r.unwrap_err());
             let p = r.unwrap();
 
-            if cs == charset::CHARSET_UTF8MB4 || cs == charset::CHARSET_UTF8 {
+            if MULTI_BYTES_CHARSETS.contains(cs) {
                 let ns: String = s.chars().take(char_num as usize).collect();
                 assert_eq!(p.as_ref(), ns.as_bytes(), "{}, {}, {}", s, char_num, cs);
             } else {
@@ -2271,7 +2279,7 @@ mod tests {
             ft.set_flen(char_num);
             let bs = s.as_bytes();
             let r = produce_str_with_specified_tp(&mut ctx, Cow::Borrowed(bs), &ft, true);
-            assert!(r.is_ok(), "{}, {}, {}", s, char_num, cs);
+            assert!(r.is_ok(), "{}, {}, {}, {}", s, char_num, cs, r.unwrap_err());
 
             let p = r.unwrap();
             assert_eq!(p.len(), char_num as usize, "{}, {}, {}", s, char_num, cs);
@@ -2306,7 +2314,15 @@ mod tests {
             ft.set_charset(cs.to_string());
             ft.set_flen(char_num);
             let r = produce_str_with_specified_tp(&mut ctx, Cow::Borrowed(&s), &ft, true);
-            assert!(r.is_ok(), "{:?}, {}, {}, {:?}", &s, char_num, cs, result);
+            assert!(
+                r.is_ok(),
+                "{:?}, {}, {}, {:?}, {}",
+                &s,
+                char_num,
+                cs,
+                result,
+                r.unwrap_err()
+            );
 
             let p = r.unwrap();
             assert_eq!(p, result, "{:?}, {}, {}, {:?}", &s, char_num, cs, result);
diff --git a/components/tidb_query_datatype/src/codec/mysql/charset.rs b/components/tidb_query_datatype/src/codec/mysql/charset.rs
index 0ac2655c619..4b0ec7bc125 100644
--- a/components/tidb_query_datatype/src/codec/mysql/charset.rs
+++ b/components/tidb_query_datatype/src/codec/mysql/charset.rs
@@ -1,5 +1,7 @@
 // Copyright 2017 TiKV Project Authors. Licensed under Apache-2.0.
 
+use lazy_static::*;
+
 /// `CHARSET_BIN` is used for marking binary charset.
 pub const CHARSET_BIN: &str = "binary";
 /// `CHARSET_UTF8` is the default charset for string types.
@@ -13,6 +15,12 @@ pub const CHARSET_ASCII: &str = "ascii";
 pub const CHARSET_LATIN1: &str = "latin1";
 /// `CHARSET_GBK` is Chinese character set.
 pub const CHARSET_GBK: &str = "gbk";
+// For a new implemented multi-byte charset, add it to MULTI_BYTES_CHARSETS
 
-/// All utf8 charsets.
-pub const UTF8_CHARSETS: &[&str] = &[CHARSET_UTF8, CHARSET_UTF8MB4, CHARSET_ASCII];
+lazy_static! {
+    pub static ref MULTI_BYTES_CHARSETS: collections::HashSet<&'static str> =
+        [CHARSET_UTF8, CHARSET_UTF8MB4, CHARSET_GBK,]
+            .iter()
+            .cloned()
+            .collect();
+}


Test patch: