feat(filter): Implement a Efficient DML/DDL Event Filtering Mechanism

cmd/filter-helper/main.go

@@ -17,11 +17,9 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/pingcap/ticdc/cmd/util"
+	"github.com/pingcap/ticdc/pkg/config"
+	"github.com/pingcap/ticdc/pkg/filter"
 	timodel "github.com/pingcap/tidb/pkg/meta/model"
-	"github.com/pingcap/tiflow/cdc/model"
-	"github.com/pingcap/tiflow/pkg/config"
-	"github.com/pingcap/tiflow/pkg/filter"
 	"github.com/spf13/cobra"
 )
 
@@ -49,13 +47,7 @@ func runFilter(cmd *cobra.Command, args []string) {
 	// fmt.Printf("Filter Rules: %v\n", filterRules)
 	// fmt.Printf("Schema Name: %s\n", schemaName)
 	// fmt.Printf("Table Name: %s\n", tableName)
-	cfg := &config.ReplicaConfig{}
-	err := util.StrictDecodeFile(cfgPath, "cdc filter helper", cfg)
-	if err != nil {
-		fmt.Printf("decode config file error: %v\n", err)
-		return
-	}
-	ft, err := filter.NewFilter(cfg, "")
+	ft, err := filter.NewFilter(config.NewDefaultFilterConfig(), "UTC", false, false)
 	if err != nil {
 		fmt.Printf("filter create error: %v\n", err)
 		return
@@ -73,30 +65,15 @@ func runFilter(cmd *cobra.Command, args []string) {
 
 	switch target {
 	case "table":
-		matched := !ft.ShouldIgnoreTable(tableAndSchema[0], tableAndSchema[1])
+		matched := !ft.ShouldIgnoreTable(tableAndSchema[0], tableAndSchema[1], nil)
 		if matched {
 			fmt.Printf("Table: %s, Matched filter rule\n", table)
 			return
 		}
 		fmt.Printf("Table: %s, Not matched filter rule\n", table)
 	case "ddl":
 		ddlType := timodel.ActionCreateTable
-		discard := ft.ShouldDiscardDDL(ddlType, tableAndSchema[0], tableAndSchema[1])
-		if discard {
-			fmt.Printf("DDL: %s, should be discard by event filter rule\n", ddl)
-			return
-		}
-		ignored, err := ft.ShouldIgnoreDDLEvent(&model.DDLEvent{
-			StartTs: uint64(0),
-			Query:   ddl,
-			Type:    ddlType,
-			TableInfo: &model.TableInfo{
-				TableName: model.TableName{
-					Schema: tableAndSchema[0],
-					Table:  tableAndSchema[1],
-				},
-			},
-		})
+		ignored, err := ft.ShouldIgnoreDDL(tableAndSchema[0], tableAndSchema[1], ddl, ddlType, nil)
 		if err != nil {
 			fmt.Printf("filter ddl error: %s, error: %v\n", ddl, err)
 			return

downstreamadapter/dispatcher/basic_dispatcher.go

@@ -201,6 +201,14 @@ func (d *BasicDispatcher) AddDMLEventsToSink(events []*commonEvent.DMLEvent) {
 }
 
 func (d *BasicDispatcher) AddBlockEventToSink(event commonEvent.BlockEvent) error {
+	if event.GetType() == commonEvent.TypeDDLEvent {
+		ddl := event.(*commonEvent.DDLEvent)
+		if ddl.NotSync {
+			log.Info("ignore DDL by NotSync", zap.Stringer("dispatcher", d.id), zap.Any("ddl", ddl))
+			d.PassBlockEventToSink(event)
+			return nil
+		}
+	}
 	d.tableProgress.Add(event)
 	return d.sink.WriteBlockEvent(event)
 }

logservice/schemastore/disk_format.go

@@ -794,6 +794,7 @@ func loadAllPhysicalTablesAtTs(
 			log.Panic("table info not found", zap.Int64("tableID", tableID))
 		}
 		if tableFilter != nil && tableFilter.ShouldIgnoreTable(schemaName, tableInfo.Name, fullTableInfo) {
+			log.Info("ignore table by filter", zap.String("schema", schemaName), zap.String("table", tableInfo.Name), zap.Any("tableInfo", fullTableInfo))
 			continue
 		}
 		splitable := isSplitable(fullTableInfo)

logservice/schemastore/persist_storage.go

@@ -411,7 +411,10 @@ func (p *persistentStorage) fetchTableDDLEvents(dispatcherID common.DispatcherID
 	events := make([]commonEvent.DDLEvent, 0, len(allTargetTs))
 	for _, ts := range allTargetTs {
 		rawEvent := readPersistedDDLEvent(storageSnap, ts)
-		ddlEvent, ok := buildDDLEvent(&rawEvent, tableFilter)
+		ddlEvent, ok, err := buildDDLEvent(&rawEvent, tableFilter)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
 		if ok {
 			events = append(events, ddlEvent)
 		} else {
@@ -478,7 +481,10 @@ func (p *persistentStorage) fetchTableTriggerDDLEvents(tableFilter filter.Filter
 		p.mu.RUnlock()
 		for _, ts := range allTargetTs {
 			rawEvent := readPersistedDDLEvent(storageSnap, ts)
-			ddlEvent, ok := buildDDLEvent(&rawEvent, tableFilter)
+			ddlEvent, ok, err := buildDDLEvent(&rawEvent, tableFilter)
+			if err != nil {
+				return nil, errors.Trace(err)
+			}
 			if ok {
 				events = append(events, ddlEvent)
 			}
@@ -818,7 +824,7 @@ func shouldSkipDDL(job *model.Job, tableMap map[int64]*BasicTableInfo) bool {
 	return false
 }
 
-func buildDDLEvent(rawEvent *PersistedDDLEvent, tableFilter filter.Filter) (commonEvent.DDLEvent, bool) {
+func buildDDLEvent(rawEvent *PersistedDDLEvent, tableFilter filter.Filter) (commonEvent.DDLEvent, bool, error) {
 	handler, ok := allDDLHandlers[model.ActionType(rawEvent.Type)]
 	if !ok {
 		log.Panic("unknown ddl type", zap.Any("ddlType", rawEvent.Type), zap.String("query", rawEvent.Query))