From 66dc93dba23967d0b0b84f11eb193b9488693f93 Mon Sep 17 00:00:00 2001 From: okJiang <819421878@qq.com> Date: Fri, 31 Dec 2021 14:21:51 +0800 Subject: [PATCH 1/8] save work --- dm/pkg/parser/common.go | 1 + dm/tests/all_mode/run.sh | 4 +++- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/dm/pkg/parser/common.go b/dm/pkg/parser/common.go index a8f40a4c9b5..c4b33e4af24 100644 --- a/dm/pkg/parser/common.go +++ b/dm/pkg/parser/common.go @@ -318,6 +318,7 @@ func SplitDDL(stmt ast.StmtNode, schema string) (sqls []string, err error) { v.Table = table return sqls, nil + case *ast.CreateViewStmt: default: return nil, terror.ErrUnknownTypeDDL.Generate(stmt) } diff --git a/dm/tests/all_mode/run.sh b/dm/tests/all_mode/run.sh index 4eba97d890f..57b61b5f96b 100755 --- a/dm/tests/all_mode/run.sh +++ b/dm/tests/all_mode/run.sh @@ -380,6 +380,9 @@ function run() { # use sync_diff_inspector to check full dump loader check_sync_diff $WORK_DIR $cur/conf/diff_config.toml + # check create view success, then skip it + run_sql_source1 "create view all_mode.t1_v as select * from all_mode.t1 where id=0;" + run_sql_source1 "SHOW SLAVE HOSTS;" check_contains 'Slave_UUID' @@ -498,7 +501,6 @@ function run() { check_log_not_contains $WORK_DIR/worker2/log/dm-worker.log "Error .* Table .* doesn't exist" # test Db not exists should be reported - run_sql_tidb "drop database all_mode" run_sql_source1 "create table all_mode.db_error (c int primary key);" run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \ From 49bc586daf4d3f16807dc46324b7f5f2c2df8af3 Mon Sep 17 00:00:00 2001 From: okJiang <819421878@qq.com> Date: Fri, 31 Dec 2021 15:15:35 +0800 Subject: [PATCH 2/8] save work --- dm/syncer/syncer.go | 27 ++++++++++++++++++--------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go index 3dfe640503b..7d0fdf942bc 100644 --- a/dm/syncer/syncer.go +++ b/dm/syncer/syncer.go @@ -2159,26 +2159,35 @@ func (s *Syncer) handleQueryEvent(ev *replication.QueryEvent, ec eventContext, o appliedDDLs: make([]string, 0), sourceTbls: make(map[string]map[string]struct{}), } - qec.p, err = event.GetParserForStatusVars(ev.StatusVars) - if err != nil { - log.L().Warn("found error when get sql_mode from binlog status_vars", zap.Error(err)) - } - stmt, err := parseOneStmt(qec) - if err != nil { + defer func() { + if err == nil { + return + } // return error if parse fail and filter fail needSkip, err2 := s.skipSQLByPattern(qec.originSQL) if err2 != nil { - return err2 + err = err2 + return } if !needSkip { - return err + return } // don't return error if parse fail and filter success metrics.SkipBinlogDurationHistogram.WithLabelValues("query", s.cfg.Name, s.cfg.SourceID).Observe(time.Since(ec.startTime).Seconds()) ec.tctx.L().Warn("skip event", zap.String("event", "query"), zap.Stringer("query event context", qec)) *ec.lastLocation = *ec.currentLocation // before record skip location, update lastLocation - return s.recordSkipSQLsLocation(&ec) + err = s.recordSkipSQLsLocation(&ec) + }() + + qec.p, err = event.GetParserForStatusVars(ev.StatusVars) + if err != nil { + log.L().Warn("found error when get sql_mode from binlog status_vars", zap.Error(err)) + } + + stmt, err := parseOneStmt(qec) + if err != nil { + return err } if node, ok := stmt.(ast.DMLNode); ok { From b151cb4100c3162826875e043366cc95fa6c4acd Mon Sep 17 00:00:00 2001 From: okJiang <819421878@qq.com> Date: Fri, 31 Dec 2021 15:19:37 +0800 Subject: [PATCH 3/8] add comment --- dm/pkg/parser/common.go | 1 - dm/syncer/syncer.go | 4 ++-- dm/tests/all_mode/run.sh | 2 +- 3 files changed, 3 insertions(+), 4 deletions(-) diff --git a/dm/pkg/parser/common.go b/dm/pkg/parser/common.go index c4b33e4af24..a8f40a4c9b5 100644 --- a/dm/pkg/parser/common.go +++ b/dm/pkg/parser/common.go @@ -318,7 +318,6 @@ func SplitDDL(stmt ast.StmtNode, schema string) (sqls []string, err error) { v.Table = table return sqls, nil - case *ast.CreateViewStmt: default: return nil, terror.ErrUnknownTypeDDL.Generate(stmt) } diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go index 7d0fdf942bc..d6fc5781f89 100644 --- a/dm/syncer/syncer.go +++ b/dm/syncer/syncer.go @@ -2164,7 +2164,7 @@ func (s *Syncer) handleQueryEvent(ev *replication.QueryEvent, ec eventContext, o if err == nil { return } - // return error if parse fail and filter fail + // return error if filter fail needSkip, err2 := s.skipSQLByPattern(qec.originSQL) if err2 != nil { err = err2 @@ -2173,7 +2173,7 @@ func (s *Syncer) handleQueryEvent(ev *replication.QueryEvent, ec eventContext, o if !needSkip { return } - // don't return error if parse fail and filter success + // don't return error if filter success metrics.SkipBinlogDurationHistogram.WithLabelValues("query", s.cfg.Name, s.cfg.SourceID).Observe(time.Since(ec.startTime).Seconds()) ec.tctx.L().Warn("skip event", zap.String("event", "query"), zap.Stringer("query event context", qec)) *ec.lastLocation = *ec.currentLocation // before record skip location, update lastLocation diff --git a/dm/tests/all_mode/run.sh b/dm/tests/all_mode/run.sh index 57b61b5f96b..dbd6a985bb0 100755 --- a/dm/tests/all_mode/run.sh +++ b/dm/tests/all_mode/run.sh @@ -380,7 +380,7 @@ function run() { # use sync_diff_inspector to check full dump loader check_sync_diff $WORK_DIR $cur/conf/diff_config.toml - # check create view success, then skip it + # check create view(should be skipped) success, then skip it run_sql_source1 "create view all_mode.t1_v as select * from all_mode.t1 where id=0;" run_sql_source1 "SHOW SLAVE HOSTS;" From 4804ceaf56693f96086cc3e9f29e12ba72e0372e Mon Sep 17 00:00:00 2001 From: okJiang <819421878@qq.com> Date: Wed, 5 Jan 2022 09:58:54 +0800 Subject: [PATCH 4/8] Update dm/tests/all_mode/run.sh Co-authored-by: Ehco --- dm/tests/all_mode/run.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dm/tests/all_mode/run.sh b/dm/tests/all_mode/run.sh index dbd6a985bb0..858a680e805 100755 --- a/dm/tests/all_mode/run.sh +++ b/dm/tests/all_mode/run.sh @@ -380,7 +380,7 @@ function run() { # use sync_diff_inspector to check full dump loader check_sync_diff $WORK_DIR $cur/conf/diff_config.toml - # check create view(should be skipped) success, then skip it + # check create view(should be skipped by func `skipSQLByPattern`) will not stop sync task run_sql_source1 "create view all_mode.t1_v as select * from all_mode.t1 where id=0;" run_sql_source1 "SHOW SLAVE HOSTS;" From 94cb5da7ef125c79ed8a5f0ed18abe1e4e5c7716 Mon Sep 17 00:00:00 2001 From: okJiang <819421878@qq.com> Date: Wed, 5 Jan 2022 10:07:24 +0800 Subject: [PATCH 5/8] add comment --- dm/syncer/syncer.go | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go index d6fc5781f89..7ee3ef06ff8 100644 --- a/dm/syncer/syncer.go +++ b/dm/syncer/syncer.go @@ -2164,7 +2164,9 @@ func (s *Syncer) handleQueryEvent(ev *replication.QueryEvent, ec eventContext, o if err == nil { return } - // return error if filter fail + // why not `skipSQLByPattern` at beginning, but at defer? + // it is order to track every ddl except for the one that will cause error. + // if `skipSQLByPattern` at beginning, some ddl should be tracked may be skipped. needSkip, err2 := s.skipSQLByPattern(qec.originSQL) if err2 != nil { err = err2 From 84ec38791be53b5b3a68ba76dc5eab01d1e13acc Mon Sep 17 00:00:00 2001 From: okJiang <819421878@qq.com> Date: Wed, 5 Jan 2022 12:27:49 +0800 Subject: [PATCH 6/8] Update dm/syncer/syncer.go Co-authored-by: lance6716 --- dm/syncer/syncer.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go index 7ee3ef06ff8..fe463ea225b 100644 --- a/dm/syncer/syncer.go +++ b/dm/syncer/syncer.go @@ -2165,7 +2165,7 @@ func (s *Syncer) handleQueryEvent(ev *replication.QueryEvent, ec eventContext, o return } // why not `skipSQLByPattern` at beginning, but at defer? - // it is order to track every ddl except for the one that will cause error. + // it is in order to track every ddl except for the one that will cause error. // if `skipSQLByPattern` at beginning, some ddl should be tracked may be skipped. needSkip, err2 := s.skipSQLByPattern(qec.originSQL) if err2 != nil { From c13e595c96d5e59d79f61c4497a2592aeea7fbe8 Mon Sep 17 00:00:00 2001 From: okJiang <819421878@qq.com> Date: Wed, 5 Jan 2022 13:36:12 +0800 Subject: [PATCH 7/8] add check in IT --- dm/tests/all_mode/run.sh | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/dm/tests/all_mode/run.sh b/dm/tests/all_mode/run.sh index 858a680e805..08b72f21db8 100755 --- a/dm/tests/all_mode/run.sh +++ b/dm/tests/all_mode/run.sh @@ -323,11 +323,11 @@ function run() { run_sql_both_source "SET @@GLOBAL.SQL_MODE='ANSI_QUOTES,NO_AUTO_VALUE_ON_ZERO'" run_sql_source1 "SET @@global.time_zone = '+01:00';" run_sql_source2 "SET @@global.time_zone = '+02:00';" - test_expression_filter - test_fail_job_between_event - test_session_config - test_query_timeout - test_stop_task_before_checkpoint + # test_expression_filter + # test_fail_job_between_event + # test_session_config + # test_query_timeout + # test_stop_task_before_checkpoint inject_points=( "github.com/pingcap/tiflow/dm/dm/worker/TaskCheckInterval=return(\"500ms\")" @@ -382,6 +382,12 @@ function run() { # check create view(should be skipped by func `skipSQLByPattern`) will not stop sync task run_sql_source1 "create view all_mode.t1_v as select * from all_mode.t1 where id=0;" + sleep 1 + run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \ + "query-status -s $SOURCE_ID1" \ + "\"result\": true" 2 \ + "\"unit\": \"Sync\"" 1 \ + "\"stage\": \"Running\"" 2 run_sql_source1 "SHOW SLAVE HOSTS;" check_contains 'Slave_UUID' From 6c0daaae1a6eba3fffa94c21920c12485e09db96 Mon Sep 17 00:00:00 2001 From: okJiang <819421878@qq.com> Date: Wed, 5 Jan 2022 13:54:00 +0800 Subject: [PATCH 8/8] Update dm/tests/all_mode/run.sh --- dm/tests/all_mode/run.sh | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/dm/tests/all_mode/run.sh b/dm/tests/all_mode/run.sh index 08b72f21db8..c3d87bcf207 100755 --- a/dm/tests/all_mode/run.sh +++ b/dm/tests/all_mode/run.sh @@ -323,11 +323,11 @@ function run() { run_sql_both_source "SET @@GLOBAL.SQL_MODE='ANSI_QUOTES,NO_AUTO_VALUE_ON_ZERO'" run_sql_source1 "SET @@global.time_zone = '+01:00';" run_sql_source2 "SET @@global.time_zone = '+02:00';" - # test_expression_filter - # test_fail_job_between_event - # test_session_config - # test_query_timeout - # test_stop_task_before_checkpoint + test_expression_filter + test_fail_job_between_event + test_session_config + test_query_timeout + test_stop_task_before_checkpoint inject_points=( "github.com/pingcap/tiflow/dm/dm/worker/TaskCheckInterval=return(\"500ms\")"