sink(ticdc): fix incorrect encoding default value in Avro protocol (#…

…11995) (#12010)

close #11994

pkg/sink/codec/avro/avro.go

@@ -555,8 +555,18 @@ func (a *BatchEncoder) columns2AvroSchema(
 			}
 		} else {
 			if col.Flag.IsNullable() {
+				// the string literal "null" must be coerced to a `nil`
+				// see https://github.com/linkedin/goavro/blob/5ec5a5ee7ec82e16e6e2b438d610e1cab2588393/record.go#L109-L114
 				// https://stackoverflow.com/questions/22938124/avro-field-default-values
+				defaultFirst := false
 				if defaultValue == nil {
+					defaultFirst = true
+				} else if s, ok := defaultValue.(string); ok && s == "null" {
+					defaultFirst = true
+				} else if b, ok := defaultValue.([]byte); ok && string(b) == "null" {
+					defaultFirst = true
+				}
+				if defaultFirst {
 					field["type"] = []interface{}{"null", avroType}
 				} else {
 					field["type"] = []interface{}{avroType, "null"}

tests/integration_tests/avro_basic/data/data.sql

@@ -164,6 +164,19 @@ insert into t(c_tinyint, c_mediumint, c_int, c_bigint, a) values (4, 5, 6, 7, 8)
 alter table t modify c_mediumint varchar(10) null;
 insert into t(c_tinyint, c_mediumint, c_int, c_bigint, a) values (5, "234", 6, 7, 8);
 
+create table t1(
+    id int primary key,
+    c1 varchar(255) default "null",
+    c2 varchar(255) default "NULL",
+    c3 varchar(255) default null
+);
+
+insert into t1(id) values(1);
+update t1 set c1 = "null", c2 = "NULL", c3 = null where id = 1;
+alter table t1 add column col json not null;
+alter table t1 modify column col json default null;
+insert into t1(id) values(2);
+
 create table finish_mark
 (
     id int PRIMARY KEY