executor: fix prepared protocol charset (#58872)

pkg/executor/prepared.go

@@ -62,6 +62,8 @@ type PrepareExec struct {
 	// If it's generated from executing "prepare stmt from '...'", the process is parse -> plan -> executor
 	// If it's generated from the prepare protocol, the process is session.PrepareStmt -> NewPrepareExec
 	// They both generate a PrepareExec struct, but the second case needs to reset the statement context while the first already do that.
+	// Also, the second case need charset_client param since SQL is directly passed from clients.
+	// While the text-prepare already transformed charset by parser.
 	needReset bool
 }
 
@@ -87,23 +89,28 @@ func (e *PrepareExec) Next(ctx context.Context, _ *chunk.Chunk) error {
 			return nil
 		}
 	}
-	charset, collation := vars.GetCharsetInfo()
 	var (
 		stmts []ast.StmtNode
 		err   error
 	)
+	var params []parser.ParseParam
+	if e.needReset {
+		params = vars.GetParseParams()
+	} else {
+		var paramsArr [2]parser.ParseParam
+		charset, collation := vars.GetCharsetInfo()
+		paramsArr[0] = parser.CharsetConnection(charset)
+		paramsArr[1] = parser.CollationConnection(collation)
+		params = paramsArr[:]
+	}
 	if sqlParser, ok := e.Ctx().(sqlexec.SQLParser); ok {
 		// FIXME: ok... yet another parse API, may need some api interface clean.
-		stmts, _, err = sqlParser.ParseSQL(ctx, e.sqlText,
-			parser.CharsetConnection(charset),
-			parser.CollationConnection(collation))
+		stmts, _, err = sqlParser.ParseSQL(ctx, e.sqlText, params...)
 	} else {
 		p := parser.New()
 		p.SetParserConfig(vars.BuildParserConfig())
 		var warns []error
-		stmts, warns, err = p.ParseSQL(e.sqlText,
-			parser.CharsetConnection(charset),
-			parser.CollationConnection(collation))
+		stmts, warns, err = p.ParseSQL(e.sqlText, params...)
 		for _, warn := range warns {
 			e.Ctx().GetSessionVars().StmtCtx.AppendWarning(util.SyntaxWarn(warn))
 		}

pkg/executor/prepared_test.go

@@ -1172,3 +1172,26 @@ func TestPrepareProtocolWorkWithForeignKey(t *testing.T) {
 	// As schema version increased, the plan cache should be invalidated.
 	tk.MustQuery("select @@last_plan_from_cache").Check(testkit.Rows("0"))
 }
+
+func TestIssue58870(t *testing.T) {
+	store := testkit.CreateMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+
+	tk.MustExec("use test")
+	tk.MustExec("set names GBK")
+	tk.MustExec(`CREATE TABLE tsecurity  (
+  security_id int(11) NOT NULL DEFAULT 0,
+  mkt_id smallint(6) NOT NULL DEFAULT 0,
+  security_code varchar(64) CHARACTER SET gbk COLLATE gbk_bin NOT NULL DEFAULT ' ',
+  security_name varchar(128) CHARACTER SET gbk COLLATE gbk_bin NOT NULL DEFAULT ' ',
+  PRIMARY KEY (security_id) USING BTREE
+) ENGINE = InnoDB CHARACTER SET = gbk COLLATE = gbk_bin ROW_FORMAT = Compact;`)
+	tk.MustExec("INSERT INTO tsecurity (security_id, security_code, mkt_id, security_name) VALUES (1, '1', 1 ,'\xB2\xE2')")
+	tk.MustExec("PREPARE a FROM 'INSERT INTO tsecurity (security_id, security_code, mkt_id, security_name) VALUES (2, 2, 2 ,\"\xB2\xE2\")'")
+	tk.MustExec("EXECUTE a")
+	stmt, _, _, err := tk.Session().PrepareStmt("INSERT INTO tsecurity (security_id, security_code, mkt_id, security_name) VALUES (3, 3, 3 ,\"\xB2\xE2\")")
+	require.Nil(t, err)
+	rs, err := tk.Session().ExecutePreparedStmt(context.TODO(), stmt, nil)
+	require.Nil(t, err)
+	require.Nil(t, rs)
+}