Add shield mysql ddl command

.gitignore

@@ -1 +1,4 @@
-bin
+bin
+.idea
+var
+go.sum

README.md

@@ -174,6 +174,22 @@ filter = ["id", "name"]
 
 In the above example, we will only sync MySQL table tfiler's columns `id` and `name` to Elasticsearch. 
 
+## Filter mysql ddl commands
+You can use `shield_ddl` to skip mysql DDL command
+
+```
+[[rule]]
+schema = "choujiang"
+table = "c_21"
+index = ""
+type = ""
+# shield ddl mysql command
+# example skip one command : delete
+# example skip many command : delete,update
+shield_ddl = "delete"
+```
+In the above example , we will skip mysql delete command
+
 ## Ignore table without a primary key
 When you sync table without a primary key, you can see below error message.
 ```

etc/river.toml

@@ -14,15 +14,15 @@ es_user = ""
 es_pass = ""
 
 # Path to store data, like master.info, if not set or empty,
-# we must use this to support breakpoint resume syncing. 
-# TODO: support other storage, like etcd. 
+# we must use this to support breakpoint resume syncing.
+# TODO: support other storage, like etcd.
 data_dir = "./var"
 
 # Inner Http status address
 stat_addr = "127.0.0.1:12800"
 stat_path = "/metrics"
 
-# pseudo server id like a slave 
+# pseudo server id like a slave
 server_id = 1001
 
 # mysql or mariadb
@@ -57,23 +57,24 @@ tables = ["t", "t_[0-9]{4}", "tfield", "tfilter"]
 # Below is for special rule mapping
 
 # Very simple example
-# 
+#
 # desc t;
 # +-------+--------------+------+-----+---------+-------+
 # | Field | Type         | Null | Key | Default | Extra |
 # +-------+--------------+------+-----+---------+-------+
 # | id    | int(11)      | NO   | PRI | NULL    |       |
 # | name  | varchar(256) | YES  |     | NULL    |       |
 # +-------+--------------+------+-----+---------+-------+
-# 
+#
 # The table `t` will be synced to ES index `test` and type `t`.
 [[rule]]
 schema = "test"
 table = "t"
 index = "test"
 type = "t"
+shield_ddl = ""
 
-# Wildcard table rule, the wildcard table must be in source tables 
+# Wildcard table rule, the wildcard table must be in source tables
 # All tables which match the wildcard format will be synced to ES index `test` and type `t`.
 # In this example, all tables must have same schema with above table `t`;
 [[rule]]
@@ -82,7 +83,7 @@ table = "t_[0-9]{4}"
 index = "test"
 type = "t"
 
-# Simple field rule 
+# Simple field rule
 #
 # desc tfield;
 # +----------+--------------+------+-----+---------+-------+
@@ -102,12 +103,12 @@ type = "tfield"
 [rule.field]
 # Map column `id` to ES field `es_id`
 id="es_id"
-# Map column `tags` to ES field `es_tags` with array type 
+# Map column `tags` to ES field `es_tags` with array type
 tags="es_tags,list"
 # Map column `keywords` to ES with array type
 keywords=",list"
 
-# Filter rule 
+# Filter rule
 #
 # desc tfilter;
 # +-------+--------------+------+-----+---------+-------+
@@ -145,5 +146,5 @@ table = "tid_[0-9]{4}"
 index = "test"
 type = "t"
 # The es doc's id will be `id`:`tag`
-# It is useful for merge muliple table into one type while theses tables have same PK 
+# It is useful for merge muliple table into one type while theses tables have same PK
 id = ["id", "tag"]

river/rule.go

@@ -32,6 +32,9 @@ type Rule struct {
 	// Elasticsearch pipeline
 	// To pre-process documents before indexing
 	Pipeline string `toml:"pipeline"`
+
+	// shield ddl command
+	ShieldDDL string `toml:"shield_ddl"`
 }
 
 func newDefaultRule(schema string, table string) *Rule {
@@ -83,3 +86,33 @@ func (r *Rule) CheckFilter(field string) bool {
 	}
 	return false
 }
+
+// CheckHasShieldCommands Check Has Shield commands
+func (r *Rule) CheckHasShieldCommands() bool {
+	return len(r.ShieldDDL) > 0
+}
+
+// CheckSkipShieldCommand check continue command
+func (r *Rule) CheckSkipShieldCommand(action string) bool {
+	if len(r.ShieldDDL) < 1 {
+		return false
+	}
+	// get command []string
+	shieldDDLs := r.getManyShieldCommands()
+	for _, v := range shieldDDLs {
+		// action == shieldDDL
+		if v == action {
+			return true
+		}
+	}
+	return false
+}
+
+func (r *Rule) getManyShieldCommands() []string {
+	// not have shield DDL command
+	if len(r.ShieldDDL) < 0 {
+		return []string{}
+	}
+	// command example delete,update
+	return strings.Split(r.ShieldDDL, ",")
+}

river/sync.go

@@ -70,6 +70,12 @@ func (h *eventHandler) OnRow(e *canal.RowsEvent) error {
 		return nil
 	}
 
+	// check continue command
+	if rule.CheckSkipShieldCommand(e.Action) {
+		log.Infof("rules have continue ddl command , command event is (%v) , rows (%v)", e.Action, e.Rows)
+		return nil
+	}
+
 	var reqs []*elastic.BulkRequest
 	var err error
 	switch e.Action {