Merge pull request #16 from larsnovikov/datetime_fields

larsnovikov · web-flow · commit e3259a2e5364 · 2019-04-29T19:14:59.000+02:00
Datetime fields
diff --git a/README.md b/README.md
@@ -29,7 +29,7 @@ Don't forget to set `binlog_do_db=<master_db_name>` and restart MySQL service.
 
 - Use `create-model <table>` to create json config for your table.
 - Create table on slave.
-- Use `build-table <table>` to copy table data from master and set start position of log for table listener.
+- Use `build-slave <table>` to copy table data from master and set start position of log for table listener.
 
 ### Custom handlers for field value
 
@@ -55,8 +55,8 @@ Don't forget to set `binlog_do_db=<master_db_name>` and restart MySQL service.
 - `set-position <table> <binlog_name> <binlog_position>` set start position of log for table listener
 - `load` start loader for replication testing (for default tables user and post)
 - `create-model <table>` create model json-file by master table structure
-- `build-table <table>` create master table dump, restore this dump in slave, set start position of log for table listener
-- `destroy-table <table>` truncate table, set empty position of log for table listener
+- `build-slave <table>` create master table dump, restore this dump in slave, set start position of log for table listener
+- `destroy-slave <table>` truncate table, set empty position of log for table listener
 
 ### Modes
 
diff --git a/examples/post.json b/examples/post.json
@@ -14,22 +14,38 @@
       {
         "name": "id",
         "key": true,
-        "mode": "int"
+        "mode": "int",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
       },
       {
         "name": "title",
         "key": false,
-        "mode": "string"
+        "mode": "string",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
       },
       {
         "name": "text",
         "key": false,
-        "mode": "string"
+        "mode": "string",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
       },
       {
         "name": "created",
         "key": false,
-        "mode": "timestamp"
+        "mode": "timestamp",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
       }
     ]
   }
diff --git a/examples/user.json b/examples/user.json
@@ -7,6 +7,9 @@
       "status",
       "active",
       "balance",
+      "time",
+      "date",
+      "datetime",
       "created"
     ]
   },
@@ -16,12 +19,20 @@
       {
         "name": "id",
         "key": true,
-        "mode": "int"
+        "mode": "int",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
       },
       {
         "name": "name",
         "key": false,
-        "mode": "string"
+        "mode": "string",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
       },
       {
         "name": "status",
@@ -37,17 +48,56 @@
       {
         "name": "active",
         "key": false,
-        "mode": "bool"
+        "mode": "bool",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
       },
       {
         "name": "balance",
         "key": false,
-        "mode": "float"
+        "mode": "float",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
+      },
+      {
+        "name": "time",
+        "key": false,
+        "mode": "time",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
+      },
+      {
+        "name": "date",
+        "key": false,
+        "mode": "date",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
+      },
+      {
+        "name": "datetime",
+        "key": false,
+        "mode": "datetime",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
       },
       {
         "name": "created",
         "key": false,
-        "mode": "timestamp"
+        "mode": "timestamp",
+        "beforeSave": {
+          "handler": "",
+          "params": null
+        }
       }
     ]
   }
diff --git a/sql/structure.sql b/sql/structure.sql
@@ -8,6 +8,9 @@ CREATE TABLE test.user
   status VARCHAR(255),
   active bool,
   balance float,
+  time time,
+  date date,
+  datetime datetime,
   created TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL ON UPDATE CURRENT_TIMESTAMP
 ) engine=InnoDB;
 
@@ -27,6 +30,9 @@ CREATE TABLE test.user
   status VARCHAR(255),
   active bool,
   balance float,
+  time time,
+  date date,
+  datetime datetime,
   created TIMESTAMP NOT NULL
 ) engine=InnoDB;
 
@@ -50,6 +56,9 @@ CREATE TABLE public."user"
   status VARCHAR(255),
   active bool,
   balance float,
+  time time,
+  date date,
+  datetime VARCHAR(255),
   created TIMESTAMP DEFAULT NULL
 );
 
@@ -64,11 +73,28 @@ CREATE TABLE public.post
 /*For ClickHouse Slave*/
 CREATE DATABASE test;
 
-CREATE TABLE test.user (id Int32, name FixedString(40), status FixedString(255), active UInt8, balance Float32, created DateTime) ENGINE = MergeTree()
+CREATE TABLE test.user
+(
+  id Int32,
+  name FixedString(40),
+  status FixedString(255),
+  active UInt8,
+  balance Float32,
+  time FixedString(255),
+  date FixedString(255),
+  datetime DateTime,
+  created DateTime
+) ENGINE = MergeTree()
 PARTITION BY id
 ORDER BY id;
 
-CREATE TABLE test.post (id Int32, title FixedString(40), text FixedString(255), created DateTime) ENGINE = MergeTree()
+CREATE TABLE test.post
+(
+  id Int32,
+  title FixedString(40),
+  text FixedString(255),
+  created DateTime
+) ENGINE = MergeTree()
 PARTITION BY id
 ORDER BY id;
 
@@ -81,6 +107,9 @@ CREATE TABLE "user"
   status VARCHAR,
   active bool,
   balance FLOAT,
+  time TIME,
+  date DATE,
+  datetime DATETIME,
   created TIMESTAMP NOT NULL
 );
 
diff --git a/src/constants/errors.go b/src/constants/errors.go
@@ -12,6 +12,7 @@ const (
 	ErrorModelFileExists     = "Model file \"%s\" already exists"
 	ErrorParserPosition      = "Catch error: \"%s\""
 	ErrorGetMinPosition      = "Can't get min position. Error: \"%s\""
+	ErrorEmptyPosition       = "Can't get position. Use \"destroy-slave %s\" and \"build-slave %s\""
 	ErrorUndefinedSlave      = "Can't get slave. Error: \"%s\""
 	ErrorCobraStarter        = "Catch cobra error: \"%s\""
 	ErrorCachePluginError    = "Catch plugin error: \"%s\""
@@ -20,5 +21,5 @@ const (
 	ErrorBuildModelConfig    = "Can't build model config. Error: \"%s\""
 	ErrorDumpRead            = "Can't read dump. Error: \"%s\""
 	ErrorParseLine           = "Can't parse line \"%s\". Error: \"%s\""
-	ErrorTableBuilt          = "Table \"%s\" already built. Use \"destroy-table %s\" before build"
+	ErrorSlaveBuilt          = "Slave table \"%s\" already built. Use \"destroy-slave %s\" before build"
 )
diff --git a/src/helpers/config.go b/src/helpers/config.go
@@ -83,8 +83,10 @@ func MakeCredentials() {
 		os.Getenv("REPLICATOR_DBNAME"),
 	}
 
-	for _, tableName := range strings.Split(os.Getenv("ALLOWED_TABLES"), ",") {
-		tables = append(tables, strings.TrimSpace(tableName))
+	if os.Getenv("ALLOWED_TABLES") != "" {
+		for _, tableName := range strings.Split(os.Getenv("ALLOWED_TABLES"), ",") {
+			tables = append(tables, strings.TrimSpace(tableName))
+		}
 	}
 
 	channelSize, _ = strconv.Atoi(os.Getenv("CHANNEL_SIZE"))
diff --git a/src/parser/binlog.go b/src/parser/binlog.go
@@ -101,7 +101,7 @@ func (h *binlogHandler) OnRow(e *canal.RowsEvent) error {
 				slave.GetSlaveByName(e.Table.Name).Delete(&header, positionSet)
 				SaveLocks[e.Table.Name] = false
 			} else {
-				log.Infof(constants.MessageIgnoreDelete, &header.Timestamp, e.Table.Name, &header.LogPos)
+				log.Infof(constants.MessageIgnoreDelete, header.Timestamp, e.Table.Name, header.LogPos)
 				return nil
 			}
 		}
@@ -123,15 +123,15 @@ func (h *binlogHandler) OnRow(e *canal.RowsEvent) error {
 				slave.GetSlaveByName(e.Table.Name).Update(&header, positionSet)
 				SaveLocks[e.Table.Name] = false
 			} else {
-				log.Infof(constants.MessageIgnoreUpdate, &header.Timestamp, e.Table.Name, &header.LogPos)
+				log.Infof(constants.MessageIgnoreUpdate, header.Timestamp, e.Table.Name, header.LogPos)
 				return nil
 			}
 		} else {
 			if SaveLocks[e.Table.Name] == false || canSave(getCalculatedPos(), e.Table.Name) {
 				slave.GetSlaveByName(e.Table.Name).Insert(&header, positionSet)
 				SaveLocks[e.Table.Name] = false
 			} else {
-				log.Infof(constants.MessageIgnoreInsert, &header.Timestamp, e.Table.Name, &header.LogPos)
+				log.Infof(constants.MessageIgnoreInsert, header.Timestamp, e.Table.Name, header.LogPos)
 				return nil
 			}
 		}
diff --git a/src/parser/parser.go b/src/parser/parser.go
@@ -118,5 +118,12 @@ func prepareType(fieldName string, fieldType string, value interface{}, params m
 	case "timestamp":
 		t, _ := time.Parse("2006-01-02 15:04:05", value.(string))
 		params[fieldName] = t
+	case "time":
+		params[fieldName] = value.(string)
+	case "date":
+		params[fieldName] = value.(string)
+	case "datetime":
+		t, _ := time.Parse("2006-01-02 15:04:05", value.(string))
+		params[fieldName] = t
 	}
 }
diff --git a/src/parser/position.go b/src/parser/position.go
@@ -29,7 +29,11 @@ func GetSavedPos(table string) mysql.Position {
 	hash := helpers.MakeHash(dbName, table)
 	pos, name := helpers.MakeTablePosKey(hash)
 
-	tablePosition, err := strconv.ParseUint(system.GetValue(pos), 10, 32)
+	position := system.GetValue(pos)
+	if position == "" {
+		log.Fatalf(constants.ErrorEmptyPosition, table, table)
+	}
+	tablePosition, err := strconv.ParseUint(position, 10, 32)
 	if err != nil {
 		log.Fatalf(constants.ErrorGetMinPosition, err)
 	}
diff --git a/src/tools/system/build-slave.go b/src/tools/system/build-slave.go
@@ -26,7 +26,7 @@ const (
 )
 
 var CmdBuildTable = &cobra.Command{
-	Use:   "build-table",
+	Use:   "build-slave",
 	Short: "Build slave table from master. Format: [table]",
 	Long:  "Build slave table from master. Format: [table]",
 	Args:  cobra.ExactArgs(1),
@@ -60,7 +60,7 @@ func canHandle() bool {
 		return true
 	}
 
-	log.Fatalf(constants.ErrorTableBuilt, helpers2.Table, helpers2.Table)
+	log.Fatalf(constants.ErrorSlaveBuilt, helpers2.Table, helpers2.Table)
 	return false
 }
 
diff --git a/src/tools/system/destroy-slave.go b/src/tools/system/destroy-slave.go
@@ -10,7 +10,7 @@ import (
 )
 
 var CmdDestroyTable = &cobra.Command{
-	Use:   "destroy-table",
+	Use:   "destroy-slave",
 	Short: "Destroy slave table from master. Format: [table]",
 	Long:  "Destroy slave table from master. Format: [table]",
 	Args:  cobra.ExactArgs(1),
diff --git a/src/tools/system/loader.go b/src/tools/system/loader.go
@@ -15,7 +15,7 @@ const (
 	// goroutine count. WARNING if you set more 1, may be concurrency problems
 	ThreadCount = 1
 	// time to create queries in minutes
-	LoadTime = 1
+	LoadTime = 10
 )
 
 var CmdLoad = &cobra.Command{
@@ -64,8 +64,8 @@ func randInt(min int, max int) int {
 
 func makeQueries(id int) {
 	queries := []string{
-		"INSERT INTO test.user (`name`, `status`, `active`, `balance`) VALUE ('Jack', 'active', false, 5.56);",
-		"UPDATE test.user SET `name`='Tommy', status='dead', active=true, balance=7.62 ORDER BY RAND() LIMIT 1",
+		"INSERT INTO test.user (`name`, `status`, `active`, `balance`, `time`, `date`, `datetime`) VALUE ('Jack', 'active', false, 5.56, '08:00:50', '2001-03-10', '2001-03-10 17:16:18');",
+		"UPDATE test.user SET `name`='Tommy', status='dead', active=true, balance=7.62, time='06:00:58', date='2010-03-10', datetime='2010-03-10 17:16:10' ORDER BY RAND() LIMIT 1",
 		"DELETE FROM test.user ORDER BY RAND() LIMIT 1;",
 		"INSERT INTO test.post (`title`, `text`) VALUE ('Title', 'London is the capital of Great Britain');",
 		"UPDATE test.post SET title='New title' ORDER BY RAND() LIMIT 1;",
diff --git a/src/tools/system/model_creator.go b/src/tools/system/model_creator.go
@@ -134,6 +134,12 @@ func getType(definition FieldDefinition) string {
 		return "float"
 	case "decimal":
 		return "float"
+	case "time":
+		return "time"
+	case "date":
+		return "date"
+	case "datetime":
+		return "datetime"
 	default:
 		log.Fatalf(constants.ErrorFieldTypeConversion, defType)
 	}

Original file line number	Diff line number	Diff line change
`@@ -83,8 +83,10 @@ func MakeCredentials() {`
`83`	`83`	`os.Getenv("REPLICATOR_DBNAME"),`
`84`	`84`	`}`
`85`	`85`
`86`		`- for _, tableName := range strings.Split(os.Getenv("ALLOWED_TABLES"), ",") {`
`87`		`- tables = append(tables, strings.TrimSpace(tableName))`
	`86`	`+ if os.Getenv("ALLOWED_TABLES") != "" {`
	`87`	`+ for _, tableName := range strings.Split(os.Getenv("ALLOWED_TABLES"), ",") {`
	`88`	`+ tables = append(tables, strings.TrimSpace(tableName))`
	`89`	`+ }`
`88`	`90`	`}`
`89`	`91`
`90`	`92`	`channelSize, _ = strconv.Atoi(os.Getenv("CHANNEL_SIZE"))`
Original file line number	Diff line number	Diff line change
`@@ -101,7 +101,7 @@ func (h binlogHandler) OnRow(e canal.RowsEvent) error {`
`101`	`101`	`slave.GetSlaveByName(e.Table.Name).Delete(&header, positionSet)`
`102`	`102`	`SaveLocks[e.Table.Name] = false`
`103`	`103`	`} else {`
`104`		`- log.Infof(constants.MessageIgnoreDelete, &header.Timestamp, e.Table.Name, &header.LogPos)`
	`104`	`+ log.Infof(constants.MessageIgnoreDelete, header.Timestamp, e.Table.Name, header.LogPos)`
`105`	`105`	`return nil`
`106`	`106`	`}`
`107`	`107`	`}`
`@@ -123,15 +123,15 @@ func (h binlogHandler) OnRow(e canal.RowsEvent) error {`
`123`	`123`	`slave.GetSlaveByName(e.Table.Name).Update(&header, positionSet)`
`124`	`124`	`SaveLocks[e.Table.Name] = false`
`125`	`125`	`} else {`
`126`		`- log.Infof(constants.MessageIgnoreUpdate, &header.Timestamp, e.Table.Name, &header.LogPos)`
	`126`	`+ log.Infof(constants.MessageIgnoreUpdate, header.Timestamp, e.Table.Name, header.LogPos)`
`127`	`127`	`return nil`
`128`	`128`	`}`
`129`	`129`	`} else {`
`130`	`130`	`if SaveLocks[e.Table.Name] == false \|\| canSave(getCalculatedPos(), e.Table.Name) {`
`131`	`131`	`slave.GetSlaveByName(e.Table.Name).Insert(&header, positionSet)`
`132`	`132`	`SaveLocks[e.Table.Name] = false`
`133`	`133`	`} else {`
`134`		`- log.Infof(constants.MessageIgnoreInsert, &header.Timestamp, e.Table.Name, &header.LogPos)`
	`134`	`+ log.Infof(constants.MessageIgnoreInsert, header.Timestamp, e.Table.Name, header.LogPos)`
`135`	`135`	`return nil`
`136`	`136`	`}`
`137`	`137`	`}`
Original file line number	Diff line number	Diff line change
`@@ -118,5 +118,12 @@ func prepareType(fieldName string, fieldType string, value interface{}, params m`
`118`	`118`	`case "timestamp":`
`119`	`119`	`t, _ := time.Parse("2006-01-02 15:04:05", value.(string))`
`120`	`120`	`params[fieldName] = t`
	`121`	`+ case "time":`
	`122`	`+ params[fieldName] = value.(string)`
	`123`	`+ case "date":`
	`124`	`+ params[fieldName] = value.(string)`
	`125`	`+ case "datetime":`
	`126`	`+ t, _ := time.Parse("2006-01-02 15:04:05", value.(string))`
	`127`	`+ params[fieldName] = t`
`121`	`128`	`}`
`122`	`129`	`}`
Original file line number	Diff line number	Diff line change
`@@ -26,7 +26,7 @@ const (`
`26`	`26`	`)`
`27`	`27`
`28`	`28`	`var CmdBuildTable = &cobra.Command{`
`29`		`- Use: "build-table",`
	`29`	`+ Use: "build-slave",`
`30`	`30`	`Short: "Build slave table from master. Format: [table]",`
`31`	`31`	`Long: "Build slave table from master. Format: [table]",`
`32`	`32`	`Args: cobra.ExactArgs(1),`
`@@ -60,7 +60,7 @@ func canHandle() bool {`
`60`	`60`	`return true`
`61`	`61`	`}`
`62`	`62`
`63`		`- log.Fatalf(constants.ErrorTableBuilt, helpers2.Table, helpers2.Table)`
	`63`	`+ log.Fatalf(constants.ErrorSlaveBuilt, helpers2.Table, helpers2.Table)`
`64`	`64`	`return false`
`65`	`65`	`}`
`66`	`66`