Added option to cast some columns to type expected by Tarantool (#24)

Author: pparshin

README.md

@@ -38,6 +38,34 @@ Replicator reads primary keys from MySQL table info and sync them automatically.
 Updating primary key in MySQL causes two Tarantool requests: delete an old row and insert a new one, because
 it is illegal to update primary key in Tarantool.
 
+### Force cast MySQL value
+
+Replicator can cast the value from MySQL to the required type 
+if your Tarantool schema does not comply with the MySQL schema.
+For example, MySQL column stores `bigint(20)` values, but Tarantool
+expects `unsigned`. 
+Without explicit casting you get an error, e.g.:
+> Tuple field 1 type does not match one required by operation: expected unsigned
+
+Supported types to cast to:
+* `unsigned`: try to cast any number to unsigned value.
+
+To enable this feature specify which column should be casted:
+
+```yaml
+...
+  mappings:
+    - source:
+        schema: 'city'
+        table: 'users'
+        columns:
+          - client_id
+      dest:
+        space: 'users'
+        cast:
+          client_id: 'unsigned'
+```
+
 ## Docker image
 
 Image available at [Docker Hub](https://hub.docker.com/r/pparshin/go-mysql-tarantool).

docker/mysql/init.d/init.sql

@@ -14,7 +14,7 @@ CREATE TABLE logins
     username  varchar(16)  not null,
     ip        varchar(16)  not null,
     date      int unsigned not null,
-    attempts  int unsigned   default 0,
+    attempts  bigint(20)     default 0,
     longitude float unsigned default 0,
     latitude  float unsigned default 0,
 

internal/bridge/attribute.go

@@ -27,12 +27,29 @@ const (
 	typePoint                         // coordinates
 )
 
+type castType int
+
+const (
+	castNone     castType = iota // do not cast
+	castUnsigned                 // unsigned
+)
+
+func castTypeFromString(str string) castType {
+	switch str {
+	case "unsigned":
+		return castUnsigned
+	default:
+		return castNone
+	}
+}
+
 // attribute represents MySQL column mapped to Tarantool.
 type attribute struct {
 	colIndex uint64   // column sequence number in MySQL table
 	tupIndex uint64   // attribute sequence number in Tarantool tuple
 	name     string   // unique attribute name
-	vtype    attrType // value type stored in the column
+	vType    attrType // value type stored in the column
+	cType    castType // value must be casted to this type
 	unsigned bool     // whether attribute contains unsigned number or not
 }
 
@@ -48,7 +65,8 @@ func newAttr(table *schema.Table, tupIndex uint64, name string) (*attribute, err
 		colIndex: uint64(idx),
 		tupIndex: tupIndex,
 		name:     col.Name,
-		vtype:    attrType(col.Type),
+		vType:    attrType(col.Type),
+		cType:    castNone,
 		unsigned: col.IsUnsigned,
 	}, nil
 }
@@ -61,13 +79,19 @@ func newAttrsFromPKs(table *schema.Table) []*attribute {
 			colIndex: uint64(pki),
 			tupIndex: uint64(i),
 			name:     col.Name,
+			vType:    attrType(col.Type),
+			cType:    castNone,
 			unsigned: col.IsUnsigned,
 		})
 	}
 
 	return pks
 }
 
+func (a *attribute) castTo(t castType) {
+	a.cType = t
+}
+
 func (a *attribute) fetchValue(row []interface{}) (interface{}, error) {
 	if a.colIndex >= uint64(len(row)) {
 		return nil, fmt.Errorf("column index (%d) equals or greater than row length (%d)", a.colIndex, len(row))
@@ -76,7 +100,12 @@ func (a *attribute) fetchValue(row []interface{}) (interface{}, error) {
 	value := row[a.colIndex]
 
 	if a.shouldCastToUInt64(value) {
-		return toUint64(value)
+		v, err := toUint64(value)
+		if err != nil {
+			return nil, err
+		}
+
+		return v, nil
 	}
 
 	return value, nil
@@ -87,11 +116,15 @@ func (a *attribute) shouldCastToUInt64(value interface{}) bool {
 		return false
 	}
 
+	if a.cType == castUnsigned {
+		return true
+	}
+
 	if !a.unsigned {
 		return false
 	}
 
-	return a.vtype == typeNumber || a.vtype == typeMediumInt
+	return a.vType == typeNumber || a.vType == typeMediumInt
 }
 
 func toUint64(i interface{}) (uint64, error) {

internal/bridge/attribute_test.go

@@ -11,7 +11,8 @@ func Test_attribute_fetchValue(t *testing.T) {
 		colIndex uint64
 		tupIndex uint64
 		name     string
-		vtype    attrType
+		vType    attrType
+		cType    castType
 		unsigned bool
 	}
 	type args struct {
@@ -30,7 +31,8 @@ func Test_attribute_fetchValue(t *testing.T) {
 				colIndex: 0,
 				tupIndex: 0,
 				name:     "name",
-				vtype:    typeString,
+				vType:    typeString,
+				cType:    castNone,
 				unsigned: false,
 			},
 			args: args{
@@ -45,7 +47,8 @@ func Test_attribute_fetchValue(t *testing.T) {
 				colIndex: 0,
 				tupIndex: 0,
 				name:     "speed",
-				vtype:    typeNumber,
+				vType:    typeNumber,
+				cType:    castNone,
 				unsigned: false,
 			},
 			args: args{
@@ -54,13 +57,46 @@ func Test_attribute_fetchValue(t *testing.T) {
 			want:    -20,
 			wantErr: false,
 		},
+		{
+			name: "Number_ForceCastToUnsigned",
+			fields: fields{
+				colIndex: 0,
+				tupIndex: 0,
+				name:     "speed",
+				vType:    typeNumber,
+				cType:    castUnsigned,
+				unsigned: false,
+			},
+			args: args{
+				row: []interface{}{20},
+			},
+			want:    uint64(20),
+			wantErr: false,
+		},
+		{
+			name: "Float_ForceCastToUnsigned_Error",
+			fields: fields{
+				colIndex: 0,
+				tupIndex: 0,
+				name:     "speed",
+				vType:    typeFloat,
+				cType:    castUnsigned,
+				unsigned: false,
+			},
+			args: args{
+				row: []interface{}{4654.123},
+			},
+			want:    0,
+			wantErr: true,
+		},
 		{
 			name: "UnsignedMediumInt",
 			fields: fields{
 				colIndex: 0,
 				tupIndex: 0,
 				name:     "id",
-				vtype:    typeMediumInt,
+				vType:    typeMediumInt,
+				cType:    castNone,
 				unsigned: true,
 			},
 			args: args{
@@ -75,7 +111,8 @@ func Test_attribute_fetchValue(t *testing.T) {
 				colIndex: 0,
 				tupIndex: 0,
 				name:     "id",
-				vtype:    typeNumber,
+				vType:    typeNumber,
+				cType:    castNone,
 				unsigned: true,
 			},
 			args: args{
@@ -90,7 +127,8 @@ func Test_attribute_fetchValue(t *testing.T) {
 				colIndex: 1,
 				tupIndex: 1,
 				name:     "name",
-				vtype:    typeString,
+				vType:    typeString,
+				cType:    castNone,
 				unsigned: false,
 			},
 			args: args{
@@ -105,7 +143,8 @@ func Test_attribute_fetchValue(t *testing.T) {
 				colIndex: 5,
 				tupIndex: 5,
 				name:     "name",
-				vtype:    typeString,
+				vType:    typeString,
+				cType:    castNone,
 				unsigned: false,
 			},
 			args: args{
@@ -123,7 +162,8 @@ func Test_attribute_fetchValue(t *testing.T) {
 				colIndex: tt.fields.colIndex,
 				tupIndex: tt.fields.tupIndex,
 				name:     tt.fields.name,
-				vtype:    tt.fields.vtype,
+				vType:    tt.fields.vType,
+				cType:    tt.fields.cType,
 				unsigned: tt.fields.unsigned,
 			}
 			got, err := a.fetchValue(tt.args.row)

internal/bridge/replicator.go

@@ -97,6 +97,7 @@ func (b *Bridge) newRules(cfg *config.Config) error {
 	rules := make(map[string]*rule, len(cfg.Replication.Mappings))
 	for _, mapping := range cfg.Replication.Mappings {
 		source := mapping.Source
+		cast := mapping.Dest.Cast
 
 		tableInfo, err := b.canal.GetTable(source.Schema, source.Table)
 		if err != nil {
@@ -107,6 +108,10 @@ func (b *Bridge) newRules(cfg *config.Config) error {
 		if len(pks) == 0 {
 			return fmt.Errorf("no primary keys found, schema: %s, table: %s", source.Schema, source.Table)
 		}
+		for _, pk := range pks {
+			typ := castTypeFromString(cast[pk.name])
+			pk.castTo(typ)
+		}
 
 		attrs := make([]*attribute, 0, len(source.Columns))
 		for i, name := range source.Columns {
@@ -125,6 +130,9 @@ func (b *Bridge) newRules(cfg *config.Config) error {
 				if err != nil {
 					return err
 				}
+				typ := castTypeFromString(cast[name])
+				attr.castTo(typ)
+
 				attrs = append(attrs, attr)
 			}
 		}

internal/bridge/replicator_test.go

@@ -6,6 +6,7 @@ import (
 	"os"
 	"path"
 	"path/filepath"
+	"sync"
 	"testing"
 	"time"
 
@@ -291,6 +292,57 @@ func (s *bridgeSuite) TestUpdatePrimaryKeys() {
 	}
 }
 
+func (s *bridgeSuite) TestForceCast() {
+	t := s.T()
+
+	for step := 1; step <= 2; step++ {
+		wantErr := step == 2
+
+		cfg := *s.cfg
+		if wantErr {
+			for i := range cfg.Replication.Mappings {
+				cfg.Replication.Mappings[i].Dest.Cast = nil
+			}
+		}
+
+		s.init(&cfg)
+
+		var wg sync.WaitGroup
+		wg.Add(1)
+
+		go func() {
+			defer wg.Done()
+
+			errors := s.bridge.Run()
+			if wantErr {
+				assert.Len(t, errors, 1)
+			} else {
+				for err := range errors {
+					assert.NoError(t, err)
+				}
+			}
+		}()
+
+		name := fmt.Sprintf("alice_%d", step)
+		_, err := s.executeSQL("INSERT INTO city.logins (username, ip, date, attempts, longitude, latitude) VALUES (?, ?, ?, ?, ?, ?)", name, "192.168.1.1", 1604571708, 404, 73.98, 40.74)
+		require.NoError(t, err)
+
+		if wantErr {
+			wg.Wait()
+		} else {
+			err = s.bridge.canal.CatchMasterPos(500 * time.Millisecond)
+			require.NoError(t, err)
+
+			require.Eventually(t, func() bool {
+				return s.hasSyncedData("logins", 1)
+			}, 500*time.Millisecond, 50*time.Millisecond)
+		}
+
+		err = s.bridge.Close()
+		assert.NoError(t, err)
+	}
+}
+
 func (s *bridgeSuite) TestReconnect() {
 	t := s.T()