drainer: add config ignore-tables (#520)

cmd/drainer/drainer.toml

@@ -29,9 +29,6 @@ pd-urls = "http://127.0.0.1:2379"
 # If this is not setted, it will not set any sql-mode.
 # sql-mode = "STRICT_TRANS_TABLES,NO_ENGINE_SUBSTITUTION"
 
-# disable sync these schema
-ignore-schemas = "INFORMATION_SCHEMA,PERFORMANCE_SCHEMA,mysql"
-
 # number of binlog events in a transaction batch
 txn-batch = 20
 
@@ -49,10 +46,14 @@ safe-mode = false
 # valid values are "mysql", "pb", "tidb", "flash", "kafka"
 db-type = "mysql"
 
+# disable sync these schema
+ignore-schemas = "INFORMATION_SCHEMA,PERFORMANCE_SCHEMA,mysql"
+
 ##replicate-do-db priority over replicate-do-table if have same db name
 ##and we support regex expression , start with '~' declare use regex expression.
 #
 #replicate-do-db = ["~^b.*","s1"]
+
 #[[syncer.replicate-do-table]]
 #db-name ="test"
 #tbl-name = "log"
@@ -61,6 +62,11 @@ db-type = "mysql"
 #db-name ="test"
 #tbl-name = "~^a.*"
 
+# disable sync these table
+#[[syncer.ignore-table]]
+#db-name = "test"
+#tbl-name = "log"
+
 # the downstream mysql protocol database
 [syncer.to]
 host = "127.0.0.1"

drainer/config.go

@@ -47,6 +47,7 @@ type SyncerConfig struct {
 	StrSQLMode       *string            `toml:"sql-mode" json:"sql-mode"`
 	SQLMode          mysql.SQLMode      `toml:"-" json:"-"`
 	IgnoreSchemas    string             `toml:"ignore-schemas" json:"ignore-schemas"`
+	IgnoreTables     []filter.TableName `toml:"ignore-table" json:"ignore-table"`
 	TxnBatch         int                `toml:"txn-batch" json:"txn-batch"`
 	WorkerCount      int                `toml:"worker-count" json:"worker-count"`
 	To               *executor.DBConfig `toml:"to" json:"to"`

drainer/syncer.go

@@ -70,7 +70,7 @@ func NewSyncer(ctx context.Context, cp checkpoint.CheckPoint, cfg *SyncerConfig)
 	syncer.positions = make(map[string]int64)
 	syncer.causality = loader.NewCausality()
 	syncer.lastSyncTime = time.Now()
-	syncer.filter = filter.NewFilter(strings.Split(cfg.IgnoreSchemas, ","), nil, cfg.DoDBs, cfg.DoTables)
+	syncer.filter = filter.NewFilter(strings.Split(cfg.IgnoreSchemas, ","), cfg.IgnoreTables, cfg.DoDBs, cfg.DoTables)
 
 	return syncer, nil
 }

tests/filter/drainer.toml

@@ -16,9 +16,6 @@ pd-urls = "http://127.0.0.1:2379"
 # syncer Configuration.
 [syncer]
 
-# disable sync these schema
-ignore-schemas = "INFORMATION_SCHEMA,PERFORMANCE_SCHEMA,mysql"
-
 # number of binlog events in a transaction batch
 txn-batch = 1
 
@@ -34,18 +31,30 @@ safe-mode = false
 # valid values are "mysql", "pb", "tidb", "flash", "kafka"
 db-type = "mysql"
 
+# disable sync these schema
+ignore-schemas = "INFORMATION_SCHEMA,PERFORMANCE_SCHEMA,mysql"
+
 ##replicate-do-db priority over replicate-do-table if have same db name
 ##and we support regex expression , start with '~' declare use regex expression.
 #
 replicate-do-db = ["~^do_start.*","do_name"]
+
 [[syncer.replicate-do-table]]
 db-name ="test"
 tbl-name = "do_name"
 
+[[syncer.replicate-do-table]]
+db-name ="test"
+tbl-name = "do_ignore_name"
+
 [[syncer.replicate-do-table]]
 db-name ="test"
 tbl-name = "~^do_start.*"
 
+[[syncer.ignore-table]]
+db-name = "test"
+tbl-name = "do_ignore_name"
+
 # the downstream mysql protocol database
 [syncer.to]
 host = "127.0.0.1"

tests/filter/run.sh

@@ -29,11 +29,13 @@ run_sql "CREATE TABLE test.do_start1(id int);"
 run_sql "CREATE TABLE test.do_name(id int);"
 run_sql "CREATE TABLE test.do_not_start1(id int);"
 run_sql "CREATE TABLE test.do_not_name(id int);"
+run_sql "CREATE TABLE test.do_ignore_name(id int);"
 
 run_sql "INSERT INTO test.do_start1(id) VALUES (1);"
 run_sql "INSERT INTO test.do_name(id) VALUES (1);"
 run_sql "INSERT INTO test.do_not_start1(id) VALUES (1);"
 run_sql "INSERT INTO test.do_not_name(id) VALUES (1);"
+run_sql "INSERT INTO test.do_ignore_name(id) VALUES (1);"
 
 
 sleep 5
@@ -50,6 +52,7 @@ check_contains "Tables_in_test: do_start1"
 check_contains "Tables_in_test: do_name"
 check_not_contains "Tables_in_test: do_not_start1"
 check_not_contains "Tables_in_test: do_not_name"
+check_not_contains "Tables_in_test: do_ignore_name"
 
 # check DML
 down_run_sql "SELECT count(*) FROM test.do_start1;"