Backport [#8730 to 12.0]: Proper handling of CHAR columns with binary collations

go/test/endtoend/onlineddl/vrepl_suite/onlineddl_vrepl_suite_test.go

@@ -58,7 +58,7 @@ var (
 
 const (
 	testDataPath   = "testdata"
-	defaultSQLMode = "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
+	defaultSQLMode = "ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_ENGINE_SUBSTITUTION"
 )
 
 func TestMain(m *testing.M) {

go/test/endtoend/onlineddl/vrepl_suite/testdata/char-collate-binary/alter

@@ -0,0 +1 @@
+change id id bigint

go/test/endtoend/onlineddl/vrepl_suite/testdata/char-collate-binary/create.sql

@@ -0,0 +1,24 @@
+drop table if exists onlineddl_test;
+create table onlineddl_test (
+  id bigint auto_increment,
+  country_code char(3) collate utf8mb4_bin,
+  primary key(id)
+) auto_increment=1;
+
+insert into onlineddl_test values (null, 'ABC');
+insert into onlineddl_test values (null, 'DEF');
+insert into onlineddl_test values (null, 'GHI');
+
+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 (null, 'jkl');
+  insert into onlineddl_test values (null, 'MNO');
+end ;;

go/vt/vttablet/tabletserver/vstreamer/vstreamer.go

@@ -895,6 +895,34 @@ func (vs *vstreamer) extractRowAndFilter(plan *streamerPlan, data []byte, dataCo
 		}
 		pos += l
 
+		// If this is a binary type in the binlog event but actually a CHAR column *with a
+		// binary collation*, then we need to factor in the max bytes per character of 3 for
+		// utf8[mb3] and 4 for utf8mb4 and trim the added null-byte padding as needed to accomodate
+		// for that
+		if value.IsBinary() && sqltypes.IsBinary(plan.Table.Fields[colNum].Type) {
+			maxBytesPerChar := uint32(1)
+			if plan.Table.Fields[colNum].Charset == uint32(mysql.CharacterSetMap["utf8"]) {
+				maxBytesPerChar = 3
+			} else if plan.Table.Fields[colNum].Charset == uint32(mysql.CharacterSetMap["utf8mb4"]) {
+				maxBytesPerChar = 4
+			}
+
+			if maxBytesPerChar > 1 {
+				maxCharLen := plan.Table.Fields[colNum].ColumnLength / maxBytesPerChar
+				if uint32(value.Len()) > maxCharLen {
+					originalVal := value.ToBytes()
+
+					// Let's be sure that we're not going to be trimming non-null bytes
+					firstNullBytePos := bytes.IndexByte(originalVal, byte(0))
+					if uint32(firstNullBytePos) <= maxCharLen {
+						rightSizedVal := make([]byte, maxCharLen)
+						copy(rightSizedVal, originalVal)
+						value = sqltypes.MakeTrusted(querypb.Type_BINARY, rightSizedVal)
+					}
+				}
+			}
+		}
+
 		values[colNum] = value
 		valueIndex++
 	}