workaround of #717

Following 40fae59. Fix incorrect base 64 conversion in kafka.

Author: ffffwh

drivers/mysql/kafka/kafka3.go

@@ -870,10 +870,10 @@ func (kr *KafkaRunner) kafkaTransformDMLEventQueries(dmlEntries []*common.Binlog
 					}
 				case mysqlconfig.VarbinaryColumnType:
 					if beforeValue != nil {
-						beforeValue = base64.StdEncoding.EncodeToString([]byte(beforeValue.(string)))
+						beforeValue = encodeStringInterfaceToBase64String(beforeValue)
 					}
 					if afterValue != nil {
-						afterValue = base64.StdEncoding.EncodeToString([]byte(afterValue.(string)))
+						afterValue = encodeStringInterfaceToBase64String(afterValue)
 					}
 				case mysqlconfig.BinaryColumnType:
 					if beforeValue != nil {
@@ -914,7 +914,28 @@ func (kr *KafkaRunner) kafkaTransformDMLEventQueries(dmlEntries []*common.Binlog
 						afterValue = getSetValue(afterValue.(int64), columnType)
 					}
 				case mysqlconfig.BlobColumnType:
-					// do nothing. just keep the string value.
+					if strings.Contains(colList[i].ColumnType, "text") {
+						// already string value
+					} else {
+						if beforeValue != nil {
+							beforeValue = encodeStringInterfaceToBase64String(beforeValue)
+						}
+						if afterValue != nil {
+							afterValue = encodeStringInterfaceToBase64String(afterValue)
+						}
+					}
+				case mysqlconfig.VarcharColumnType:
+					// workaround of #717
+					if strings.Contains(colList[i].ColumnType, "binary") {
+						if beforeValue != nil {
+							beforeValue = encodeStringInterfaceToBase64String(beforeValue)
+						}
+						if afterValue != nil {
+							afterValue = encodeStringInterfaceToBase64String(afterValue)
+						}
+					} else {
+						// keep as is
+					}
 				case mysqlconfig.TextColumnType:
 					if beforeValue != nil {
 						beforeValue = castBytesOrStringToString(beforeValue)
@@ -1080,6 +1101,7 @@ func getSetValue(num int64, set string) string {
 }
 
 func getBinaryValue(binary string, value string) string {
+	// binary = "binary(64)"
 	binaryLen := binary[7 : len(binary)-1]
 	lens, err := strconv.Atoi(binaryLen)
 	if err != nil {
@@ -1225,3 +1247,7 @@ func castBytesOrStringToString(v interface{}) string {
 		panic("only []byte or string is allowed")
 	}
 }
+
+func encodeStringInterfaceToBase64String(v interface{}) string {
+	return base64.StdEncoding.EncodeToString([]byte(v.(string)))
+}

drivers/mysql/kafka/kafka3_test.go

@@ -0,0 +1,31 @@
+package kafka
+
+import "testing"
+
+func Test_getBinaryValue(t *testing.T) {
+	type args struct {
+		binary string
+		value  string
+	}
+	tests := []struct {
+		name string
+		args args
+		want string
+	}{
+		{
+			name: "binary1",
+			args: args{
+				binary: "binary(16)",
+				value:  "",
+			},
+			want: "AAAAAAAAAAAAAAAAAAAAAA==",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := getBinaryValue(tt.args.binary, tt.args.value); got != tt.want {
+				t.Errorf("getBinaryValue() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}

drivers/mysql/mysql/base/utils_test.go

@@ -8,6 +8,9 @@ package base
 
 import (
 	gosql "database/sql"
+	"fmt"
+	sqle "github.com/actiontech/dtle/drivers/mysql/mysql/sqle/inspector"
+	"github.com/pingcap/parser"
 	"reflect"
 	"testing"
 	"time"
@@ -264,3 +267,66 @@ func Test_stringInterval(t *testing.T) {
 		})
 	}
 }
+
+func TestGetTableColumnsSqle(t *testing.T) {
+	// Not a real test. Just printing intermediate result.
+	sqleCtx := sqle.NewContext(nil)
+	sqleCtx.AddSchema("a"); sqleCtx.LoadSchemas(nil); sqleCtx.LoadTables("a", nil)
+	p := parser.New()
+	sqls := []string{
+		"create table a.text_columns(id int(11) not null primary key,c_text longtext)",
+		"create table a.binary_columns(id int(11) not null primary key, c_binary varbinary(255))",
+	}
+	for i := range sqls {
+		stmt, err := p.ParseOneStmt(sqls[i], "", "")
+		if err != nil {
+			t.Error(err)
+		}
+		sqleCtx.UpdateContext(stmt, "mysql")
+	}
+
+	type args struct {
+		sqleContext *sqle.Context
+		schema      string
+		table       string
+	}
+	tests := []struct {
+		name    string
+		args    args
+		want    *common.ColumnList
+		wantErr bool
+	}{
+		{
+			name:    "column-text",
+			args:    args{
+				sqleContext: sqleCtx,
+				schema:      "a",
+				table:       "text_columns",
+			},
+			want:    nil,
+			wantErr: false,
+		}, {
+			name:    "column-binary",
+			args:    args{
+				sqleContext: sqleCtx,
+				schema:      "a",
+				table:       "binary_columns",
+			},
+			want:    nil,
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := GetTableColumnsSqle(tt.args.sqleContext, tt.args.schema, tt.args.table)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("GetTableColumnsSqle() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			fmt.Printf("table %v\n", tt.args.table)
+			for i := range got.Columns {
+				println(got.Columns[i].Type)
+			}
+		})
+	}
+}