kafka: default value of empty string #939

Author: ffffwh

drivers/mysql/kafka/kafka3.go

@@ -1246,9 +1246,6 @@ func kafkaColumnListToColDefs(colList *common.ColumnList, loc *time.Location) (v
 	for i, _ := range cols {
 		var field *Schema
 		defaultValue := cols[i].Default
-		if defaultValue == "" {
-			defaultValue = nil
-		}
 		optional := cols[i].Nullable
 		fieldName := cols[i].RawName
 		switch cols[i].Type {

drivers/mysql/mysql/base/utils.go

@@ -77,7 +77,6 @@ func ParseBinlogCoordinatesFromRow(row *sql.Row) (r *common.BinlogCoordinatesX,
 	return r, nil
 }
 
-// GetTableColumns reads column list from given table
 func GetTableColumns(db usql.QueryAble, databaseName, tableName string) (*common.ColumnList, error) {
 	databaseNameEscaped := umconf.EscapeName(databaseName)
 	tableNameEscaped := umconf.EscapeName(tableName)
@@ -88,10 +87,14 @@ func GetTableColumns(db usql.QueryAble, databaseName, tableName string) (*common
 		aColumn := umconf.Column{
 			RawName:    rowMap.GetString("Field"),
 			ColumnType: rowMap.GetString("Type"),
-			Default:    rowMap.GetString("Default"),
 			Key:        strings.ToUpper(rowMap.GetString("Key")),
 			Nullable:   strings.ToUpper(rowMap.GetString("Null")) == "YES",
 		}
+		if d, ok := rowMap["Default"]; ok {
+			if d.Valid {
+				aColumn.Default = d.String
+			}
+		}
 		aColumn.EscapedName = umconf.EscapeName(aColumn.RawName)
 		columns = append(columns, aColumn)
 		return nil

drivers/mysql/mysql/base/utils_test.go

@@ -17,8 +17,8 @@ import (
 
 	"github.com/actiontech/dtle/drivers/mysql/common"
 
-	test "github.com/outbrain/golib/tests"
 	gomysql "github.com/go-mysql-org/go-mysql/mysql"
+	test "github.com/outbrain/golib/tests"
 )
 
 func TestStringContainsAll(t *testing.T) {
@@ -175,6 +175,20 @@ func TestGetTableColumns(t *testing.T) {
 	}
 }
 
+//func TestGetTableColumns2(t *testing.T) {
+//	db, err := sql.CreateDB(fmt.Sprintf("%s@(%s)/?timeout=5s&tls=false&autocommit=true&charset=utf8mb4,utf8,latin1&multiStatements=true", "root:password", "10.186.62.40:3307"))
+//	if err != nil {
+//		t.Error(err)
+//	}
+//	cl, err := GetTableColumns(db, "a", "a939")
+//	if err != nil {
+//		t.Error(err)
+//	}
+//	for _, col := range cl.Columns {
+//		fmt.Printf("%v default '%v'\n", col.RawName, col.Default)
+//	}
+//}
+
 func TestApplyColumnTypes(t *testing.T) {
 	type args struct {
 		db           *gosql.Tx

helper/try-sqle-inspector/trysqleinspector.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"fmt"
 	"github.com/actiontech/dtle/drivers/mysql/mysql/base"
 	"github.com/actiontech/dtle/drivers/mysql/mysql/sqle/inspector"
 	//"github.com/actiontech/dtle/internal/client/driver/mysql/base"
@@ -28,7 +29,7 @@ func main() {
 	//ctx.LoadTables("a", nil)
 	//ctx.UseSchema("a")
 
-	case7()
+	case8()
 	//case6()
 	//case5()
 	//case4()
@@ -43,6 +44,23 @@ func panicIfErr(err interface{}, args ...interface{}) {
 	}
 }
 
+func case8() { // #939
+	do("create schema a")
+	ctx.LoadTables("a", nil)
+	ctx.UseSchema("a")
+
+	do(`create table a.a939 (id int primary key auto_increment,
+  val1 varchar(50) not null default 'aaa',
+  val2 varchar(50) not null default '',
+  val3 varchar(50) not null);`)
+
+	colList, _, err := base.GetTableColumnsSqle(ctx, "a", "a939")
+	panicIfErr(err)
+	for _, col := range colList.Columns {
+		fmt.Printf("%v default '%v'\n", col.RawName, col.Default)
+	}
+}
+
 func case7() {
 	do("create schema fk")
 	ctx.LoadTables("fk", nil)