filter(ticdc): Move filter from sink to mounter and owner/schema.

cdc/entry/mounter.go

@@ -32,6 +32,7 @@ import (
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tiflow/cdc/model"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
+	pfilter "github.com/pingcap/tiflow/pkg/filter"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 )
@@ -66,15 +67,11 @@ type Mounter interface {
 }
 
 type mounterImpl struct {
-	schemaStorage  SchemaStorage
-	tz             *time.Location
-	workerNum      int
-	enableOldValue bool
-	changefeedID   model.ChangeFeedID
-
-	// index is an atomic variable to dispatch input events to workers.
-	index int64
-
+	schemaStorage       SchemaStorage
+	tz                  *time.Location
+	enableOldValue      bool
+	changefeedID        model.ChangeFeedID
+	filter              *pfilter.Filter
 	metricMountDuration prometheus.Observer
 	metricTotalRows     prometheus.Gauge
 }
@@ -83,12 +80,14 @@ type mounterImpl struct {
 func NewMounter(schemaStorage SchemaStorage,
 	changefeedID model.ChangeFeedID,
 	tz *time.Location,
+	filter *pfilter.Filter,
 	enableOldValue bool,
 ) Mounter {
 	return &mounterImpl{
 		schemaStorage:  schemaStorage,
 		changefeedID:   changefeedID,
 		enableOldValue: enableOldValue,
+		filter:         filter,
 		metricMountDuration: mountDuration.
 			WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 		metricTotalRows: totalRowsCountGauge.
@@ -112,6 +111,10 @@ func (m *mounterImpl) DecodeEvent(ctx context.Context, pEvent *model.Polymorphic
 	pEvent.Row = rowEvent
 	pEvent.RawKV.Value = nil
 	pEvent.RawKV.OldValue = nil
+	pEvent.Row.Filtered = m.filter.ShouldIgnoreDMLEvent(
+		pEvent.Row.StartTs,
+		pEvent.Row.Table.Schema,
+		pEvent.Row.Table.Table)
 	duration := time.Since(start)
 	if duration > time.Second {
 		m.metricMountDuration.Observe(duration.Seconds())

cdc/entry/mounter_test.go

@@ -29,6 +29,8 @@ import (
 	"github.com/pingcap/tidb/testkit"
 	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/pkg/config"
+	pfilter "github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/regionspan"
 	"github.com/stretchr/testify/require"
 	"github.com/tikv/client-go/v2/oracle"
@@ -289,9 +291,12 @@ func testMounterDisableOldValue(t *testing.T, tc struct {
 	ver, err := store.CurrentVersion(oracle.GlobalTxnScope)
 	require.Nil(t, err)
 	scheamStorage.AdvanceResolvedTs(ver.Ver)
+	config := config.GetDefaultReplicaConfig()
+	filter, err := pfilter.NewFilter(config)
+	require.Nil(t, err)
 	mounter := NewMounter(scheamStorage,
 		model.DefaultChangeFeedID("c1"),
-		time.UTC, false).(*mounterImpl)
+		time.UTC, filter, false).(*mounterImpl)
 	mounter.tz = time.Local
 	ctx := context.Background()
 

cdc/model/sink.go

@@ -270,6 +270,8 @@ type RowChangedEvent struct {
 
 	// SplitTxn marks this RowChangedEvent as the first line of a new txn.
 	SplitTxn bool `json:"-" msg:"-"`
+	// Filtered marks this RowChangedEvent was filtered by filter in mounter.
+	Filtered bool `json:"-" msg:"-"`
 }
 
 // IsDelete returns true if the row is a delete event

cdc/owner/changefeed.go

@@ -31,6 +31,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/scheduler"
 	cdcContext "github.com/pingcap/tiflow/pkg/context"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
+	pfilter "github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/orchestrator"
 	"github.com/pingcap/tiflow/pkg/txnutil/gc"
 	"github.com/pingcap/tiflow/pkg/upstream"
@@ -73,6 +74,7 @@ type changefeed struct {
 	schema      *schemaWrap4Owner
 	sink        DDLSink
 	ddlPuller   DDLPuller
+	filter      *pfilter.Filter
 	initialized bool
 	// isRemoved is true if the changefeed is removed
 	isRemoved bool
@@ -351,6 +353,10 @@ LOOP:
 	if err != nil {
 		return errors.Trace(err)
 	}
+	c.filter, err = pfilter.NewFilter(c.state.Info.Config)
+	if err != nil {
+		return errors.Trace(err)
+	}
 	cancelCtx, cancel := cdcContext.WithCancel(ctx)
 	c.cancel = cancel
 
@@ -557,6 +563,11 @@ func (c *changefeed) handleBarrier(ctx cdcContext.Context) (uint64, error) {
 	return barrierTs, nil
 }
 
+// asyncExecDDLJob execute ddl job asynchronously, it returns true if the jod is done.
+// 0. build ddl events from job.
+// 1. Apply ddl job to c.schema.
+// 2. Emit ddl event to redo manager.
+// 3. emit ddl event to ddl sink.
 func (c *changefeed) asyncExecDDLJob(ctx cdcContext.Context,
 	job *timodel.Job,
 ) (bool, error) {
@@ -567,6 +578,7 @@ func (c *changefeed) asyncExecDDLJob(ctx cdcContext.Context,
 	}
 
 	if c.ddlEventCache == nil {
+		// We must build ddl events from job before we call c.schema.HandleDDL(job).
 		ddlEvents, err := c.schema.BuildDDLEvents(job)
 		if err != nil {
 			log.Error("build DDL event fail", zap.String("changefeed", c.id.ID),
@@ -580,13 +592,14 @@ func (c *changefeed) asyncExecDDLJob(ctx cdcContext.Context,
 		checkpointTs := c.state.Info.GetCheckpointTs(c.state.Status)
 		// refresh checkpointTs and currentTableNames when a ddl job is received
 		c.sink.emitCheckpointTs(checkpointTs, c.currentTableNames)
+		// we apply ddl to update changefeed schema here.
 		err = c.schema.HandleDDL(job)
 		if err != nil {
 			return false, errors.Trace(err)
 		}
-		c.ddlEventCache = ddlEvents
-		for _, ddlEvent := range ddlEvents {
-			if c.redoManager.Enabled() {
+		c.ddlEventCache = c.filterDDLEvent(ddlEvents)
+		if c.redoManager.Enabled() {
+			for _, ddlEvent := range c.ddlEventCache {
 				err = c.redoManager.EmitDDLEvent(ctx, ddlEvent)
 				if err != nil {
 					return false, err
@@ -618,6 +631,9 @@ func (c *changefeed) asyncExecDDLEvent(ctx cdcContext.Context,
 	ddlEvent *model.DDLEvent,
 ) (done bool, err error) {
 	ddlEvent.Query, err = addSpecialComment(ddlEvent.Query)
+	if err != nil {
+		return false, err
+	}
 	if ddlEvent.TableInfo != nil &&
 		c.schema.IsIneligibleTableID(ddlEvent.TableInfo.TableID) {
 		log.Warn("ignore the DDL event of ineligible table",
@@ -632,6 +648,26 @@ func (c *changefeed) asyncExecDDLEvent(ctx cdcContext.Context,
 	return done, nil
 }
 
+func (c *changefeed) filterDDLEvent(ddlEvents []*model.DDLEvent) []*model.DDLEvent {
+	res := make([]*model.DDLEvent, 0)
+	// filter ddl event here
+	for _, ddlEvent := range ddlEvents {
+		if c.filter.ShouldIgnoreDDLEvent(ddlEvent.StartTs, ddlEvent.Type, ddlEvent.TableInfo.Schema, ddlEvent.TableInfo.Table) {
+			log.Info(
+				"DDL event ignored",
+				zap.String("query", ddlEvent.Query),
+				zap.Uint64("startTs", ddlEvent.StartTs),
+				zap.Uint64("commitTs", ddlEvent.CommitTs),
+				zap.String("namespace", c.id.Namespace),
+				zap.String("changefeed", c.id.ID),
+			)
+			continue
+		}
+		res = append(res, ddlEvent)
+	}
+	return res
+}
+
 func (c *changefeed) updateMetrics(currentTs int64, checkpointTs, resolvedTs model.Ts) {
 	phyCkpTs := oracle.ExtractPhysical(checkpointTs)
 	c.metricsChangefeedCheckpointTsGauge.Set(float64(phyCkpTs))

cdc/owner/changefeed_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/scheduler"
 	"github.com/pingcap/tiflow/pkg/config"
 	cdcContext "github.com/pingcap/tiflow/pkg/context"
+	pfilter "github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/orchestrator"
 	"github.com/pingcap/tiflow/pkg/txnutil/gc"
 	"github.com/pingcap/tiflow/pkg/upstream"
@@ -165,7 +166,7 @@ func (m *mockScheduler) Rebalance() {}
 // Close closes the scheduler and releases resources.
 func (m *mockScheduler) Close(ctx context.Context) {}
 
-func createChangefeed4Test(ctx cdcContext.Context, t *testing.T) (
+func createChangefeed4Test(ctx cdcContext.Context, t *testing.T, config *config.ReplicaConfig) (
 	*changefeed, *orchestrator.ChangefeedReactorState,
 	map[model.CaptureID]*model.CaptureInfo, *orchestrator.ReactorStateTester,
 ) {
@@ -189,6 +190,9 @@ func createChangefeed4Test(ctx cdcContext.Context, t *testing.T) (
 		return &mockScheduler{}, nil
 	}
 	cf.upstream = up
+	f, err := pfilter.NewFilter(config)
+	require.Nil(t, err)
+	cf.filter = f
 	state := orchestrator.NewChangefeedReactorState(ctx.ChangefeedVars().ID)
 	tester := orchestrator.NewReactorStateTester(t, state, nil)
 	state.PatchInfo(func(info *model.ChangeFeedInfo) (*model.ChangeFeedInfo, bool, error) {
@@ -204,7 +208,7 @@ func createChangefeed4Test(ctx cdcContext.Context, t *testing.T) (
 
 func TestPreCheck(t *testing.T) {
 	ctx := cdcContext.NewBackendContext4Test(true)
-	cf, state, captures, tester := createChangefeed4Test(ctx, t)
+	cf, state, captures, tester := createChangefeed4Test(ctx, t, config.GetDefaultReplicaConfig())
 	cf.Tick(ctx, state, captures)
 	tester.MustApplyPatches()
 	require.NotNil(t, state.Status)
@@ -224,7 +228,7 @@ func TestPreCheck(t *testing.T) {
 
 func TestInitialize(t *testing.T) {
 	ctx := cdcContext.NewBackendContext4Test(true)
-	cf, state, captures, tester := createChangefeed4Test(ctx, t)
+	cf, state, captures, tester := createChangefeed4Test(ctx, t, config.GetDefaultReplicaConfig())
 	defer cf.Close(ctx)
 	// pre check
 	cf.Tick(ctx, state, captures)
@@ -238,7 +242,7 @@ func TestInitialize(t *testing.T) {
 
 func TestChangefeedHandleError(t *testing.T) {
 	ctx := cdcContext.NewBackendContext4Test(true)
-	cf, state, captures, tester := createChangefeed4Test(ctx, t)
+	cf, state, captures, tester := createChangefeed4Test(ctx, t, config.GetDefaultReplicaConfig())
 	defer cf.Close(ctx)
 	// pre check
 	cf.Tick(ctx, state, captures)
@@ -280,7 +284,7 @@ func TestExecDDL(t *testing.T) {
 		},
 	})
 
-	cf, state, captures, tester := createChangefeed4Test(ctx, t)
+	cf, state, captures, tester := createChangefeed4Test(ctx, t, config.GetDefaultReplicaConfig())
 	cf.upstream.KVStorage = helper.Storage()
 	defer cf.Close(ctx)
 	tickThreeTime := func() {
@@ -370,7 +374,7 @@ func TestEmitCheckpointTs(t *testing.T) {
 		},
 	})
 
-	cf, state, captures, tester := createChangefeed4Test(ctx, t)
+	cf, state, captures, tester := createChangefeed4Test(ctx, t, config.GetDefaultReplicaConfig())
 	cf.upstream.KVStorage = helper.Storage()
 
 	defer cf.Close(ctx)
@@ -421,7 +425,7 @@ func TestSyncPoint(t *testing.T) {
 	ctx := cdcContext.NewBackendContext4Test(true)
 	ctx.ChangefeedVars().Info.SyncPointEnabled = true
 	ctx.ChangefeedVars().Info.SyncPointInterval = 1 * time.Second
-	cf, state, captures, tester := createChangefeed4Test(ctx, t)
+	cf, state, captures, tester := createChangefeed4Test(ctx, t, config.GetDefaultReplicaConfig())
 	defer cf.Close(ctx)
 
 	// pre check
@@ -451,7 +455,7 @@ func TestSyncPoint(t *testing.T) {
 func TestFinished(t *testing.T) {
 	ctx := cdcContext.NewBackendContext4Test(true)
 	ctx.ChangefeedVars().Info.TargetTs = ctx.ChangefeedVars().Info.StartTs + 1000
-	cf, state, captures, tester := createChangefeed4Test(ctx, t)
+	cf, state, captures, tester := createChangefeed4Test(ctx, t, config.GetDefaultReplicaConfig())
 	defer cf.Close(ctx)
 
 	// pre check
@@ -514,7 +518,7 @@ func testChangefeedReleaseResource(
 	redoLogDir string,
 	expectedInitialized bool,
 ) {
-	cf, state, captures, tester := createChangefeed4Test(ctx, t)
+	cf, state, captures, tester := createChangefeed4Test(ctx, t, config.GetDefaultReplicaConfig())
 
 	// pre check
 	cf.Tick(ctx, state, captures)
@@ -780,7 +784,7 @@ func TestExecRenameTablesDDL(t *testing.T) {
 	helper := entry.NewSchemaTestHelper(t)
 	defer helper.Close()
 	ctx := cdcContext.NewBackendContext4Test(true)
-	cf, state, captures, tester := createChangefeed4Test(ctx, t)
+	cf, state, captures, tester := createChangefeed4Test(ctx, t, config.GetDefaultReplicaConfig())
 	defer cf.Close(ctx)
 	// pre check
 	cf.Tick(ctx, state, captures)
@@ -869,7 +873,7 @@ func TestExecDropTablesDDL(t *testing.T) {
 	helper := entry.NewSchemaTestHelper(t)
 	defer helper.Close()
 	ctx := cdcContext.NewBackendContext4Test(true)
-	cf, state, captures, tester := createChangefeed4Test(ctx, t)
+	cf, state, captures, tester := createChangefeed4Test(ctx, t, config.GetDefaultReplicaConfig())
 	defer cf.Close(ctx)
 
 	// pre check
@@ -923,7 +927,7 @@ func TestExecDropViewsDDL(t *testing.T) {
 	helper := entry.NewSchemaTestHelper(t)
 	defer helper.Close()
 	ctx := cdcContext.NewBackendContext4Test(true)
-	cf, state, captures, tester := createChangefeed4Test(ctx, t)
+	cf, state, captures, tester := createChangefeed4Test(ctx, t, config.GetDefaultReplicaConfig())
 	defer cf.Close(ctx)
 
 	// pre check
@@ -979,3 +983,47 @@ func TestExecDropViewsDDL(t *testing.T) {
 	execDropStmt(jobs[0], "DROP VIEW `test1`.`view2`")
 	execDropStmt(jobs[1], "DROP VIEW `test1`.`view1`")
 }
+
+func TestFilterDDLEvent(t *testing.T) {
+	ddlEvents := []*model.DDLEvent{
+		{
+			StartTs:  1000,
+			CommitTs: 1010,
+			TableInfo: &model.SimpleTableInfo{
+				Schema: "test",
+				Table:  "t1",
+			},
+			Type:  timodel.ActionAddColumn,
+			Query: "ALTER TABLE test.t1 ADD COLUMN a int",
+		},
+		{
+			StartTs:  1020,
+			CommitTs: 1030,
+			TableInfo: &model.SimpleTableInfo{
+				Schema: "test",
+				Table:  "t2",
+			},
+			Type:  timodel.ActionAddColumn,
+			Query: "ALTER TABLE test.t2 ADD COLUMN a int",
+		},
+		{
+			StartTs:  1020,
+			CommitTs: 1030,
+			TableInfo: &model.SimpleTableInfo{
+				Schema: "test",
+				Table:  "t3",
+			},
+			Type:  timodel.ActionAddColumn,
+			Query: "ALTER TABLE test.t3 ADD COLUMN a int",
+		},
+	}
+	rc := config.GetDefaultReplicaConfig()
+	rc.Filter = &config.FilterConfig{
+		Rules: []string{"test.t1"},
+	}
+	ctx := cdcContext.NewBackendContext4Test(true)
+	cf, _, _, _ := createChangefeed4Test(ctx, t, rc)
+	defer cf.Close(ctx)
+	res := cf.filterDDLEvent(ddlEvents)
+	require.Equal(t, ddlEvents[:1], res)
+}

cdc/owner/ddl_sink.go

@@ -27,7 +27,6 @@ import (
 	"github.com/pingcap/tiflow/cdc/sink/mysql"
 	cdcContext "github.com/pingcap/tiflow/pkg/context"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
-	"github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/util"
 	"go.uber.org/zap"
 )
@@ -94,14 +93,9 @@ func newDDLSink() DDLSink {
 type ddlSinkInitHandler func(ctx cdcContext.Context, a *ddlSinkImpl, id model.ChangeFeedID, info *model.ChangeFeedInfo) error
 
 func ddlSinkInitializer(ctx cdcContext.Context, a *ddlSinkImpl, id model.ChangeFeedID, info *model.ChangeFeedInfo) error {
-	filter, err := filter.NewFilter(info.Config)
-	if err != nil {
-		return errors.Trace(err)
-	}
-
 	stdCtx := contextutil.PutChangefeedIDInCtx(ctx, id)
 	stdCtx = contextutil.PutRoleInCtx(stdCtx, util.RoleOwner)
-	s, err := sink.New(stdCtx, id, info.SinkURI, filter, info.Config, a.errCh)
+	s, err := sink.New(stdCtx, id, info.SinkURI, info.Config, a.errCh)
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -189,7 +183,7 @@ func (s *ddlSinkImpl) run(ctx cdcContext.Context, id model.ChangeFeedID, info *m
 				failpoint.Inject("InjectChangefeedDDLError", func() {
 					err = cerror.ErrExecDDLFailed.GenWithStackByArgs()
 				})
-				if err == nil || cerror.ErrDDLEventIgnored.Equal(errors.Cause(err)) {
+				if err == nil {
 					log.Info("Execute DDL succeeded",
 						zap.String("namespace", ctx.ChangefeedVars().ID.Namespace),
 						zap.String("changefeed", ctx.ChangefeedVars().ID.ID),