handle timestamp default value for kafka #537

Author: ffffwh

driver/mysql/base/utils.go

@@ -543,11 +543,10 @@ func GetTableColumnsSqle(sqleContext *sqle.Context, schema string,
 				newColumn.Nullable = false
 			case ast.ColumnOptionAutoIncrement:
 			case ast.ColumnOptionDefaultValue:
-				value, ok := colOpt.Expr.(ast.ValueExpr)
-				if !ok {
-					newColumn.Default = nil
-				} else {
-					switch v := value.GetValue().(type) {
+				expectNil := false
+				switch exp := colOpt.Expr.(type) {
+				case ast.ValueExpr:
+					switch v := exp.GetValue().(type) {
 					case int64, uint64, float32, float64, string, []byte:
 						newColumn.Default = v
 					case types.BinaryLiteral:
@@ -556,11 +555,20 @@ func GetTableColumnsSqle(sqleContext *sqle.Context, schema string,
 						newColumn.Default = v.String()
 					case nil:
 						newColumn.Default = nil
-					default:
-						newColumn.Default = nil
-						g.Logger.Warn("GetTableColumnsSqle: unknown type for a default value",
-							"schema", schema, "table", table, "type", hclog.Fmt("%T", v))
+						expectNil = true
 					}
+				case *ast.FuncCallExpr:
+					if col.Tp.Tp == parsermysql.TypeTimestamp {
+						if exp.FnName.L == "current_timestamp" {
+							// See # 537. "1970-01-01 00:00:00" is debezium's representation for CURRENT_TIMESTAMP.
+							// TODO It would be better to put it in dest/kafka.
+							newColumn.Default = "1970-01-01 00:00:00"
+						}
+					}
+				}
+				if newColumn.Default == nil && !expectNil {
+					g.Logger.Warn("GetTableColumnsSqle: cannot handle a default value",
+						"schema", schema, "table", table, "column", col.Name, "type", hclog.Fmt("%T", colOpt.Expr))
 				}
 			case ast.ColumnOptionUniqKey:
 				newColumn.Key = "UNI"

driver/mysql/base/utils_test.go

@@ -255,6 +255,7 @@ func TestGetTableColumnsSqle(t *testing.T) {
 	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))",
+		"create table a.timestamp1(id int(11) primary key, val timestamp default current_timestamp)",
 	}
 	for i := range sqls {
 		stmt, err := p.ParseOneStmt(sqls[i], "", "")
@@ -293,6 +294,15 @@ func TestGetTableColumnsSqle(t *testing.T) {
 			},
 			want:    nil,
 			wantErr: false,
+		}, {
+			name:    "timestamp",
+			args:    args{
+				sqleContext: sqleCtx,
+				schema:      "a",
+				table:       "timestamp1",
+			},
+			want:    nil,
+			wantErr: false,
 		},
 	}
 	for _, tt := range tests {