Repository: pingcap/tiflow
Pull request #11094 (2024-05-14 08:57:28)

*: move binlog-filter package to tiflow
<!--
Thank you for contributing to TiFlow! 
Please read MD's [CONTRIBUTING](https://github.com/pingcap/tiflow/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: ref #5035

### What is changed and how it works?


### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test

#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new features need a release note -->

```release-note
 `None`.
```


ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please assign [hongyunyan](https://github.com/hongyunyan) for approval. For more information see [the Code Review Process](https://book.prow.tidb.net/#/workflows/pr).

The full list of commands accepted by this bot can be found [here](https://prow.tidb.net/command-help?repo=pingcap%2Ftiflow).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/pingcap/tiflow/blob/master/OWNERS)**
- **[cdc/api/v2/OWNERS](https://github.com/pingcap/tiflow/blob/master/cdc/api/v2/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["hongyunyan"]} -->
ti-chi-bot[bot]: @hawkingrei: adding LGTM is restricted to approvers and reviewers in OWNERS files.

<details>

In response to [this](https://github.com/pingcap/tiflow/pull/11094#pullrequestreview-2054737189):

>


Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pingcap/tiflow/pull/11094?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
Attention: Patch coverage is `65.56604%` with `73 lines` in your changes are missing coverage. Please review.
> Project coverage is 57.4918%. Comparing base [(`52b4301`)](https://app.codecov.io/gh/pingcap/tiflow/commit/52b4301388ae6859f082dd1c054c6a4da9d26152?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) to head [(`3225f1a`)](https://app.codecov.io/gh/pingcap/tiflow/pull/11094?dropdown=coverage&src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap).

<details><summary>Additional details and impacted files</summary>


| [Components](https://app.codecov.io/gh/pingcap/tiflow/pull/11094/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | Coverage Δ | |
|---|---|---|
| [cdc](https://app.codecov.io/gh/pingcap/tiflow/pull/11094/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `61.8568% <65.5660%> (-0.0234%)` | :arrow_down: |
| [dm](https://app.codecov.io/gh/pingcap/tiflow/pull/11094/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `51.2229% <ø> (-0.0324%)` | :arrow_down: |
| [engine](https://app.codecov.io/gh/pingcap/tiflow/pull/11094/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `63.3950% <ø> (-0.0495%)` | :arrow_down: |

| [Flag](https://app.codecov.io/gh/pingcap/tiflow/pull/11094/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | Coverage Δ | |
|---|---|---|
| [unit](https://app.codecov.io/gh/pingcap/tiflow/pull/11094/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `57.4918% <65.5660%> (-0.0223%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap#carryforward-flags-in-the-pull-request-comment) to find out more.

```diff
@@               Coverage Diff                @@
##             master     #11094        +/-   ##
================================================
- Coverage   57.5141%   57.4918%   -0.0223%     
================================================
  Files           851        853         +2     
  Lines        125524     125736       +212     
================================================
+ Hits          72194      72288        +94     
- Misses        47968      48060        +92     
- Partials       5362       5388        +26     
```

</details>
3AceShowHand: /retest

Patch:
diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go
index 75f1207ff..5bb6e3568 100644
--- a/cdc/api/v2/model.go
+++ b/cdc/api/v2/model.go
@@ -18,8 +18,8 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/cdc/model"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/integrity"
diff --git a/cmd/dm-syncer/config.go b/cmd/dm-syncer/config.go
index 35bd99771..020ee1ab9 100644
--- a/cmd/dm-syncer/config.go
+++ b/cmd/dm-syncer/config.go
@@ -21,12 +21,12 @@ import (
 	"github.com/BurntSushi/toml"
 	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/pingcap/errors"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	router "github.com/pingcap/tidb/pkg/util/table-router"
 	"github.com/pingcap/tiflow/dm/config"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
 	"github.com/pingcap/tiflow/dm/pkg/log"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/version"
 )
 
diff --git a/dm/config/source_config.go b/dm/config/source_config.go
index ff2ff0027..9c0ab5e7e 100644
--- a/dm/config/source_config.go
+++ b/dm/config/source_config.go
@@ -27,7 +27,6 @@ import (
 
 	"github.com/BurntSushi/toml"
 	"github.com/go-mysql-org/go-mysql/mysql"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
 	"github.com/pingcap/tiflow/dm/pkg/conn"
 	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
@@ -35,6 +34,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v2"
 )
diff --git a/dm/config/subtask.go b/dm/config/subtask.go
index e963a6be7..21fcd92a7 100644
--- a/dm/config/subtask.go
+++ b/dm/config/subtask.go
@@ -28,7 +28,6 @@ import (
 	"time"
 
 	"github.com/BurntSushi/toml"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb-tools/pkg/column-mapping"
 	extstorage "github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tidb/pkg/util/dbutil"
@@ -41,6 +40,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 	"github.com/pingcap/tiflow/engine/pkg/promutil"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/version"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
diff --git a/dm/config/task.go b/dm/config/task.go
index ae005c00f..1f916fe04 100644
--- a/dm/config/task.go
+++ b/dm/config/task.go
@@ -28,7 +28,6 @@ import (
 	"github.com/coreos/go-semver/semver"
 	"github.com/docker/go-units"
 	"github.com/dustin/go-humanize"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb-tools/pkg/column-mapping"
 	"github.com/pingcap/tidb/pkg/lightning/config"
 	"github.com/pingcap/tidb/pkg/parser"
@@ -38,6 +37,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v2"
 )
diff --git a/dm/config/task_converters.go b/dm/config/task_converters.go
index c9562c9e7..07f90d1af 100644
--- a/dm/config/task_converters.go
+++ b/dm/config/task_converters.go
@@ -17,7 +17,6 @@ import (
 	"fmt"
 	"strings"
 
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb-tools/pkg/column-mapping"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	router "github.com/pingcap/tidb/pkg/util/table-router"
@@ -27,6 +26,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/storage"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"go.uber.org/zap"
 )
 
diff --git a/dm/syncer/ddl.go b/dm/syncer/ddl.go
index 41b575d49..84ff95dcb 100644
--- a/dm/syncer/ddl.go
+++ b/dm/syncer/ddl.go
@@ -20,7 +20,6 @@ import (
 
 	"github.com/go-mysql-org/go-mysql/replication"
 	"github.com/pingcap/failpoint"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	tidbddl "github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/ast"
@@ -47,6 +46,7 @@ import (
 	onlineddl "github.com/pingcap/tiflow/dm/syncer/online-ddl-tools"
 	sm "github.com/pingcap/tiflow/dm/syncer/safe-mode"
 	"github.com/pingcap/tiflow/dm/syncer/shardddl"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
 )
diff --git a/dm/syncer/filter.go b/dm/syncer/filter.go
index 799398ca6..11863921d 100644
--- a/dm/syncer/filter.go
+++ b/dm/syncer/filter.go
@@ -15,11 +15,11 @@ package syncer
 
 import (
 	"github.com/go-mysql-org/go-mysql/replication"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 	onlineddl "github.com/pingcap/tiflow/dm/syncer/online-ddl-tools"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"go.uber.org/zap"
 )
 
diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go
index c044ca3c5..57622a69f 100644
--- a/dm/syncer/syncer.go
+++ b/dm/syncer/syncer.go
@@ -31,7 +31,6 @@ import (
 	mysql2 "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	tidbddl "github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/ast"
@@ -69,6 +68,7 @@ import (
 	sm "github.com/pingcap/tiflow/dm/syncer/safe-mode"
 	"github.com/pingcap/tiflow/dm/syncer/shardddl"
 	"github.com/pingcap/tiflow/dm/unit"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/errorutil"
 	"github.com/pingcap/tiflow/pkg/sqlmodel"
 	clientv3 "go.etcd.io/etcd/client/v3"
diff --git a/dm/worker/source_worker.go b/dm/worker/source_worker.go
index adc349ac3..e67e50d7e 100644
--- a/dm/worker/source_worker.go
+++ b/dm/worker/source_worker.go
@@ -23,7 +23,6 @@ import (
 	"github.com/golang/protobuf/proto"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/dm/config"
 	"github.com/pingcap/tiflow/dm/pb"
 	"github.com/pingcap/tiflow/dm/pkg/binlog"
@@ -35,6 +34,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/streamer"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/relay"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
diff --git a/dm/worker/subtask.go b/dm/worker/subtask.go
index 016709e27..8db5b050a 100644
--- a/dm/worker/subtask.go
+++ b/dm/worker/subtask.go
@@ -652,8 +652,15 @@ func (st *SubTask) Update(ctx context.Context, cfg *config.SubTaskConfig) error
 
 // OperateSchema operates schema for an upstream table.
 func (st *SubTask) OperateSchema(ctx context.Context, req *pb.OperateWorkerSchemaRequest) (schema string, err error) {
-	if st.Stage() != pb.Stage_Paused && req.Op != pb.SchemaOp_ListMigrateTargets {
-		return "", terror.ErrWorkerNotPausedStage.Generate(st.Stage().String())
+	switch req.Op {
+	case pb.SchemaOp_ListMigrateTargets:
+		if st.Stage() != pb.Stage_Running && st.Stage() != pb.Stage_Paused {
+			return "", terror.ErrWorkerNotPausedStage.Generate(st.Stage().String())
+		}
+	default:
+		if st.Stage() != pb.Stage_Paused {
+			return "", terror.ErrWorkerNotPausedStage.Generate(st.Stage().String())
+		}
 	}
 
 	syncUnit, ok := st.currUnit.(*syncer.Syncer)
diff --git a/engine/jobmaster/dm/config/config.go b/engine/jobmaster/dm/config/config.go
index 396db5945..3d38df3c1 100644
--- a/engine/jobmaster/dm/config/config.go
+++ b/engine/jobmaster/dm/config/config.go
@@ -20,7 +20,6 @@ import (
 
 	"github.com/dustin/go-humanize"
 	"github.com/google/uuid"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb-tools/pkg/column-mapping"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	router "github.com/pingcap/tidb/pkg/util/table-router"
@@ -28,6 +27,7 @@ import (
 	dmconfig "github.com/pingcap/tiflow/dm/config"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
 	"github.com/pingcap/tiflow/dm/master"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/errors"
 	"go.uber.org/atomic"
 	"gopkg.in/yaml.v2"
diff --git a/pkg/binlog-filter/filter.go b/pkg/binlog-filter/filter.go
new file mode 100644
index 000000000..5fa84cd4a
--- /dev/null
+++ b/pkg/binlog-filter/filter.go
@@ -0,0 +1,431 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package filter
+
+import (
+	"regexp"
+	"strings"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/log"
+	selector "github.com/pingcap/tidb/pkg/util/table-rule-selector"
+	"go.uber.org/zap"
+)
+
+// ActionType indicates how to handle matched items
+type ActionType string
+
+// show that how to handle rules
+const (
+	Ignore ActionType = "Ignore"
+	Do     ActionType = "Do"
+	Error  ActionType = "Error"
+)
+
+// EventType is DML/DDL Event type
+type EventType string
+
+// show DML/DDL Events
+const (
+	ddl             EventType = "ddl"
+	dml             EventType = "dml"
+	incompatibleDDL EventType = "incompatible DDL"
+
+	// it indicates all dml/ddl events in rule
+	AllEvent EventType = "all"
+	AllDDL   EventType = "all ddl"
+	AllDML   EventType = "all dml"
+
+	// it indicates no any dml/ddl events in rule,
+	// and equals empty rule.DDLEvent/DMLEvent
+	NoneEvent EventType = "none"
+	NoneDDL   EventType = "none ddl"
+	NoneDML   EventType = "none dml"
+
+	InsertEvent EventType = "insert"
+	UpdateEvent EventType = "update"
+	DeleteEvent EventType = "delete"
+
+	CreateDatabase EventType = "create database"
+	DropDatabase   EventType = "drop database"
+	AlterDatabase  EventType = "alter database"
+	CreateTable    EventType = "create table"
+	DropTable      EventType = "drop table"
+	TruncateTable  EventType = "truncate table"
+	RenameTable    EventType = "rename table"
+	CreateIndex    EventType = "create index"
+	DropIndex      EventType = "drop index"
+	CreateView     EventType = "create view"
+	DropView       EventType = "drop view"
+	AlterTable     EventType = "alter table"
+
+	CreateSchema EventType = "create schema" // alias of CreateDatabase
+	DropSchema   EventType = "drop schema"   // alias of DropDatabase
+	AlterSchema  EventType = "alter schema"  // alias of AlterDatabase
+
+	AddTablePartition      EventType = "add table partition"
+	DropTablePartition     EventType = "drop table partition"
+	TruncateTablePartition EventType = "truncate table partition"
+	// if need, add more	AlertTableOption     = "alert table option"
+
+	IncompatibleDDLChanges EventType = "incompatible ddl changes"
+	ValueRangeDecrease     EventType = "value range decrease"
+	PrecisionDecrease      EventType = "precision decrease"
+	ModifyColumn           EventType = "modify column"
+	RenameColumn           EventType = "rename column"
+	RenameIndex            EventType = "rename index"
+	DropColumn             EventType = "drop column"
+	DropPrimaryKey         EventType = "drop primary key"
+	DropUniqueKey          EventType = "drop unique key"
+	ModifyDefaultValue     EventType = "modify default value"
+	ModifyConstraint       EventType = "modify constraint"
+	ModifyColumnsOrder     EventType = "modify columns order"
+	ModifyCharset          EventType = "modify charset"
+	ModifyCollation        EventType = "modify collation"
+	RemoveAutoIncrement    EventType = "remove auto increment"
+	ModifyStorageEngine    EventType = "modify storage engine"
+	ReorganizePartition    EventType = "reorganize table partition"
+	RebuildPartition       EventType = "rebuild table partition"
+	CoalescePartition      EventType = "coalesce table partition"
+	SplitPartition         EventType = "split table partition"
+	ExchangePartition      EventType = "exchange table partition"
+
+	ModifySchemaCharsetAndCollate EventType = "modify schema charset and collate"
+	ModifyTableCharsetAndCollate  EventType = "modify table charset and collate"
+	ModifyTableComment            EventType = "modify table comment"
+	RecoverTable                  EventType = "recover table"
+	AlterTablePartitioning        EventType = "alter table partitioning"
+	RemovePartitioning            EventType = "remove table partitioning"
+	AddColumn                     EventType = "add column"
+	SetDefaultValue               EventType = "set default value"
+	RebaseAutoID                  EventType = "rebase auto id"
+	AddPrimaryKey                 EventType = "add primary key"
+	AlterIndexVisibility          EventType = "alter index visibility"
+	AlterTTLInfo                  EventType = "alter ttl info"
+	AlterTTLRemove                EventType = "alter ttl remove"
+	MultiSchemaChange             EventType = "multi schema change"
+
+	// NullEvent is used to represents unsupported ddl event type when we
+	// convert a ast.StmtNode or a string to EventType.
+	NullEvent EventType = ""
+)
+
+// ClassifyEvent classify event into dml/ddl
+func ClassifyEvent(event EventType) (EventType, error) {
+	switch event {
+	case InsertEvent,
+		UpdateEvent,
+		DeleteEvent:
+		return dml, nil
+	case CreateDatabase,
+		AlterDatabase,
+		AlterSchema,
+		CreateTable,
+		CreateIndex,
+		CreateView,
+		DropView,
+		AlterTable,
+		CreateSchema,
+		AddTablePartition:
+		return ddl, nil
+	case NullEvent:
+		return NullEvent, nil
+	case ValueRangeDecrease,
+		PrecisionDecrease,
+		ModifyColumn,
+		RenameColumn,
+		RenameIndex,
+		DropColumn,
+		DropPrimaryKey,
+		DropUniqueKey,
+		ModifyDefaultValue,
+		ModifyConstraint,
+		ModifyColumnsOrder,
+		ModifyCharset,
+		ModifyCollation,
+		RemoveAutoIncrement,
+		ModifyStorageEngine,
+		ReorganizePartition,
+		RebuildPartition,
+		CoalescePartition,
+		SplitPartition,
+		ExchangePartition,
+
+		DropDatabase,
+		DropTable,
+		DropIndex,
+		RenameTable,
+		TruncateTable,
+		DropSchema,
+		DropTablePartition,
+		TruncateTablePartition,
+
+		ModifySchemaCharsetAndCollate,
+		ModifyTableCharsetAndCollate,
+		ModifyTableComment,
+		RecoverTable,
+		AlterTablePartitioning,
+		RemovePartitioning,
+		AddColumn,
+		SetDefaultValue,
+		RebaseAutoID,
+		AddPrimaryKey,
+		AlterIndexVisibility,
+		AlterTTLInfo,
+		AlterTTLRemove,
+		MultiSchemaChange:
+		return incompatibleDDL, nil
+	default:
+		return NullEvent, errors.NotValidf("event type %s", event)
+	}
+}
+
+// BinlogEventRule is a rule to filter binlog events
+type BinlogEventRule struct {
+	SchemaPattern string      `json:"schema-pattern" toml:"schema-pattern" yaml:"schema-pattern"`
+	TablePattern  string      `json:"table-pattern" toml:"table-pattern" yaml:"table-pattern"`
+	Events        []EventType `json:"events" toml:"events" yaml:"events"`
+	SQLPattern    []string    `json:"sql-pattern" toml:"sql-pattern" yaml:"sql-pattern"` // regular expression
+	sqlRegularExp *regexp.Regexp
+
+	Action ActionType `json:"action" toml:"action" yaml:"action"`
+}
+
+// ToLower covert schema/table pattern to lower case
+func (b *BinlogEventRule) ToLower() {
+	b.SchemaPattern = strings.ToLower(b.SchemaPattern)
+	b.TablePattern = strings.ToLower(b.TablePattern)
+}
+
+// Valid checks validity of rule.
+func (b *BinlogEventRule) Valid() error {
+	if len(b.SQLPattern) > 0 {
+		reg, err := regexp.Compile("(?i)" + strings.Join(b.SQLPattern, "|"))
+		if err != nil {
+			return errors.Annotatef(err, "compile regular expression %+v", b.SQLPattern)
+		}
+		b.sqlRegularExp = reg
+	}
+
+	if b.Action != Do && b.Action != Ignore && b.Action != Error {
+		return errors.Errorf("action of binlog event rule %+v should not be empty", b)
+	}
+
+	for i := range b.Events {
+		et, err := toEventType(string(b.Events[i]))
+		if err != nil {
+			return errors.NotValidf("event type %s", b.Events[i])
+		}
+		b.Events[i] = et
+	}
+	return nil
+}
+
+// BinlogEvent filters binlog events by given rules
+type BinlogEvent struct {
+	selector.Selector
+
+	caseSensitive bool
+}
+
+// NewBinlogEvent returns a binlog event filter
+func NewBinlogEvent(caseSensitive bool, rules []*BinlogEventRule) (*BinlogEvent, error) {
+	b := &BinlogEvent{
+		Selector:      selector.NewTrieSelector(),
+		caseSensitive: caseSensitive,
+	}
+
+	for _, rule := range rules {
+		if err := b.AddRule(rule); err != nil {
+			log.Error("invalid binlog event rule", zap.Any("rule", rule), zap.Error(err))
+		}
+	}
+
+	return b, nil
+}
+
+// AddRule adds a rule into binlog event filter
+func (b *BinlogEvent) AddRule(rule *BinlogEventRule) error {
+	if b == nil || rule == nil {
+		return nil
+	}
+	err := rule.Valid()
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if !b.caseSensitive {
+		rule.ToLower()
+	}
+
+	err = b.Insert(rule.SchemaPattern, rule.TablePattern, rule, selector.Insert)
+	if err != nil {
+		return errors.Annotatef(err, "add rule %+v into binlog event filter", rule)
+	}
+
+	return nil
+}
+
+// UpdateRule updates binlog event filter rule
+func (b *BinlogEvent) UpdateRule(rule *BinlogEventRule) error {
+	if b == nil || rule == nil {
+		return nil
+	}
+	err := rule.Valid()
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if !b.caseSensitive {
+		rule.ToLower()
+	}
+
+	err = b.Insert(rule.SchemaPattern, rule.TablePattern, rule, selector.Replace)
+	if err != nil {
+		return errors.Annotatef(err, "update rule %+v into binlog event filter", rule)
+	}
+
+	return nil
+}
+
+// RemoveRule removes a rule from binlog event filter
+func (b *BinlogEvent) RemoveRule(rule *BinlogEventRule) error {
+	if b == nil || rule == nil {
+		return nil
+	}
+	if !b.caseSensitive {
+		rule.ToLower()
+	}
+
+	err := b.Remove(rule.SchemaPattern, rule.TablePattern)
+	if err != nil {
+		return errors.Annotatef(err, "remove rule %+v", rule)
+	}
+
+	return nil
+}
+
+// Filter filters events or queries by given rules
+// returns action and error
+func (b *BinlogEvent) Filter(schema, table string, event EventType, rawQuery string) (ActionType, error) {
+	if b == nil {
+		return Do, nil
+	}
+
+	tp, err := ClassifyEvent(event)
+	if err != nil {
+		return Ignore, errors.Trace(err)
+	}
+
+	schemaL, tableL := schema, table
+	if !b.caseSensitive {
+		schemaL, tableL = strings.ToLower(schema), strings.ToLower(table)
+	}
+
+	rules := b.Match(schemaL, tableL)
+	if len(rules) == 0 {
+		return Do, nil
+	}
+
+	for _, rule := range rules {
+		binlogEventRule, ok := rule.(*BinlogEventRule)
+		if !ok {
+			return "", errors.NotValidf("rule %+v", rule)
+		}
+
+		if tp != NullEvent {
+			matched := b.matchEvent(tp, event, binlogEventRule.Events)
+
+			if matched {
+				// ignore has highest priority
+				if binlogEventRule.Action == Ignore {
+					return Ignore, nil
+				}
+				if binlogEventRule.Action == Error {
+					return Error, nil
+				}
+			} else {
+				if binlogEventRule.Action == Do {
+					return Ignore, nil
+				}
+			}
+		}
+
+		if len(rawQuery) > 0 {
+			if len(binlogEventRule.SQLPattern) == 0 {
+				// sql pattern is disabled , just continue
+				continue
+			}
+
+			matched := binlogEventRule.sqlRegularExp.FindStringIndex(rawQuery) != nil
+			if matched {
+				// Ignore has highest priority
+				if binlogEventRule.Action == Ignore {
+					return Ignore, nil
+				}
+				if binlogEventRule.Action == Error {
+					return Error, nil
+				}
+			} else {
+				if binlogEventRule.Action == Do {
+					return Ignore, nil
+				}
+			}
+		}
+	}
+
+	return Do, nil
+}
+
+func (b *BinlogEvent) matchEvent(tp, event EventType, rules []EventType) bool {
+	for _, rule := range rules {
+		if rule == AllEvent {
+			return true
+		}
+
+		if rule == NoneEvent {
+			return false
+		}
+
+		if tp == ddl || tp == incompatibleDDL {
+			if rule == AllDDL {
+				return true
+			}
+
+			if rule == NoneDDL {
+				return false
+			}
+		}
+
+		if tp == dml {
+			if rule == AllDML {
+				return true
+			}
+
+			if rule == NoneDML {
+				return false
+			}
+		}
+
+		if tp == incompatibleDDL {
+			if rule == IncompatibleDDLChanges {
+				return true
+			}
+		}
+
+		if rule == event {
+			return true
+		}
+	}
+
+	return false
+}
diff --git a/pkg/binlog-filter/util.go b/pkg/binlog-filter/util.go
new file mode 100644
index 000000000..e076ba62e
--- /dev/null
+++ b/pkg/binlog-filter/util.go
@@ -0,0 +1,132 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package filter
+
+import (
+	"strings"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb/pkg/parser/ast"
+)
+
+// AstToDDLEvent returns filter.DDLEvent
+func AstToDDLEvent(node ast.StmtNode) EventType {
+	switch n := node.(type) {
+	case *ast.CreateDatabaseStmt:
+		return CreateDatabase
+	case *ast.DropDatabaseStmt:
+		return DropDatabase
+	case *ast.CreateTableStmt:
+		return CreateTable
+	case *ast.DropTableStmt:
+		if n.IsView {
+			return DropView
+		}
+		return DropTable
+	case *ast.TruncateTableStmt:
+		return TruncateTable
+	case *ast.RenameTableStmt:
+		return RenameTable
+	case *ast.CreateIndexStmt:
+		return CreateIndex
+	case *ast.DropIndexStmt:
+		return DropIndex
+	case *ast.AlterTableStmt:
+		return AlterTable
+	case *ast.CreateViewStmt:
+		return CreateView
+	case *ast.AlterDatabaseStmt:
+		return AlterDatabase
+	}
+
+	return NullEvent
+}
+
+// toEventType converts event type string to EventType and check if it is valid.
+func toEventType(es string) (EventType, error) {
+	event := EventType(strings.ToLower(es))
+	switch event {
+	case AllEvent,
+		AllDDL,
+		AllDML,
+		NullEvent,
+		NoneEvent,
+		NoneDDL,
+		NoneDML,
+		InsertEvent,
+		UpdateEvent,
+		DeleteEvent,
+		CreateDatabase,
+		DropDatabase,
+		AlterDatabase,
+		CreateTable,
+		DropTable,
+		TruncateTable,
+		RenameTable,
+		CreateIndex,
+		DropIndex,
+		CreateView,
+		DropView,
+		AlterTable,
+		AddTablePartition,
+		DropTablePartition,
+		TruncateTablePartition,
+
+		IncompatibleDDLChanges,
+		ValueRangeDecrease,
+		PrecisionDecrease,
+		ModifyColumn,
+		RenameColumn,
+		RenameIndex,
+		DropColumn,
+		DropPrimaryKey,
+		DropUniqueKey,
+		ModifyDefaultValue,
+		ModifyConstraint,
+		ModifyColumnsOrder,
+		ModifyCharset,
+		ModifyCollation,
+		RemoveAutoIncrement,
+		ModifyStorageEngine,
+		ReorganizePartition,
+		RebuildPartition,
+		CoalescePartition,
+		SplitPartition,
+		ExchangePartition,
+
+		ModifySchemaCharsetAndCollate,
+		ModifyTableCharsetAndCollate,
+		ModifyTableComment,
+		RecoverTable,
+		AlterTablePartitioning,
+		RemovePartitioning,
+		AddColumn,
+		SetDefaultValue,
+		RebaseAutoID,
+		AddPrimaryKey,
+		AlterIndexVisibility,
+		AlterTTLInfo,
+		AlterTTLRemove,
+		MultiSchemaChange:
+		return event, nil
+	case CreateSchema: // alias of CreateDatabase
+		return CreateDatabase, nil
+	case DropSchema: // alias of DropDatabase
+		return DropDatabase, nil
+	case AlterSchema:
+		return AlterDatabase, nil
+	default:
+		return NullEvent, errors.NotValidf("event type %s", es)
+	}
+}
diff --git a/pkg/config/filter.go b/pkg/config/filter.go
index 6ba326938..f8409a5ff 100644
--- a/pkg/config/filter.go
+++ b/pkg/config/filter.go
@@ -14,7 +14,7 @@
 package config
 
 import (
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 )
 
 // FilterConfig represents filter config for a changefeed
diff --git a/pkg/filter/filter.go b/pkg/filter/filter.go
index b7e61f060..d7c73a729 100644
--- a/pkg/filter/filter.go
+++ b/pkg/filter/filter.go
@@ -14,10 +14,10 @@
 package filter
 
 import (
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 )
 
diff --git a/pkg/filter/sql_event_filter.go b/pkg/filter/sql_event_filter.go
index 0cea36228..dd3b9c8a4 100644
--- a/pkg/filter/sql_event_filter.go
+++ b/pkg/filter/sql_event_filter.go
@@ -16,9 +16,9 @@ package filter
 import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"go.uber.org/zap"
diff --git a/pkg/filter/utils.go b/pkg/filter/utils.go
index 7a431ee58..be42603e6 100644
--- a/pkg/filter/utils.go
+++ b/pkg/filter/utils.go
@@ -16,10 +16,10 @@ package filter
 import (
 	"fmt"
 
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	tifilter "github.com/pingcap/tidb/pkg/util/filter"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 )


Test patch:
diff --git a/cdc/api/v2/model_test.go b/cdc/api/v2/model_test.go
index 27eff85db..9afc4e186 100644
--- a/cdc/api/v2/model_test.go
+++ b/cdc/api/v2/model_test.go
@@ -19,8 +19,8 @@ import (
 	"testing"
 	"time"
 
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/cdc/model"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/redo"
 	"github.com/pingcap/tiflow/pkg/util"
diff --git a/dm/config/source_config_test.go b/dm/config/source_config_test.go
index 2327222fb..27f459bcf 100644
--- a/dm/config/source_config_test.go
+++ b/dm/config/source_config_test.go
@@ -27,12 +27,12 @@ import (
 
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/go-mysql-org/go-mysql/mysql"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
 	"github.com/pingcap/tiflow/dm/pkg/conn"
 	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
 	"github.com/pingcap/tiflow/dm/pkg/encrypt"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v2"
 )
diff --git a/dm/config/task_converters_test.go b/dm/config/task_converters_test.go
index b6471885c..24eb74a09 100644
--- a/dm/config/task_converters_test.go
+++ b/dm/config/task_converters_test.go
@@ -17,12 +17,12 @@ import (
 	"testing"
 
 	"github.com/pingcap/check"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
 	"github.com/pingcap/tiflow/dm/openapi"
 	"github.com/pingcap/tiflow/dm/openapi/fixtures"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/stretchr/testify/require"
 )
 
diff --git a/dm/config/task_test.go b/dm/config/task_test.go
index 5d6cbf69b..bb90c5b83 100644
--- a/dm/config/task_test.go
+++ b/dm/config/task_test.go
@@ -23,7 +23,6 @@ import (
 	"testing"
 
 	"github.com/coreos/go-semver/semver"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	router "github.com/pingcap/tidb/pkg/util/table-router"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
@@ -31,6 +30,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/encrypt"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/stretchr/testify/require"
 )
 
diff --git a/dm/master/bootstrap_test.go b/dm/master/bootstrap_test.go
index 3d81dab1d..0db395e17 100644
--- a/dm/master/bootstrap_test.go
+++ b/dm/master/bootstrap_test.go
@@ -24,7 +24,6 @@ import (
 
 	"github.com/golang/mock/gomock"
 	. "github.com/pingcap/check"
-	filter "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/dm/config"
 	"github.com/pingcap/tiflow/dm/master/workerrpc"
 	"github.com/pingcap/tiflow/dm/pb"
@@ -33,6 +32,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/ha"
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
+	filter "github.com/pingcap/tiflow/pkg/binlog-filter"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/tests/v3/integration"
 )
diff --git a/dm/syncer/filter_test.go b/dm/syncer/filter_test.go
index d166a4fbc..579b14ed4 100644
--- a/dm/syncer/filter_test.go
+++ b/dm/syncer/filter_test.go
@@ -21,7 +21,6 @@ import (
 	"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/pkg/parser"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	"github.com/pingcap/tiflow/dm/config"
@@ -32,6 +31,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/schema"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 	"github.com/pingcap/tiflow/dm/syncer/dbconn"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 )
 
 type testFilterSuite struct {
diff --git a/dm/syncer/syncer_test.go b/dm/syncer/syncer_test.go
index 624ae290e..ce93d38e8 100644
--- a/dm/syncer/syncer_test.go
+++ b/dm/syncer/syncer_test.go
@@ -31,7 +31,6 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/failpoint"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	cm "github.com/pingcap/tidb-tools/pkg/column-mapping"
 	"github.com/pingcap/tidb/pkg/infoschema"
 	"github.com/pingcap/tidb/pkg/parser"
@@ -58,6 +57,7 @@ import (
 	"github.com/pingcap/tiflow/dm/syncer/binlogstream"
 	"github.com/pingcap/tiflow/dm/syncer/dbconn"
 	"github.com/pingcap/tiflow/dm/syncer/metrics"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/errorutil"
 	"github.com/pingcap/tiflow/pkg/sqlmodel"
 	"github.com/stretchr/testify/require"
diff --git a/dm/tests/openapi/run.sh b/dm/tests/openapi/run.sh
index 407754c99..5577da3d0 100644
--- a/dm/tests/openapi/run.sh
+++ b/dm/tests/openapi/run.sh
@@ -752,6 +752,7 @@ function test_start_task_with_condition() {
 	run_sql_source2 "CREATE TABLE openapi.t2(i TINYINT, j INT UNIQUE KEY);"
 
 	openapi_task_check "create_incremental_task_with_gtid_success" $task_name "" "" "" "" "" ""
+	curl http://127.0.0.1:8261/api/v1/tasks/incremental_task_use_start_time_after_create/sources/mysql-01/migrate_targets
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status $task_name" \
 		"\"stage\": \"Stopped\"" 2
diff --git a/pkg/binlog-filter/filter_test.go b/pkg/binlog-filter/filter_test.go
new file mode 100644
index 000000000..6d4e97782
--- /dev/null
+++ b/pkg/binlog-filter/filter_test.go
@@ -0,0 +1,332 @@
+// Copyright 2018 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package filter
+
+import (
+	"testing"
+
+	"github.com/pingcap/errors"
+	selector "github.com/pingcap/tidb/pkg/util/table-rule-selector"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFilter(t *testing.T) {
+	rules := []*BinlogEventRule{
+		{"Test_1_*", "abc*", []EventType{DeleteEvent, InsertEvent, CreateIndex, DropIndex, DropView}, []string{"^DROP\\s+PROCEDURE", "^CREATE\\s+PROCEDURE"}, nil, Ignore},
+		{"xxx_*", "abc_*", []EventType{AllDML, NoneDDL}, nil, nil, Ignore},
+		{"yyy_*", "abc_*", []EventType{EventType("ALL DML")}, nil, nil, Do},
+		{"Test_1_*", "abc*", []EventType{"wrong event"}, []string{"^DROP\\s+PROCEDURE", "^CREATE\\s+PROCEDURE"}, nil, Ignore},
+		{"cdc", "t1", []EventType{RebaseAutoID}, nil, nil, Ignore},
+	}
+
+	cases := []struct {
+		schema, table string
+		event         EventType
+		sql           string
+		action        ActionType
+	}{
+		{"test_1_a", "abc1", DeleteEvent, "", Ignore},
+		{"test_1_a", "abc1", InsertEvent, "", Ignore},
+		{"test_1_a", "abc1", UpdateEvent, "", Do},
+		{"test_1_a", "abc1", CreateIndex, "", Ignore},
+		{"test_1_a", "abc1", RenameTable, "", Do},
+		{"test_1_a", "abc1", NullEvent, "drop procedure abc", Ignore},
+		{"test_1_a", "abc1", NullEvent, "create procedure abc", Ignore},
+		{"test_1_a", "abc1", NullEvent, "create function abc", Do},
+		{"xxx_1", "abc_1", NullEvent, "create function abc", Do},
+		{"xxx_1", "abc_1", InsertEvent, "", Ignore},
+		{"xxx_1", "abc_1", CreateIndex, "", Do},
+		{"yyy_1", "abc_1", InsertEvent, "", Do},
+		{"yyy_1", "abc_1", CreateIndex, "", Ignore},
+		{"test_1_a", "abc1", DropView, "", Ignore},
+		{"cdc", "t1", RebaseAutoID, "", Ignore},
+	}
+
+	// initial binlog event filter
+	filter, err := NewBinlogEvent(false, rules)
+	require.NoError(t, err)
+
+	// insert duplicate rules
+	for _, rule := range rules {
+		err = filter.AddRule(rule)
+		require.Error(t, err)
+	}
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, cs.event, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+
+	// update rules
+	rules[0].Events = []EventType{}
+	rules[1].Action = Do
+	rules[2].Events = []EventType{"ALL DDL"}
+	rules = rules[:3]
+	for _, rule := range rules {
+		err = filter.UpdateRule(rule)
+		require.NoError(t, err)
+	}
+
+	cases[0].action = Do      // delete
+	cases[1].action = Do      // insert
+	cases[3].action = Do      // create index
+	cases[9].action = Do      // match all event and insert
+	cases[10].action = Ignore // match none event and create index
+	cases[11].action = Ignore // no match
+	cases[12].action = Do     // match all ddl
+	cases[13].action = Do     // match all ddl
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, cs.event, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+
+	// test multiple rules
+	rule := &BinlogEventRule{"test_*", "ab*", []EventType{InsertEvent, AllDDL}, []string{"^DROP\\s+PROCEDURE"}, nil, Do}
+	err = filter.AddRule(rule)
+	require.NoError(t, err)
+	cases[0].action = Ignore // delete
+	cases[2].action = Ignore // update
+	cases[4].action = Do     // rename table
+	cases[7].action = Ignore // create function
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, cs.event, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+
+	// remove rule
+	err = filter.RemoveRule(rules[0])
+	require.NoError(t, err)
+	// remove not existing rule
+	err = filter.RemoveRule(rules[0])
+	require.Error(t, err)
+	cases[3].action = Do // create index
+	cases[5].action = Do // drop procedure
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, cs.event, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+
+	// mismatched
+	action, err := filter.Filter("xxx_a", "", InsertEvent, "")
+	require.NoError(t, err)
+	require.Equal(t, Do, action)
+
+	// invalid rule
+	err = filter.Selector.Insert("test_1_*", "abc*", "error", selector.Insert)
+	require.NoError(t, err)
+	_, err = filter.Filter("test_1_a", "abc", InsertEvent, "")
+	require.Error(t, err)
+}
+
+func TestCaseSensitive(t *testing.T) {
+	// we test case insensitive in TestFilter
+	rules := []*BinlogEventRule{
+		{"Test_1_*", "abc*", []EventType{DeleteEvent, InsertEvent, CreateIndex, DropIndex}, []string{"^DROP\\s+PROCEDURE", "^CREATE\\s+PROCEDURE"}, nil, Ignore},
+		{"xxx_*", "abc_*", []EventType{AllDML, NoneDDL}, nil, nil, Ignore},
+	}
+
+	cases := []struct {
+		schema, table string
+		event         EventType
+		sql           string
+		action        ActionType
+	}{
+		{"test_1_a", "abc1", DeleteEvent, "", Do},
+		{"test_1_a", "abc1", InsertEvent, "", Do},
+		{"test_1_a", "abc1", UpdateEvent, "", Do},
+		{"test_1_a", "abc1", CreateIndex, "", Do},
+		{"test_1_a", "abc1", RenameTable, "", Do},
+		{"test_1_a", "abc1", NullEvent, "drop procedure abc", Do},
+		{"test_1_a", "abc1", NullEvent, "create procedure abc", Do},
+		{"test_1_a", "abc1", NullEvent, "create function abc", Do},
+		{"xxx_1", "abc_1", NullEvent, "create function abc", Do},
+		{"xxx_1", "abc_1", InsertEvent, "", Ignore},
+		{"xxx_1", "abc_1", CreateIndex, "", Do},
+	}
+
+	// initial binlog event filter
+	filter, err := NewBinlogEvent(true, rules)
+	require.NoError(t, err)
+
+	// insert duplicate rules
+	for _, rule := range rules {
+		err = filter.AddRule(rule)
+		require.Error(t, err)
+	}
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, cs.event, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+}
+
+func TestGlobalFilter(t *testing.T) {
+	schemaRule := &BinlogEventRule{
+		SchemaPattern: "*",
+		SQLPattern:    []string{"^FLUSH"},
+		Action:        Ignore,
+	}
+	tableRule := &BinlogEventRule{
+		SchemaPattern: "*",
+		TablePattern:  "*",
+		SQLPattern:    []string{"^FLUSH"},
+		Action:        Ignore,
+	}
+
+	cases := []struct {
+		schema string
+		table  string
+		sql    string
+		action ActionType
+	}{
+		{
+			schema: "db",
+			table:  "tbl",
+			sql:    "FLUSH ENGINE LOGS",
+			action: Ignore,
+		},
+		{
+			schema: "db",
+			table:  "",
+			sql:    "FLUSH ENGINE LOGS",
+			action: Ignore,
+		},
+		{
+			schema: "",
+			table:  "tbl",
+			sql:    "FLUSH ENGINE LOGS",
+			action: Ignore,
+		},
+		{
+			schema: "",
+			table:  "",
+			sql:    "FLUSH ENGINE LOGS",
+			action: Ignore,
+		},
+	}
+
+	// initial binlog event filter with schema rule
+	filter, err := NewBinlogEvent(false, []*BinlogEventRule{schemaRule})
+	require.NoError(t, err)
+
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, NullEvent, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+
+	// remove schema rule
+	err = filter.RemoveRule(schemaRule)
+	require.NoError(t, err)
+
+	// add table rule
+	err = filter.AddRule(tableRule)
+	require.NoError(t, err)
+
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, NullEvent, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+}
+
+func TestToEventType(t *testing.T) {
+	cases := []struct {
+		eventStr string
+		event    EventType
+		err      error
+	}{
+		{"", NullEvent, nil},
+		{"insert", InsertEvent, nil},
+		{"Insert", InsertEvent, nil},
+		{"update", UpdateEvent, nil},
+		{"UPDATE", UpdateEvent, nil},
+		{"delete", DeleteEvent, nil},
+		{"create", NullEvent, errors.NotValidf("event type %s", "create")},
+		{"create schema", CreateDatabase, nil},
+		{"create SCHEMA", CreateDatabase, nil},
+		{"create database", CreateDatabase, nil},
+		{"drop schema", DropDatabase, nil},
+		{"drop Schema", DropDatabase, nil},
+		{"drop database", DropDatabase, nil},
+		{"alter database", AlterDatabase, nil},
+		{"alter schema", AlterDatabase, nil},
+		{"create index", CreateIndex, nil},
+		{"add table partition", AddTablePartition, nil},
+		{"drop taBle partition", DropTablePartition, nil},
+		{"truncate tablE parTition", TruncateTablePartition, nil},
+		{"rebase auto id", RebaseAutoID, nil},
+		{"xxx", NullEvent, errors.NotValidf("event type %s", "xxx")},
+		{"I don't know", NullEvent, errors.NotValidf("event type %s", "I don't know")},
+	}
+
+	for _, cs := range cases {
+		event, err := toEventType(cs.eventStr)
+		require.Equal(t, cs.event, event)
+		if err != nil {
+			require.ErrorContains(t, err, cs.err.Error())
+		} else {
+			require.NoError(t, err)
+		}
+	}
+}
+
+func TestClassifyEvent(t *testing.T) {
+	cases := []struct {
+		event    EventType
+		evenType EventType
+		err      error
+	}{
+		{NullEvent, NullEvent, nil},
+		// dml
+		{InsertEvent, dml, nil},
+		{UpdateEvent, dml, nil},
+		{DeleteEvent, dml, nil},
+		// ddl
+		{CreateDatabase, ddl, nil},
+		{CreateSchema, ddl, nil},
+		{DropDatabase, incompatibleDDL, nil},
+		{DropSchema, incompatibleDDL, nil},
+		{AlterSchema, ddl, nil},
+		{CreateTable, ddl, nil},
+		{DropTable, incompatibleDDL, nil},
+		{TruncateTable, incompatibleDDL, nil},
+		{RenameTable, incompatibleDDL, nil},
+		{CreateIndex, ddl, nil},
+		{DropIndex, incompatibleDDL, nil},
+		{CreateView, ddl, nil},
+		{DropView, ddl, nil},
+		{AlterTable, ddl, nil},
+		{AddTablePartition, ddl, nil},
+		{DropTablePartition, incompatibleDDL, nil},
+		{RebaseAutoID, incompatibleDDL, nil},
+		{TruncateTablePartition, incompatibleDDL, nil},
+		{"create", NullEvent, errors.NotValidf("event type %s", "create")},
+		{EventType("xxx"), NullEvent, errors.NotValidf("event type %s", "xxx")},
+		{EventType("I don't know"), NullEvent, errors.NotValidf("event type %s", "I don't know")},
+	}
+
+	for _, cs := range cases {
+		et, err := ClassifyEvent(cs.event)
+		require.Equal(t, cs.evenType, et)
+		if err != nil {
+			require.ErrorContains(t, err, cs.err.Error())
+		} else {
+			require.NoError(t, err)
+		}
+	}
+}
diff --git a/pkg/filter/filter_test.go b/pkg/filter/filter_test.go
index 43f1b2335..9ed19e04c 100644
--- a/pkg/filter/filter_test.go
+++ b/pkg/filter/filter_test.go
@@ -16,9 +16,9 @@ package filter
 import (
 	"testing"
 
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tiflow/cdc/model"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/stretchr/testify/require"
 )
diff --git a/pkg/filter/sql_event_filter_test.go b/pkg/filter/sql_event_filter_test.go
index d23310334..0ded5109c 100644
--- a/pkg/filter/sql_event_filter_test.go
+++ b/pkg/filter/sql_event_filter_test.go
@@ -17,9 +17,9 @@ import (
 	"testing"
 
 	"github.com/pingcap/errors"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tiflow/cdc/model"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/stretchr/testify/require"