From 63d4ae31374d3cbc8c9401ac2389811ea41047be Mon Sep 17 00:00:00 2001
From: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
Date: Wed, 14 Aug 2024 09:52:10 +0300
Subject: [PATCH] Online DDL: do not CONVERT column charset when migration does
 not modify column's charset or collation

Signed-off-by: Shlomi Noach <2607934+shlomi-noach@users.noreply.github.com>
---
 .../allow_schemadiff_normalization            |  0
 .../testdata/non-utf8-charset-pk/create.sql   | 30 +++++++++++++++++++
 .../non-utf8-charset-pk/ignore_versions       |  1 +
 go/vt/vttablet/onlineddl/vrepl.go             |  3 +-
 4 files changed, 33 insertions(+), 1 deletion(-)
 create mode 100644 go/test/endtoend/onlineddl/vrepl_suite/testdata/non-utf8-charset-pk/allow_schemadiff_normalization
 create mode 100644 go/test/endtoend/onlineddl/vrepl_suite/testdata/non-utf8-charset-pk/create.sql
 create mode 100644 go/test/endtoend/onlineddl/vrepl_suite/testdata/non-utf8-charset-pk/ignore_versions

diff --git a/go/test/endtoend/onlineddl/vrepl_suite/testdata/non-utf8-charset-pk/allow_schemadiff_normalization b/go/test/endtoend/onlineddl/vrepl_suite/testdata/non-utf8-charset-pk/allow_schemadiff_normalization
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/go/test/endtoend/onlineddl/vrepl_suite/testdata/non-utf8-charset-pk/create.sql b/go/test/endtoend/onlineddl/vrepl_suite/testdata/non-utf8-charset-pk/create.sql
new file mode 100644
index 00000000000..c0313e62c8d
--- /dev/null
+++ b/go/test/endtoend/onlineddl/vrepl_suite/testdata/non-utf8-charset-pk/create.sql
@@ -0,0 +1,30 @@
+drop table if exists onlineddl_test;
+create table onlineddl_test (
+  id varchar(128)  charset latin1 collate latin1_swedish_ci,
+  t1 varchar(128)  charset latin1 collate latin1_swedish_ci,
+  t2 varchar(128)  charset latin1 collate latin1_swedish_ci,
+  tutf8 varchar(128) charset utf8,
+  tutf8mb4 varchar(128) charset utf8mb4,
+  tlatin1 varchar(128)  charset latin1 collate latin1_swedish_ci,
+  primary key(id)
+) auto_increment=1;
+
+insert into onlineddl_test values (concat('átesting-', md5(rand())), md5(rand()), md5(rand()), md5(rand()), md5(rand()), md5(rand()));
+insert into onlineddl_test values (concat('átesting-', md5(rand())), 'átesting', 'átesting', 'átesting', 'átesting', 'átesting');
+insert into onlineddl_test values (concat('átesting-', md5(rand())), 'testátest', 'testátest', 'testátest', '🍻😀', 'átesting');
+
+drop event if exists onlineddl_test;
+delimiter ;;
+create event onlineddl_test
+  on schedule every 1 second
+  starts current_timestamp
+  ends current_timestamp + interval 60 second
+  on completion not preserve
+  enable
+  do
+begin
+  insert into onlineddl_test values (concat('átesting-', md5(rand())), md5(rand()), md5(rand()), md5(rand()), md5(rand()), md5(rand()));
+  insert into onlineddl_test values (concat('átesting-', md5(rand())), 'átesting-binlog', 'átesting-binlog', 'átesting-binlog', 'átesting-binlog', 'átesting-binlog');
+  insert into onlineddl_test values (concat('átesting-', md5(rand())), 'testátest-binlog', 'testátest-binlog', 'testátest-binlog', '🍻😀', 'átesting-binlog');
+  insert into onlineddl_test values (concat('átesting-', md5(rand())), 'átesting-bnull', 'átesting-bnull', 'átesting-bnull', null, null);
+end ;;
diff --git a/go/test/endtoend/onlineddl/vrepl_suite/testdata/non-utf8-charset-pk/ignore_versions b/go/test/endtoend/onlineddl/vrepl_suite/testdata/non-utf8-charset-pk/ignore_versions
new file mode 100644
index 00000000000..0790a1e68fd
--- /dev/null
+++ b/go/test/endtoend/onlineddl/vrepl_suite/testdata/non-utf8-charset-pk/ignore_versions
@@ -0,0 +1 @@
+(5.5|5.6|5.7)
diff --git a/go/vt/vttablet/onlineddl/vrepl.go b/go/vt/vttablet/onlineddl/vrepl.go
index 14c52d352bf..097a286105a 100644
--- a/go/vt/vttablet/onlineddl/vrepl.go
+++ b/go/vt/vttablet/onlineddl/vrepl.go
@@ -301,9 +301,10 @@ func (v *VRepl) generateFilterQuery() error {
 			if targetCol.IsTextual() && toCollation == collations.Unknown {
 				return vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "Character set %s not supported for column %s", targetCol.Charset(), targetCol.Name())
 			}
-
 			if trivialCharset(fromCollation) && trivialCharset(toCollation) && targetCol.Type() != "json" {
 				sb.WriteString(escapeName(name))
+			} else if fromCollation == toCollation && targetCol.Type() != "json" {
+				sb.WriteString(escapeName(name))
 			} else {
 				v.convertCharset[targetName] = &binlogdatapb.CharsetConversion{
 					FromCharset: sourceCol.Charset(),