filter (ticdc): refine filter code

cdc/puller/ddl_puller.go

@@ -126,7 +126,8 @@ func (p *ddlJobPullerImpl) handleRawKVEntry(ctx context.Context, ddlRawKV *model
 			zap.String("namespace", p.changefeedID.Namespace),
 			zap.String("changefeed", p.changefeedID.ID),
 			zap.String("query", job.Query),
-			zap.Stringer("job", job), zap.Bool("skip", skip))
+			zap.Stringer("job", job), zap.Bool("skip", skip),
+			zap.Uint64("jobStartTs", job.StartTS))
 		if skip {
 			return nil
 		}

pkg/filter/expr_filter.go

@@ -407,6 +407,9 @@ func (f *dmlExprFilter) shouldSkipDML(
 	rawRow model.RowChangedDatums,
 	ti *model.TableInfo,
 ) (bool, error) {
+	if len(f.rules) == 0 {
+		return false, nil
+	}
 	// for defense purpose, normally the row and ti should not be nil.
 	if ti == nil || row == nil || rawRow.IsEmpty() {
 		return false, nil

pkg/filter/expr_filter_test.go

@@ -316,6 +316,34 @@ func TestShouldSkipDMLBasic(t *testing.T) {
 				},
 			},
 		},
+		{ // always return false when no rule configured
+			ddl: "create table test.season(id int primary key, name char(50), start char(100), end char(100))",
+			cfg: &config.FilterConfig{},
+			cases: []innerCase{
+				{ // do not ignore any event of test.season table
+					schema: "test",
+					table:  "season",
+					columns: []*model.Column{
+						{Name: "none"},
+					},
+					row:    []interface{}{1, "Spring", "January", "March"},
+					ignore: false,
+				},
+				{ // do not ignore any event of test.season table
+					schema: "test",
+					table:  "season",
+					preColumns: []*model.Column{
+						{Name: "none"},
+					},
+					preRow: []interface{}{2, "Summer", "April", "June"},
+					columns: []*model.Column{
+						{Name: "none"},
+					},
+					row:    []interface{}{2, "Summer", "April", "July"},
+					ignore: false,
+				},
+			},
+		},
 	}
 
 	sessCtx := utils.NewSessionCtx(map[string]string{

pkg/filter/sql_event_filter.go

@@ -148,6 +148,9 @@ func (f *sqlEventFilter) getRules(schema, table string) []*sqlEventRule {
 func (f *sqlEventFilter) shouldSkipDDL(
 	ddlType timodel.ActionType, schema, table, query string,
 ) (bool, error) {
+	if len(f.rules) == 0 {
+		return false, nil
+	}
 	log.Info("sql event filter handle ddl event",
 		zap.Any("ddlType", ddlType), zap.String("schema", schema),
 		zap.String("table", table), zap.String("query", query))

pkg/filter/sql_event_filter_test.go

@@ -168,6 +168,23 @@ func TestShouldSkipDDL(t *testing.T) {
 				},
 			},
 		},
+		{ // sqlEventFilterRule is nil, always return false
+			cfg: &config.FilterConfig{},
+			cases: []innerCase{
+				{
+					schema: "test",
+					table:  "t1",
+					query:  "ALTER TABLE t1 MODIFY COLUMN age int(11) NOT NULL",
+					skip:   false,
+				},
+				{
+					schema: "test",
+					table:  "t1",
+					query:  "ALTER TABLE t1 DROP COLUMN age",
+					skip:   false,
+				},
+			},
+		},
 	}
 
 	for _, tc := range testCases {

pkg/filter/utils.go

@@ -63,6 +63,22 @@ func ddlToEventType(p *parser.Parser, query string, jobType timodel.ActionType)
 		return bf.DropTablePartition, nil
 	case timodel.ActionTruncateTablePartition:
 		return bf.TruncateTablePartition, nil
+	case timodel.ActionCreateTable:
+		return bf.CreateTable, nil
+	case timodel.ActionDropTable:
+		return bf.DropTable, nil
+	case timodel.ActionAddIndex:
+		return bf.CreateIndex, nil
+	case timodel.ActionDropIndex:
+		return bf.DropIndex, nil
+	case timodel.ActionTruncateTable:
+		return bf.TruncateTable, nil
+	case timodel.ActionRenameTable, timodel.ActionRenameTables:
+		return bf.RenameTable, nil
+	case timodel.ActionCreateView:
+		return bf.CreateView, nil
+	case timodel.ActionDropView:
+		return bf.DropView, nil
 	}
 	stmt, err := p.ParseOneStmt(query, "", "")
 	if err != nil {