syncer(dm): save table checkpoint after a DDL is filtered

dm/syncer/checkpoint.go

@@ -128,6 +128,7 @@ func (b *binlogPoint) rollback(schemaTracker *schema.Tracker, schema string) (is
 	b.flushedPoint.location.ResetSuffix()
 	b.savedPoint.location = b.flushedPoint.location
 	if b.savedPoint.ti == nil {
+		// TODO: if we forget to save table info for table checkpoint, this is also nil!
 		return // for global checkpoint, no need to rollback the schema.
 	}
 

dm/syncer/filter.go

@@ -62,6 +62,7 @@ func (s *Syncer) skipQueryEvent(qec *queryEventContext, ddlInfo *ddlInfo) (bool,
 			if err != nil {
 				s.tctx.L().Warn("track ddl failed", zap.Stringer("ddl info", ddlInfo))
 			}
+			s.saveTablePoint(table, *qec.lastLocation)
 			s.tctx.L().Warn("track skipped ddl and return empty string", zap.String("origin sql", qec.originSQL), zap.Stringer("ddl info", ddlInfo))
 			ddlInfo.originDDL = ""
 			return true, nil

dm/syncer/filter_test.go

@@ -18,11 +18,13 @@ import (
 	"database/sql"
 
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-mysql-org/go-mysql/replication"
 	. "github.com/pingcap/check"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/parser"
 	"github.com/pingcap/tidb/util/filter"
+	"github.com/pingcap/tiflow/dm/pkg/binlog"
 
 	"github.com/pingcap/tiflow/dm/dm/config"
 	"github.com/pingcap/tiflow/dm/pkg/conn"
@@ -69,6 +71,7 @@ func (s *testFilterSuite) TestSkipQueryEvent(c *C) {
 	syncer.ddlDBConn = dbconn.NewDBConn(syncer.cfg, s.baseConn)
 	syncer.schemaTracker, err = schema.NewTracker(context.Background(), syncer.cfg.Name, defaultTestSessionCfg, syncer.ddlDBConn)
 	c.Assert(err, IsNil)
+	defer syncer.schemaTracker.Close()
 	syncer.exprFilterGroup = NewExprFilterGroup(utils.NewSessionCtx(nil), nil)
 
 	// test binlog filter
@@ -126,11 +129,16 @@ func (s *testFilterSuite) TestSkipQueryEvent(c *C) {
 	}
 	p := parser.New()
 
+	loc := binlog.NewLocation(mysql.MySQLFlavor)
+
 	for _, ca := range cases {
 		qec := &queryEventContext{
-			eventContext: &eventContext{tctx: tcontext.Background()},
-			p:            p,
-			ddlSchema:    ca.schema,
+			eventContext: &eventContext{
+				tctx:         tcontext.Background(),
+				lastLocation: &loc,
+			},
+			p:         p,
+			ddlSchema: ca.schema,
 		}
 		ddlInfo, err := syncer.genDDLInfo(qec, ca.sql)
 		c.Assert(err, IsNil)

dm/syncer/syncer.go

@@ -3312,6 +3312,9 @@ func (s *Syncer) loadTableStructureFromDump(ctx context.Context) error {
 					zap.Error(err))
 				setFirstErr(err)
 			}
+			// TODO: we should save table checkpoint here, but considering when
+			// the first time of flushing checkpoint, user may encounter https://github.com/pingcap/tiflow/issues/5010
+			// we should fix that problem first.
 		}
 	}
 	return firstErr

dm/tests/tracker_ignored_ddl/run.sh

@@ -27,19 +27,26 @@ function run() {
 	check_not_contains "ignore_1"
 	echo "increment1 check success"
 
+	# a not ignored DDL to trigger a checkpoint flush
+	run_sql_source1 "create table tracker_ignored_ddl.test (c int primary key);"
+
 	run_sql_file $cur/data/db.increment2.sql $MYSQL_HOST1 $MYSQL_PORT1 $MYSQL_PASSWORD1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
 		"Error 1054: Unknown column" 1
 
+	# force a resume, the error is still there, but we want to check https://github.com/pingcap/tiflow/issues/5272#issuecomment-1109283279
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"resume-task test"
+
 	# need operate tidb
 	run_sql_tidb "alter table $TEST_NAME.t1 add column ignore_1 int;"
 
 	# resume task
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"resume-task test" \
 		"\"result\": true" 2
-	sleep 1
+	sleep 3
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test" \
 		"\"stage\": \"Running\"" 2