Repository: pingcap/tiflow
Pull request #4663 (2022-02-23 09:24:38)

sink(ticdc): Add a topic expression when dispatching messages (Part II)
<!--
Thank you for contributing to TiDB-CDC! Please read MD's [CONTRIBUTING](https://github.com/pingcap/tidb-cdc/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 #4423 

### What is changed and how it works?
This is the second part of https://github.com/pingcap/tiflow/pull/4540. This PR uses the topic expression to implement topic dispatchers. Detailed dispatching rules are described in the pervious PR.

### Check List <!--REMOVE the items that are not applicable-->

Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test

Code changes

 - Has exported function/method change
 - Has exported variable/fields change

### Release note <!-- bugfixes or new feature need a release note -->

```release-note 
None
```


ti-chi-bot: [REVIEW NOTIFICATION]

This pull request has not been approved.


To complete the [pull request process](https://book.prow.tidb.io/#/en/workflows/pr), please ask the reviewers in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/4663/owners) to review by filling `/cc @reviewer` in the comment.
After your PR has acquired the required number of LGTMs, you can assign this pull request to the committer in the [list](https://prow.tidb.io/tichi/repos/pingcap/tiflow/pulls/4663/owners) by filling  `/assign @committer` in the comment to help you merge this pull request.

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

<details>

Reviewer can indicate their review by submitting an approval review.
Reviewer can cancel approval by submitting a request changes review.
</details>

<!--Review Notification Identifier-->
codecov-commenter: # [Codecov](https://codecov.io/gh/pingcap/tiflow/pull/4663?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
> Merging [#4663](https://codecov.io/gh/pingcap/tiflow/pull/4663?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (fcf7a38) into [master](https://codecov.io/gh/pingcap/tiflow/commit/96075543f5692d4a75eec041c82bac672eb3594b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (9607554) will **decrease** coverage by `0.5409%`.
> The diff coverage is `45.3168%`.


| Flag | Coverage Δ | |
|---|---|---|
| cdc | `59.6868% <50.4524%> (-0.2355%)` | :arrow_down: |
| dm | `51.2654% <30.4157%> (-0.7635%)` | :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      #4663        +/-   ##
================================================
- Coverage   55.6402%   55.0993%   -0.5410%     
================================================
  Files           494        514        +20     
  Lines         61283      63685      +2402     
================================================
+ Hits          34098      35090       +992     
- Misses        23750      25104      +1354     
- Partials       3435       3491        +56     
```

overvenus: /merge
ti-chi-bot: This pull request has been accepted and is ready to merge. <details>Commit hash: fcf7a38638617dbc91abe6cf107c31bdc8ff51ba</details>
ti-chi-bot: @zhaoxinyu: Your PR was out of date, I have automatically updated it for you.

At the same time I will also trigger all tests for you:

/run-all-tests

If the CI test fails, you just re-trigger the test that failed and the bot will merge the PR for you after the CI passes.


<details>

Instructions for interacting with me using PR comments are available [here](https://prow.tidb.io/command-help).  If you have questions or suggestions related to my behavior, please file an issue against the [ti-community-infra/tichi](https://github.com/ti-community-infra/tichi/issues/new?title=Prow%20issue:) repository.
</details>
zhaoxinyu: /run-dm-integration-test

Patch:
diff --git a/cdc/sink/dispatcher/default.go b/cdc/sink/dispatcher/default.go
deleted file mode 100644
index 31d75eb0a..000000000
--- a/cdc/sink/dispatcher/default.go
+++ /dev/null
@@ -1,44 +0,0 @@
-// Copyright 2020 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 dispatcher
-
-import (
-	"github.com/pingcap/tiflow/cdc/model"
-)
-
-type defaultDispatcher struct {
-	partitionNum   int32
-	tbd            *tableDispatcher
-	ivd            *indexValueDispatcher
-	enableOldValue bool
-}
-
-func newDefaultDispatcher(partitionNum int32, enableOldValue bool) *defaultDispatcher {
-	return &defaultDispatcher{
-		partitionNum:   partitionNum,
-		tbd:            newTableDispatcher(partitionNum),
-		ivd:            newIndexValueDispatcher(partitionNum),
-		enableOldValue: enableOldValue,
-	}
-}
-
-func (d *defaultDispatcher) Dispatch(row *model.RowChangedEvent) int32 {
-	if d.enableOldValue {
-		return d.tbd.Dispatch(row)
-	}
-	if len(row.IndexColumns) != 1 {
-		return d.tbd.Dispatch(row)
-	}
-	return d.ivd.Dispatch(row)
-}
diff --git a/cdc/sink/dispatcher/dispatcher.go b/cdc/sink/dispatcher/dispatcher.go
deleted file mode 100644
index 6b4818295..000000000
--- a/cdc/sink/dispatcher/dispatcher.go
+++ /dev/null
@@ -1,128 +0,0 @@
-// Copyright 2020 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 dispatcher
-
-import (
-	"strings"
-
-	"github.com/pingcap/log"
-	filter "github.com/pingcap/tidb-tools/pkg/table-filter"
-	"github.com/pingcap/tiflow/cdc/model"
-	"github.com/pingcap/tiflow/pkg/config"
-	cerror "github.com/pingcap/tiflow/pkg/errors"
-	"go.uber.org/zap"
-)
-
-// Dispatcher is an abstraction for dispatching rows into different partitions
-type Dispatcher interface {
-	// Dispatch returns an index of partitions according to RowChangedEvent
-	Dispatch(row *model.RowChangedEvent) int32
-}
-
-type dispatchRule int
-
-const (
-	dispatchRuleDefault dispatchRule = iota
-	dispatchRuleRowID
-	dispatchRuleTS
-	dispatchRuleTable
-	dispatchRuleIndexValue
-)
-
-func (r *dispatchRule) fromString(rule string) {
-	switch strings.ToLower(rule) {
-	case "default":
-		*r = dispatchRuleDefault
-	case "rowid":
-		*r = dispatchRuleRowID
-	case "ts":
-		*r = dispatchRuleTS
-	case "table":
-		*r = dispatchRuleTable
-	case "index-value":
-		*r = dispatchRuleIndexValue
-	default:
-		*r = dispatchRuleDefault
-		log.Warn("can't support dispatch rule, using default rule", zap.String("rule", rule))
-	}
-}
-
-type dispatcherSwitcher struct {
-	rules []struct {
-		Dispatcher
-		filter.Filter
-	}
-}
-
-func (s *dispatcherSwitcher) Dispatch(row *model.RowChangedEvent) int32 {
-	return s.matchDispatcher(row).Dispatch(row)
-}
-
-func (s *dispatcherSwitcher) matchDispatcher(row *model.RowChangedEvent) Dispatcher {
-	for _, rule := range s.rules {
-		if !rule.MatchTable(row.Table.Schema, row.Table.Table) {
-			continue
-		}
-		return rule.Dispatcher
-	}
-	log.Panic("the dispatch rule must cover all tables")
-	return nil
-}
-
-// NewDispatcher creates a new dispatcher
-func NewDispatcher(cfg *config.ReplicaConfig, partitionNum int32) (Dispatcher, error) {
-	ruleConfigs := append(cfg.Sink.DispatchRules, &config.DispatchRule{
-		Matcher:    []string{"*.*"},
-		Dispatcher: "default",
-	})
-	rules := make([]struct {
-		Dispatcher
-		filter.Filter
-	}, 0, len(ruleConfigs))
-
-	for _, ruleConfig := range ruleConfigs {
-		f, err := filter.Parse(ruleConfig.Matcher)
-		if err != nil {
-			return nil, cerror.WrapError(cerror.ErrFilterRuleInvalid, err)
-		}
-		if !cfg.CaseSensitive {
-			f = filter.CaseInsensitive(f)
-		}
-		var d Dispatcher
-		var rule dispatchRule
-		rule.fromString(ruleConfig.Dispatcher)
-		switch rule {
-		case dispatchRuleRowID, dispatchRuleIndexValue:
-			if cfg.EnableOldValue {
-				log.Warn("This index-value distribution mode " +
-					"does not guarantee row-level orderliness when " +
-					"switching on the old value, so please use caution!")
-			}
-			d = newIndexValueDispatcher(partitionNum)
-		case dispatchRuleTS:
-			d = newTsDispatcher(partitionNum)
-		case dispatchRuleTable:
-			d = newTableDispatcher(partitionNum)
-		case dispatchRuleDefault:
-			d = newDefaultDispatcher(partitionNum, cfg.EnableOldValue)
-		}
-		rules = append(rules, struct {
-			Dispatcher
-			filter.Filter
-		}{Dispatcher: d, Filter: f})
-	}
-	return &dispatcherSwitcher{
-		rules: rules,
-	}, nil
-}
diff --git a/cdc/sink/dispatcher/event_router.go b/cdc/sink/dispatcher/event_router.go
new file mode 100644
index 000000000..8439ee824
--- /dev/null
+++ b/cdc/sink/dispatcher/event_router.go
@@ -0,0 +1,169 @@
+// Copyright 2022 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 dispatcher
+
+import (
+	"strings"
+
+	"github.com/pingcap/log"
+	filter "github.com/pingcap/tidb-tools/pkg/table-filter"
+	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/cdc/sink/dispatcher/partition"
+	"github.com/pingcap/tiflow/cdc/sink/dispatcher/topic"
+	"github.com/pingcap/tiflow/pkg/config"
+	cerror "github.com/pingcap/tiflow/pkg/errors"
+	"go.uber.org/zap"
+)
+
+type partitionDispatchRule int
+
+const (
+	partitionDispatchRuleDefault partitionDispatchRule = iota
+	partitionDispatchRuleRowID
+	partitionDispatchRuleTS
+	partitionDispatchRuleTable
+	partitionDispatchRuleIndexValue
+)
+
+func (r *partitionDispatchRule) fromString(rule string) {
+	switch strings.ToLower(rule) {
+	case "default":
+		*r = partitionDispatchRuleDefault
+	case "rowid":
+		*r = partitionDispatchRuleRowID
+	case "ts":
+		*r = partitionDispatchRuleTS
+	case "table":
+		*r = partitionDispatchRuleTable
+	case "index-value":
+		*r = partitionDispatchRuleIndexValue
+	default:
+		*r = partitionDispatchRuleDefault
+		log.Warn("can't support dispatch rule, using default rule", zap.String("rule", rule))
+	}
+}
+
+// EventRouter is a router, it determines which topic and which partition
+// an event should be dispatched to.
+type EventRouter struct {
+	rules []struct {
+		partitionDispatcher partition.Dispatcher
+		topicDispatcher     topic.Dispatcher
+		filter.Filter
+	}
+}
+
+// DispatchRowChangedEvent returns the target topic and partition.
+func (s *EventRouter) DispatchRowChangedEvent(row *model.RowChangedEvent) (string, int32) {
+	topicDispatcher, partitionDispatcher := s.matchDispatcher(row)
+	topic := topicDispatcher.DispatchRowChangedEvent(row)
+	partition := partitionDispatcher.DispatchRowChangedEvent(row)
+
+	return topic, partition
+}
+
+// matchDispatcher returns the target topic dispatcher and partition dispatcher if a
+// row changed event matches a specific table filter.
+func (s *EventRouter) matchDispatcher(row *model.RowChangedEvent) (topic.Dispatcher, partition.Dispatcher) {
+	for _, rule := range s.rules {
+		if !rule.MatchTable(row.Table.Schema, row.Table.Table) {
+			continue
+		}
+		return rule.topicDispatcher, rule.partitionDispatcher
+	}
+	log.Panic("the dispatch rule must cover all tables")
+	return nil, nil
+}
+
+// getPartitionDispatcher returns the partition dispatcher for a specific partition rule.
+func getPartitionDispatcher(ruleConfig *config.DispatchRule, partitionNum int32,
+	enableOldValue bool) partition.Dispatcher {
+	var (
+		d    partition.Dispatcher
+		rule partitionDispatchRule
+	)
+	rule.fromString(ruleConfig.PartitionRule)
+	switch rule {
+	case partitionDispatchRuleRowID, partitionDispatchRuleIndexValue:
+		if enableOldValue {
+			log.Warn("This index-value distribution mode " +
+				"does not guarantee row-level orderliness when " +
+				"switching on the old value, so please use caution!")
+		}
+		d = partition.NewIndexValueDispatcher(partitionNum)
+	case partitionDispatchRuleTS:
+		d = partition.NewTsDispatcher(partitionNum)
+	case partitionDispatchRuleTable:
+		d = partition.NewTableDispatcher(partitionNum)
+	case partitionDispatchRuleDefault:
+		d = partition.NewDefaultDispatcher(partitionNum, enableOldValue)
+	}
+
+	return d
+}
+
+// getTopicDispatcher returns the topic dispatcher for a specific topic rule (aka topic expression).
+func getTopicDispatcher(ruleConfig *config.DispatchRule, defaultTopic string) (topic.Dispatcher, error) {
+	if ruleConfig.TopicRule == "" {
+		return topic.NewStaticTopicDispatcher(defaultTopic), nil
+	}
+
+	// check if this rule is a valid topic expression
+	topicExpr := topic.Expression(ruleConfig.TopicRule)
+	err := topicExpr.Validate()
+	if err != nil {
+		return nil, err
+	}
+	return topic.NewDynamicTopicDispatcher(defaultTopic, topicExpr), nil
+}
+
+// NewEventRouter creates a new EventRouter
+func NewEventRouter(cfg *config.ReplicaConfig, partitionNum int32, defaultTopic string) (*EventRouter, error) {
+	// If an event does not match any dispatching rules in the config file,
+	// it will be dispatched by the default partition dispatcher and
+	// static topic dispatcher because it matches *.* rule.
+	ruleConfigs := append(cfg.Sink.DispatchRules, &config.DispatchRule{
+		Matcher:       []string{"*.*"},
+		PartitionRule: "default",
+		TopicRule:     "",
+	})
+	rules := make([]struct {
+		partitionDispatcher partition.Dispatcher
+		topicDispatcher     topic.Dispatcher
+		filter.Filter
+	}, 0, len(ruleConfigs))
+
+	for _, ruleConfig := range ruleConfigs {
+		f, err := filter.Parse(ruleConfig.Matcher)
+		if err != nil {
+			return nil, cerror.WrapError(cerror.ErrFilterRuleInvalid, err)
+		}
+		if !cfg.CaseSensitive {
+			f = filter.CaseInsensitive(f)
+		}
+
+		d := getPartitionDispatcher(ruleConfig, partitionNum, cfg.EnableOldValue)
+		t, err := getTopicDispatcher(ruleConfig, defaultTopic)
+		if err != nil {
+			return nil, err
+		}
+		rules = append(rules, struct {
+			partitionDispatcher partition.Dispatcher
+			topicDispatcher     topic.Dispatcher
+			filter.Filter
+		}{partitionDispatcher: d, topicDispatcher: t, Filter: f})
+	}
+	return &EventRouter{
+		rules: rules,
+	}, nil
+}
diff --git a/cdc/sink/dispatcher/partition/default.go b/cdc/sink/dispatcher/partition/default.go
new file mode 100644
index 000000000..8ca29c51c
--- /dev/null
+++ b/cdc/sink/dispatcher/partition/default.go
@@ -0,0 +1,48 @@
+// Copyright 2022 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 partition
+
+import (
+	"github.com/pingcap/tiflow/cdc/model"
+)
+
+// DefaultDispatcher is the default partition dispatcher.
+type DefaultDispatcher struct {
+	partitionNum   int32
+	tbd            *TableDispatcher
+	ivd            *IndexValueDispatcher
+	enableOldValue bool
+}
+
+// NewDefaultDispatcher creates a DefaultDispatcher.
+func NewDefaultDispatcher(partitionNum int32, enableOldValue bool) *DefaultDispatcher {
+	return &DefaultDispatcher{
+		partitionNum:   partitionNum,
+		tbd:            NewTableDispatcher(partitionNum),
+		ivd:            NewIndexValueDispatcher(partitionNum),
+		enableOldValue: enableOldValue,
+	}
+}
+
+// DispatchRowChangedEvent returns the target partition to which
+// a row changed event should be dispatched.
+func (d *DefaultDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent) int32 {
+	if d.enableOldValue {
+		return d.tbd.DispatchRowChangedEvent(row)
+	}
+	if len(row.IndexColumns) != 1 {
+		return d.tbd.DispatchRowChangedEvent(row)
+	}
+	return d.ivd.DispatchRowChangedEvent(row)
+}
diff --git a/cdc/sink/dispatcher/partition/dispatcher.go b/cdc/sink/dispatcher/partition/dispatcher.go
new file mode 100644
index 000000000..5268fee92
--- /dev/null
+++ b/cdc/sink/dispatcher/partition/dispatcher.go
@@ -0,0 +1,24 @@
+// Copyright 2022 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 partition
+
+import (
+	"github.com/pingcap/tiflow/cdc/model"
+)
+
+// Dispatcher is an abstraction for dispatching rows into different partitions
+type Dispatcher interface {
+	// Dispatch returns an index of partitions according to RowChangedEvent
+	DispatchRowChangedEvent(row *model.RowChangedEvent) int32
+}
diff --git a/cdc/sink/dispatcher/index_value.go b/cdc/sink/dispatcher/partition/index_value.go
similarity index 70%
rename from cdc/sink/dispatcher/index_value.go
rename to cdc/sink/dispatcher/partition/index_value.go
index 79913bc63..66a171c17 100644
--- a/cdc/sink/dispatcher/index_value.go
+++ b/cdc/sink/dispatcher/partition/index_value.go
@@ -1,4 +1,4 @@
-// Copyright 2020 PingCAP, Inc.
+// Copyright 2022 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -11,26 +11,30 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package dispatcher
+package partition
 
 import (
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/hash"
 )
 
-type indexValueDispatcher struct {
+// IndexValueDispatcher is a partition dispatcher which dispatches events based on the index value.
+type IndexValueDispatcher struct {
 	partitionNum int32
 	hasher       *hash.PositionInertia
 }
 
-func newIndexValueDispatcher(partitionNum int32) *indexValueDispatcher {
-	return &indexValueDispatcher{
+// NewIndexValueDispatcher creates a IndexValueDispatcher.
+func NewIndexValueDispatcher(partitionNum int32) *IndexValueDispatcher {
+	return &IndexValueDispatcher{
 		partitionNum: partitionNum,
 		hasher:       hash.NewPositionInertia(),
 	}
 }
 
-func (r *indexValueDispatcher) Dispatch(row *model.RowChangedEvent) int32 {
+// DispatchRowChangedEvent returns the target partition to which
+// a row changed event should be dispatched.
+func (r *IndexValueDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent) int32 {
 	r.hasher.Reset()
 	r.hasher.Write([]byte(row.Table.Schema), []byte(row.Table.Table))
 	// FIXME(leoppro): if the row events includes both pre-cols and cols
diff --git a/cdc/sink/dispatcher/table.go b/cdc/sink/dispatcher/partition/table.go
similarity index 63%
rename from cdc/sink/dispatcher/table.go
rename to cdc/sink/dispatcher/partition/table.go
index 98617f7ce..a256d5495 100644
--- a/cdc/sink/dispatcher/table.go
+++ b/cdc/sink/dispatcher/partition/table.go
@@ -1,4 +1,4 @@
-// Copyright 2020 PingCAP, Inc.
+// Copyright 2022 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -11,26 +11,31 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package dispatcher
+package partition
 
 import (
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/hash"
 )
 
-type tableDispatcher struct {
+// TableDispatcher is a partition dispatcher which dispatches events
+// based on the schema and table name.
+type TableDispatcher struct {
 	partitionNum int32
 	hasher       *hash.PositionInertia
 }
 
-func newTableDispatcher(partitionNum int32) *tableDispatcher {
-	return &tableDispatcher{
+// NewTableDispatcher creates a TableDispatcher.
+func NewTableDispatcher(partitionNum int32) *TableDispatcher {
+	return &TableDispatcher{
 		partitionNum: partitionNum,
 		hasher:       hash.NewPositionInertia(),
 	}
 }
 
-func (t *tableDispatcher) Dispatch(row *model.RowChangedEvent) int32 {
+// DispatchRowChangedEvent returns the target partition to which
+// a row changed event should be dispatched.
+func (t *TableDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent) int32 {
 	t.hasher.Reset()
 	// distribute partition by table
 	t.hasher.Write([]byte(row.Table.Schema), []byte(row.Table.Table))
diff --git a/cdc/sink/dispatcher/ts.go b/cdc/sink/dispatcher/partition/ts.go
similarity index 57%
rename from cdc/sink/dispatcher/ts.go
rename to cdc/sink/dispatcher/partition/ts.go
index e061734d2..72ed9e037 100644
--- a/cdc/sink/dispatcher/ts.go
+++ b/cdc/sink/dispatcher/partition/ts.go
@@ -1,4 +1,4 @@
-// Copyright 2020 PingCAP, Inc.
+// Copyright 2022 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -11,20 +11,24 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package dispatcher
+package partition
 
 import "github.com/pingcap/tiflow/cdc/model"
 
-type tsDispatcher struct {
+// TsDispatcher is a partition dispatcher which dispatch events based on ts.
+type TsDispatcher struct {
 	partitionNum int32
 }
 
-func newTsDispatcher(partitionNum int32) *tsDispatcher {
-	return &tsDispatcher{
+// NewTsDispatcher creates a TsDispatcher.
+func NewTsDispatcher(partitionNum int32) *TsDispatcher {
+	return &TsDispatcher{
 		partitionNum: partitionNum,
 	}
 }
 
-func (t *tsDispatcher) Dispatch(row *model.RowChangedEvent) int32 {
+// DispatchRowChangedEvent returns the target partition to which
+// a row changed event should be dispatched.
+func (t *TsDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent) int32 {
 	return int32(row.CommitTs % uint64(t.partitionNum))
 }
diff --git a/cdc/sink/dispatcher/topic/dispatcher.go b/cdc/sink/dispatcher/topic/dispatcher.go
new file mode 100644
index 000000000..c49bb1684
--- /dev/null
+++ b/cdc/sink/dispatcher/topic/dispatcher.go
@@ -0,0 +1,76 @@
+// Copyright 2022 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 topic
+
+import (
+	"github.com/pingcap/tiflow/cdc/model"
+)
+
+// Dispatcher is an abstraction for dispatching rows and ddls into different topics.
+type Dispatcher interface {
+	DispatchRowChangedEvent(row *model.RowChangedEvent) string
+	DispatchDDLEvent(ddl *model.DDLEvent) string
+}
+
+// StaticTopicDispatcher is a topic dispatcher which dispatches rows and ddls to the default topic.
+type StaticTopicDispatcher struct {
+	defaultTopic string
+}
+
+// NewStaticTopicDispatcher returns a StaticTopicDispatcher.
+func NewStaticTopicDispatcher(defaultTopic string) *StaticTopicDispatcher {
+	return &StaticTopicDispatcher{
+		defaultTopic: defaultTopic,
+	}
+}
+
+// DispatchRowChangedEvent returns the target topic to which a row should be dispatched to.
+func (s *StaticTopicDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent) string {
+	return s.defaultTopic
+}
+
+// DispatchDDLEvent returns the target topic to which a ddl should be dispatched.
+func (s *StaticTopicDispatcher) DispatchDDLEvent(ddl *model.DDLEvent) string {
+	return s.defaultTopic
+}
+
+// DynamicTopicDispatcher is a topic dispatcher which dispatches rows and ddls
+// dynamically to the target topics.
+type DynamicTopicDispatcher struct {
+	defaultTopic string
+	expression   Expression
+}
+
+// NewDynamicTopicDispatcher creates a DynamicTopicDispatcher.
+func NewDynamicTopicDispatcher(defaultTopic string, topicExpr Expression) *DynamicTopicDispatcher {
+	return &DynamicTopicDispatcher{
+		defaultTopic: defaultTopic,
+		expression:   topicExpr,
+	}
+}
+
+// DispatchRowChangedEvent returns the target topic to which a row should be dispatched to.
+func (d *DynamicTopicDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent) string {
+	return d.expression.Substitute(row.Table.Schema, row.Table.Table)
+}
+
+// DispatchDDLEvent returns the target topic to which a ddl should be dispatched.
+// If the ddl is a schema-level event such as 'CREATE DATABASE', 'DROP DATABASE', etc.,
+// that event will be dispatched to default topic.
+func (d *DynamicTopicDispatcher) DispatchDDLEvent(ddl *model.DDLEvent) string {
+	if len(ddl.TableInfo.Table) == 0 {
+		return d.defaultTopic
+	}
+	return d.expression.Substitute(ddl.TableInfo.Schema, ddl.TableInfo.Table)
+}
diff --git a/cdc/sink/mq.go b/cdc/sink/mq.go
index 69c2800d6..d468a598c 100644
--- a/cdc/sink/mq.go
+++ b/cdc/sink/mq.go
@@ -54,7 +54,7 @@ const (
 
 type mqSink struct {
 	mqProducer     producer.Producer
-	dispatcher     dispatcher.Dispatcher
+	eventRouter    *dispatcher.EventRouter
 	encoderBuilder codec.EncoderBuilder
 	filter         *filter.Filter
 	protocol       config.Protocol
@@ -71,7 +71,7 @@ type mqSink struct {
 
 func newMqSink(
 	ctx context.Context, credential *security.Credential, mqProducer producer.Producer,
-	filter *filter.Filter, replicaConfig *config.ReplicaConfig, opts map[string]string,
+	filter *filter.Filter, defaultTopic string, replicaConfig *config.ReplicaConfig, opts map[string]string,
 	errCh chan error,
 ) (*mqSink, error) {
 	var protocol config.Protocol
@@ -89,7 +89,7 @@ func newMqSink(
 	}
 
 	partitionNum := mqProducer.GetPartitionNum()
-	d, err := dispatcher.NewDispatcher(replicaConfig, partitionNum)
+	d, err := dispatcher.NewEventRouter(replicaConfig, partitionNum, defaultTopic)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -111,7 +111,7 @@ func newMqSink(
 
 	s := &mqSink{
 		mqProducer:     mqProducer,
-		dispatcher:     d,
+		eventRouter:    d,
 		encoderBuilder: encoderBuilder,
 		filter:         filter,
 		protocol:       protocol,
@@ -158,7 +158,7 @@ func (k *mqSink) EmitRowChangedEvents(ctx context.Context, rows ...*model.RowCha
 				zap.Any("role", k.role))
 			continue
 		}
-		partition := k.dispatcher.Dispatch(row)
+		_, partition := k.eventRouter.DispatchRowChangedEvent(row)
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
@@ -342,7 +342,7 @@ func newKafkaSaramaSink(ctx context.Context, sinkURI *url.URL,
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
-	sink, err := newMqSink(ctx, producerConfig.Credential, sProducer, filter, replicaConfig, opts, errCh)
+	sink, err := newMqSink(ctx, producerConfig.Credential, sProducer, filter, topic, replicaConfig, opts, errCh)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
@@ -376,7 +376,7 @@ func newPulsarSink(ctx context.Context, sinkURI *url.URL, filter *filter.Filter,
 	// For now, it's a placeholder. Avro format have to make connection to Schema Registry,
 	// and it may need credential.
 	credential := &security.Credential{}
-	sink, err := newMqSink(ctx, credential, producer, filter, replicaConfig, opts, errCh)
+	sink, err := newMqSink(ctx, credential, producer, filter, "", replicaConfig, opts, errCh)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
diff --git a/pkg/cmd/cli/cli_changefeed_create.go b/pkg/cmd/cli/cli_changefeed_create.go
index 1f819a5ce..d5b91a1af 100644
--- a/pkg/cmd/cli/cli_changefeed_create.go
+++ b/pkg/cmd/cli/cli_changefeed_create.go
@@ -218,7 +218,7 @@ func (o *createChangefeedOptions) completeCfg(
 	}
 
 	for _, rules := range cfg.Sink.DispatchRules {
-		switch strings.ToLower(rules.Dispatcher) {
+		switch strings.ToLower(rules.PartitionRule) {
 		case "rowid", "index-value":
 			if cfg.EnableOldValue {
 				cmd.Printf("[WARN] This index-value distribution mode "+
diff --git a/pkg/cmd/util/changefeed.toml b/pkg/cmd/util/changefeed.toml
index db5f686d7..9c305a4bd 100644
--- a/pkg/cmd/util/changefeed.toml
+++ b/pkg/cmd/util/changefeed.toml
@@ -27,8 +27,8 @@ worker-num = 16
 # For MQ Sinks, you can configure event distribution rules through dispatchers
 # Dispatchers support default, ts, rowid and table
 dispatchers = [
-    { matcher = ['test1.*', 'test2.*'], dispatcher = "ts" },
-    { matcher = ['test3.*', 'test4.*'], dispatcher = "rowid" },
+    { matcher = ['test1.*', 'test2.*'], dispatcher = "ts", topic = "hello_{schema}" },
+    { matcher = ['test3.*', 'test4.*'], dispatcher = "rowid", topic = "{schema}_world" },
 ]
 # 对于 MQ 类的 Sink，可以通过 column-selectors 配置 column 选择器
 # For MQ Sinks, you can configure column selector rules through column-selectors
diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go
index 8f29bd88f..ca55f8ae4 100644
--- a/pkg/config/replica_config.go
+++ b/pkg/config/replica_config.go
@@ -121,8 +121,8 @@ func (c *replicaConfig) fillFromV1(v1 *outdated.ReplicaConfigV1) {
 	}
 	for _, dispatch := range v1.Sink.DispatchRules {
 		c.Sink.DispatchRules = append(c.Sink.DispatchRules, &DispatchRule{
-			Matcher:    []string{fmt.Sprintf("%s.%s", dispatch.Schema, dispatch.Name)},
-			Dispatcher: dispatch.Rule,
+			Matcher:       []string{fmt.Sprintf("%s.%s", dispatch.Schema, dispatch.Name)},
+			PartitionRule: dispatch.Rule,
 		})
 	}
 }
diff --git a/pkg/config/sink.go b/pkg/config/sink.go
index f82b9b4aa..36aa95544 100644
--- a/pkg/config/sink.go
+++ b/pkg/config/sink.go
@@ -40,8 +40,9 @@ type SinkConfig struct {
 
 // DispatchRule represents partition rule for a table
 type DispatchRule struct {
-	Matcher    []string `toml:"matcher" json:"matcher"`
-	Dispatcher string   `toml:"dispatcher" json:"dispatcher"`
+	Matcher       []string `toml:"matcher" json:"matcher"`
+	PartitionRule string   `toml:"dispatcher" json:"dispatcher"`
+	TopicRule     string   `toml:"topic" json:"topic"`
 }
 
 // ColumnSelector represents a column selector for a table.


Test patch:
diff --git a/cdc/model/changefeed_test.go b/cdc/model/changefeed_test.go
index f74dc2df1..27f3b1951 100644
--- a/cdc/model/changefeed_test.go
+++ b/cdc/model/changefeed_test.go
@@ -149,8 +149,8 @@ func TestFillV1(t *testing.T) {
 			},
 			Sink: &config.SinkConfig{
 				DispatchRules: []*config.DispatchRule{
-					{Matcher: []string{"test.tbl3"}, Dispatcher: "ts"},
-					{Matcher: []string{"test.tbl4"}, Dispatcher: "rowid"},
+					{Matcher: []string{"test.tbl3"}, PartitionRule: "ts"},
+					{Matcher: []string{"test.tbl4"}, PartitionRule: "rowid"},
 				},
 			},
 			Cyclic: &config.CyclicConfig{
diff --git a/cdc/sink/dispatcher/event_router_test.go b/cdc/sink/dispatcher/event_router_test.go
new file mode 100644
index 000000000..008f5778a
--- /dev/null
+++ b/cdc/sink/dispatcher/event_router_test.go
@@ -0,0 +1,107 @@
+// Copyright 2022 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 dispatcher
+
+import (
+	"testing"
+
+	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/cdc/sink/dispatcher/partition"
+	"github.com/pingcap/tiflow/cdc/sink/dispatcher/topic"
+	"github.com/pingcap/tiflow/pkg/config"
+	"github.com/stretchr/testify/require"
+)
+
+func TestEventRouter(t *testing.T) {
+	t.Parallel()
+
+	d, err := NewEventRouter(config.GetDefaultReplicaConfig(), 4, "")
+	require.Nil(t, err)
+	topicDispatcher, partitionDispatcher := d.matchDispatcher(&model.RowChangedEvent{
+		Table: &model.TableName{
+			Schema: "test", Table: "test",
+		},
+	})
+	require.IsType(t, &topic.StaticTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.DefaultDispatcher{}, partitionDispatcher)
+
+	d, err = NewEventRouter(&config.ReplicaConfig{
+		Sink: &config.SinkConfig{
+			DispatchRules: []*config.DispatchRule{
+				{Matcher: []string{"test_default1.*"}, PartitionRule: "default"},
+				{Matcher: []string{"test_default2.*"}, PartitionRule: "unknown-dispatcher"},
+				{Matcher: []string{"test_table.*"}, PartitionRule: "table", TopicRule: "hello_{schema}_world"},
+				{Matcher: []string{"test_index_value.*"}, PartitionRule: "index-value", TopicRule: "{schema}_world"},
+				{Matcher: []string{"test.*"}, PartitionRule: "rowid", TopicRule: "hello_{schema}"},
+				{Matcher: []string{"*.*", "!*.test"}, PartitionRule: "ts", TopicRule: "{schema}_{table}"},
+			},
+		},
+	}, 4, "")
+	require.Nil(t, err)
+	topicDispatcher, partitionDispatcher = d.matchDispatcher(&model.RowChangedEvent{
+		Table: &model.TableName{
+			Schema: "test", Table: "table1",
+		},
+	})
+	require.IsType(t, &topic.DynamicTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.IndexValueDispatcher{}, partitionDispatcher)
+
+	topicDispatcher, partitionDispatcher = d.matchDispatcher(&model.RowChangedEvent{
+		Table: &model.TableName{
+			Schema: "sbs", Table: "table2",
+		},
+	})
+	require.IsType(t, &topic.DynamicTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.TsDispatcher{}, partitionDispatcher)
+
+	topicDispatcher, partitionDispatcher = d.matchDispatcher(&model.RowChangedEvent{
+		Table: &model.TableName{
+			Schema: "sbs", Table: "test",
+		},
+	})
+	require.IsType(t, &topic.StaticTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.DefaultDispatcher{}, partitionDispatcher)
+
+	topicDispatcher, partitionDispatcher = d.matchDispatcher(&model.RowChangedEvent{
+		Table: &model.TableName{
+			Schema: "test_default1", Table: "test",
+		},
+	})
+	require.IsType(t, &topic.StaticTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.DefaultDispatcher{}, partitionDispatcher)
+
+	topicDispatcher, partitionDispatcher = d.matchDispatcher(&model.RowChangedEvent{
+		Table: &model.TableName{
+			Schema: "test_default2", Table: "test",
+		},
+	})
+	require.IsType(t, &topic.StaticTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.DefaultDispatcher{}, partitionDispatcher)
+
+	topicDispatcher, partitionDispatcher = d.matchDispatcher(&model.RowChangedEvent{
+		Table: &model.TableName{
+			Schema: "test_table", Table: "test",
+		},
+	})
+	require.IsType(t, &topic.DynamicTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.TableDispatcher{}, partitionDispatcher)
+
+	topicDispatcher, partitionDispatcher = d.matchDispatcher(&model.RowChangedEvent{
+		Table: &model.TableName{
+			Schema: "test_index_value", Table: "test",
+		},
+	})
+	require.IsType(t, &topic.DynamicTopicDispatcher{}, topicDispatcher)
+	require.IsType(t, &partition.IndexValueDispatcher{}, partitionDispatcher)
+}
diff --git a/cdc/sink/dispatcher/default_test.go b/cdc/sink/dispatcher/partition/default_test.go
similarity index 88%
rename from cdc/sink/dispatcher/default_test.go
rename to cdc/sink/dispatcher/partition/default_test.go
index ccddd3052..c2e7e9c46 100644
--- a/cdc/sink/dispatcher/default_test.go
+++ b/cdc/sink/dispatcher/partition/default_test.go
@@ -1,4 +1,4 @@
-// Copyright 2020 PingCAP, Inc.
+// Copyright 2022 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -11,7 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package dispatcher
+package partition
 
 import (
 	"testing"
@@ -25,7 +25,7 @@ func TestDefaultDispatcher(t *testing.T) {
 
 	testCases := []struct {
 		row             *model.RowChangedEvent
-		exceptPartition int32
+		expectPartition int32
 	}{
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
@@ -40,7 +40,7 @@ func TestDefaultDispatcher(t *testing.T) {
 				},
 			},
 			IndexColumns: [][]int{{0}},
-		}, exceptPartition: 11},
+		}, expectPartition: 11},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -54,7 +54,7 @@ func TestDefaultDispatcher(t *testing.T) {
 				},
 			},
 			IndexColumns: [][]int{{0}},
-		}, exceptPartition: 1},
+		}, expectPartition: 1},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -68,7 +68,7 @@ func TestDefaultDispatcher(t *testing.T) {
 				},
 			},
 			IndexColumns: [][]int{{0}},
-		}, exceptPartition: 7},
+		}, expectPartition: 7},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -85,7 +85,7 @@ func TestDefaultDispatcher(t *testing.T) {
 				},
 			},
 			IndexColumns: [][]int{{0}},
-		}, exceptPartition: 1},
+		}, expectPartition: 1},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -102,7 +102,7 @@ func TestDefaultDispatcher(t *testing.T) {
 				},
 			},
 			IndexColumns: [][]int{{0}},
-		}, exceptPartition: 11},
+		}, expectPartition: 11},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -119,7 +119,7 @@ func TestDefaultDispatcher(t *testing.T) {
 				},
 			},
 			IndexColumns: [][]int{{0}},
-		}, exceptPartition: 13},
+		}, expectPartition: 13},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -136,7 +136,7 @@ func TestDefaultDispatcher(t *testing.T) {
 				},
 			},
 			IndexColumns: [][]int{{0}},
-		}, exceptPartition: 13},
+		}, expectPartition: 13},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -155,7 +155,7 @@ func TestDefaultDispatcher(t *testing.T) {
 				},
 			},
 			IndexColumns: [][]int{{0}, {1}},
-		}, exceptPartition: 3},
+		}, expectPartition: 3},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -173,7 +173,7 @@ func TestDefaultDispatcher(t *testing.T) {
 				},
 			},
 			IndexColumns: [][]int{{0}, {1}},
-		}, exceptPartition: 3},
+		}, expectPartition: 3},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -191,11 +191,11 @@ func TestDefaultDispatcher(t *testing.T) {
 				},
 			},
 			IndexColumns: [][]int{{0}, {1}},
-		}, exceptPartition: 3},
+		}, expectPartition: 3},
 	}
-	p := newDefaultDispatcher(16, false)
+	p := NewDefaultDispatcher(16, false)
 	for _, tc := range testCases {
-		require.Equal(t, tc.exceptPartition, p.Dispatch(tc.row))
+		require.Equal(t, tc.expectPartition, p.DispatchRowChangedEvent(tc.row))
 	}
 }
 
@@ -221,6 +221,6 @@ func TestDefaultDispatcherWithOldValue(t *testing.T) {
 		IndexColumns: [][]int{{0}, {1}},
 	}
 
-	p := newDefaultDispatcher(16, true)
-	require.Equal(t, int32(3), p.Dispatch(row))
+	p := NewDefaultDispatcher(16, true)
+	require.Equal(t, int32(3), p.DispatchRowChangedEvent(row))
 }
diff --git a/cdc/sink/dispatcher/index_value_test.go b/cdc/sink/dispatcher/partition/index_value_test.go
similarity index 88%
rename from cdc/sink/dispatcher/index_value_test.go
rename to cdc/sink/dispatcher/partition/index_value_test.go
index b52ed5488..6b8f397e9 100644
--- a/cdc/sink/dispatcher/index_value_test.go
+++ b/cdc/sink/dispatcher/partition/index_value_test.go
@@ -1,4 +1,4 @@
-// Copyright 2020 PingCAP, Inc.
+// Copyright 2022 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -11,7 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package dispatcher
+package partition
 
 import (
 	"testing"
@@ -25,7 +25,7 @@ func TestIndexValueDispatcher(t *testing.T) {
 
 	testCases := []struct {
 		row             *model.RowChangedEvent
-		exceptPartition int32
+		expectPartition int32
 	}{
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
@@ -43,7 +43,7 @@ func TestIndexValueDispatcher(t *testing.T) {
 					Flag:  0,
 				},
 			},
-		}, exceptPartition: 2},
+		}, expectPartition: 2},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -60,7 +60,7 @@ func TestIndexValueDispatcher(t *testing.T) {
 					Flag:  0,
 				},
 			},
-		}, exceptPartition: 11},
+		}, expectPartition: 11},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -77,7 +77,7 @@ func TestIndexValueDispatcher(t *testing.T) {
 					Flag:  0,
 				},
 			},
-		}, exceptPartition: 2},
+		}, expectPartition: 2},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -94,7 +94,7 @@ func TestIndexValueDispatcher(t *testing.T) {
 					Flag:  model.HandleKeyFlag,
 				},
 			},
-		}, exceptPartition: 5},
+		}, expectPartition: 5},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -111,7 +111,7 @@ func TestIndexValueDispatcher(t *testing.T) {
 					Flag:  model.HandleKeyFlag,
 				},
 			},
-		}, exceptPartition: 5},
+		}, expectPartition: 5},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -128,7 +128,7 @@ func TestIndexValueDispatcher(t *testing.T) {
 					Flag:  model.HandleKeyFlag,
 				},
 			},
-		}, exceptPartition: 14},
+		}, expectPartition: 14},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
@@ -145,10 +145,10 @@ func TestIndexValueDispatcher(t *testing.T) {
 					Flag:  model.HandleKeyFlag,
 				},
 			},
-		}, exceptPartition: 2},
+		}, expectPartition: 2},
 	}
-	p := newIndexValueDispatcher(16)
+	p := NewIndexValueDispatcher(16)
 	for _, tc := range testCases {
-		require.Equal(t, tc.exceptPartition, p.Dispatch(tc.row))
+		require.Equal(t, tc.expectPartition, p.DispatchRowChangedEvent(tc.row))
 	}
 }
diff --git a/cdc/sink/dispatcher/table_test.go b/cdc/sink/dispatcher/partition/table_test.go
similarity index 81%
rename from cdc/sink/dispatcher/table_test.go
rename to cdc/sink/dispatcher/partition/table_test.go
index d0bb11496..d65895e82 100644
--- a/cdc/sink/dispatcher/table_test.go
+++ b/cdc/sink/dispatcher/partition/table_test.go
@@ -1,4 +1,4 @@
-// Copyright 2020 PingCAP, Inc.
+// Copyright 2022 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -11,7 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package dispatcher
+package partition
 
 import (
 	"testing"
@@ -25,7 +25,7 @@ func TestTableDispatcher(t *testing.T) {
 
 	testCases := []struct {
 		row             *model.RowChangedEvent
-		exceptPartition int32
+		expectPartition int32
 	}{
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
@@ -33,52 +33,52 @@ func TestTableDispatcher(t *testing.T) {
 				Table:  "t1",
 			},
 			CommitTs: 1,
-		}, exceptPartition: 15},
+		}, expectPartition: 15},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
 				Table:  "t1",
 			},
 			CommitTs: 2,
-		}, exceptPartition: 15},
+		}, expectPartition: 15},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
 				Table:  "t1",
 			},
 			CommitTs: 3,
-		}, exceptPartition: 15},
+		}, expectPartition: 15},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
 				Table:  "t2",
 			},
 			CommitTs: 1,
-		}, exceptPartition: 5},
+		}, expectPartition: 5},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
 				Table:  "t2",
 			},
 			CommitTs: 2,
-		}, exceptPartition: 5},
+		}, expectPartition: 5},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
 				Table:  "t2",
 			},
 			CommitTs: 3,
-		}, exceptPartition: 5},
+		}, expectPartition: 5},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
 				Table:  "t3",
 			},
 			CommitTs: 3,
-		}, exceptPartition: 3},
+		}, expectPartition: 3},
 	}
-	p := newTableDispatcher(16)
+	p := NewTableDispatcher(16)
 	for _, tc := range testCases {
-		require.Equal(t, tc.exceptPartition, p.Dispatch(tc.row))
+		require.Equal(t, tc.expectPartition, p.DispatchRowChangedEvent(tc.row))
 	}
 }
diff --git a/cdc/sink/dispatcher/ts_test.go b/cdc/sink/dispatcher/partition/ts_test.go
similarity index 81%
rename from cdc/sink/dispatcher/ts_test.go
rename to cdc/sink/dispatcher/partition/ts_test.go
index a5f33019f..145a88bc0 100644
--- a/cdc/sink/dispatcher/ts_test.go
+++ b/cdc/sink/dispatcher/partition/ts_test.go
@@ -1,4 +1,4 @@
-// Copyright 2020 PingCAP, Inc.
+// Copyright 2022 PingCAP, Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -11,7 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package dispatcher
+package partition
 
 import (
 	"testing"
@@ -25,7 +25,7 @@ func TestTsDispatcher(t *testing.T) {
 
 	testCases := []struct {
 		row             *model.RowChangedEvent
-		exceptPartition int32
+		expectPartition int32
 	}{
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
@@ -33,45 +33,45 @@ func TestTsDispatcher(t *testing.T) {
 				Table:  "t1",
 			},
 			CommitTs: 1,
-		}, exceptPartition: 1},
+		}, expectPartition: 1},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
 				Table:  "t1",
 			},
 			CommitTs: 2,
-		}, exceptPartition: 2},
+		}, expectPartition: 2},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
 				Table:  "t1",
 			},
 			CommitTs: 3,
-		}, exceptPartition: 3},
+		}, expectPartition: 3},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
 				Table:  "t2",
 			},
 			CommitTs: 1,
-		}, exceptPartition: 1},
+		}, expectPartition: 1},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
 				Table:  "t2",
 			},
 			CommitTs: 2,
-		}, exceptPartition: 2},
+		}, expectPartition: 2},
 		{row: &model.RowChangedEvent{
 			Table: &model.TableName{
 				Schema: "test",
 				Table:  "t2",
 			},
 			CommitTs: 3,
-		}, exceptPartition: 3},
+		}, expectPartition: 3},
 	}
-	p := &tsDispatcher{partitionNum: 16}
+	p := &TsDispatcher{partitionNum: 16}
 	for _, tc := range testCases {
-		require.Equal(t, tc.exceptPartition, p.Dispatch(tc.row))
+		require.Equal(t, tc.expectPartition, p.DispatchRowChangedEvent(tc.row))
 	}
 }
diff --git a/cdc/sink/dispatcher/switcher_test.go b/cdc/sink/dispatcher/switcher_test.go
deleted file mode 100644
index ca3eeab00..000000000
--- a/cdc/sink/dispatcher/switcher_test.go
+++ /dev/null
@@ -1,83 +0,0 @@
-// Copyright 2020 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 dispatcher
-
-import (
-	"testing"
-
-	"github.com/pingcap/tiflow/cdc/model"
-	"github.com/pingcap/tiflow/pkg/config"
-	"github.com/stretchr/testify/require"
-)
-
-func TestSwitcher(t *testing.T) {
-	t.Parallel()
-
-	d, err := NewDispatcher(config.GetDefaultReplicaConfig(), 4)
-	require.Nil(t, err)
-	require.IsType(t, &defaultDispatcher{}, d.(*dispatcherSwitcher).matchDispatcher(&model.RowChangedEvent{
-		Table: &model.TableName{
-			Schema: "test", Table: "test",
-		},
-	}))
-
-	d, err = NewDispatcher(&config.ReplicaConfig{
-		Sink: &config.SinkConfig{
-			DispatchRules: []*config.DispatchRule{
-				{Matcher: []string{"test_default1.*"}, Dispatcher: "default"},
-				{Matcher: []string{"test_default2.*"}, Dispatcher: "unknown-dispatcher"},
-				{Matcher: []string{"test_table.*"}, Dispatcher: "table"},
-				{Matcher: []string{"test_index_value.*"}, Dispatcher: "index-value"},
-				{Matcher: []string{"test.*"}, Dispatcher: "rowid"},
-				{Matcher: []string{"*.*", "!*.test"}, Dispatcher: "ts"},
-			},
-		},
-	}, 4)
-	require.Nil(t, err)
-	require.IsType(t, &indexValueDispatcher{}, d.(*dispatcherSwitcher).matchDispatcher(&model.RowChangedEvent{
-		Table: &model.TableName{
-			Schema: "test", Table: "table1",
-		},
-	}))
-	require.IsType(t, &tsDispatcher{}, d.(*dispatcherSwitcher).matchDispatcher(&model.RowChangedEvent{
-		Table: &model.TableName{
-			Schema: "sbs", Table: "table2",
-		},
-	}))
-	require.IsType(t, &defaultDispatcher{}, d.(*dispatcherSwitcher).matchDispatcher(&model.RowChangedEvent{
-		Table: &model.TableName{
-			Schema: "sbs", Table: "test",
-		},
-	}))
-	require.IsType(t, &defaultDispatcher{}, d.(*dispatcherSwitcher).matchDispatcher(&model.RowChangedEvent{
-		Table: &model.TableName{
-			Schema: "test_default1", Table: "test",
-		},
-	}))
-	require.IsType(t, &defaultDispatcher{}, d.(*dispatcherSwitcher).matchDispatcher(&model.RowChangedEvent{
-		Table: &model.TableName{
-			Schema: "test_default2", Table: "test",
-		},
-	}))
-	require.IsType(t, &tableDispatcher{}, d.(*dispatcherSwitcher).matchDispatcher(&model.RowChangedEvent{
-		Table: &model.TableName{
-			Schema: "test_table", Table: "test",
-		},
-	}))
-	require.IsType(t, &indexValueDispatcher{}, d.(*dispatcherSwitcher).matchDispatcher(&model.RowChangedEvent{
-		Table: &model.TableName{
-			Schema: "test_index_value", Table: "test",
-		},
-	}))
-}
diff --git a/cdc/sink/dispatcher/topic/dispatcher_test.go b/cdc/sink/dispatcher/topic/dispatcher_test.go
new file mode 100644
index 000000000..80bee694b
--- /dev/null
+++ b/cdc/sink/dispatcher/topic/dispatcher_test.go
@@ -0,0 +1,226 @@
+// Copyright 2022 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 topic
+
+import (
+	"testing"
+
+	timodel "github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/stretchr/testify/require"
+)
+
+func TestStaticTopicDispatcher(t *testing.T) {
+	row := &model.RowChangedEvent{
+		Table: &model.TableName{
+			Schema: "db1",
+			Table:  "tbl1",
+		},
+	}
+
+	ddl := &model.DDLEvent{
+		TableInfo: &model.SimpleTableInfo{
+			Schema: "db1",
+			Table:  "tbl1",
+		},
+	}
+
+	p := NewStaticTopicDispatcher("cdctest")
+	require.Equal(t, p.DispatchRowChangedEvent(row), "cdctest")
+	require.Equal(t, p.DispatchDDLEvent(ddl), "cdctest")
+}
+
+func TestDynamicTopicDispatcherForSchema(t *testing.T) {
+	t.Parallel()
+
+	topicExpr := Expression("hello_{schema}_world")
+	err := topicExpr.Validate()
+	require.Nil(t, err)
+	testCase := []struct {
+		row         *model.RowChangedEvent
+		expectTopic string
+	}{
+		{
+			row: &model.RowChangedEvent{
+				Table: &model.TableName{
+					Schema: "test1",
+					Table:  "tb1",
+				},
+			},
+			expectTopic: "hello_test1_world",
+		},
+		{
+			row: &model.RowChangedEvent{
+				Table: &model.TableName{
+					Schema: "test1",
+					Table:  "tb2",
+				},
+			},
+			expectTopic: "hello_test1_world",
+		},
+		{
+			row: &model.RowChangedEvent{
+				Table: &model.TableName{
+					Schema: "test2",
+					Table:  "tb1",
+				},
+			},
+			expectTopic: "hello_test2_world",
+		},
+		{
+			row: &model.RowChangedEvent{
+				Table: &model.TableName{
+					Schema: "test2",
+					Table:  "tb2",
+				},
+			},
+			expectTopic: "hello_test2_world",
+		},
+	}
+
+	p := NewDynamicTopicDispatcher("cdctest", topicExpr)
+	for _, tc := range testCase {
+		require.Equal(t, tc.expectTopic, p.DispatchRowChangedEvent(tc.row))
+	}
+}
+
+func TestDynamicTopicDispatcherForTable(t *testing.T) {
+	t.Parallel()
+
+	topicExpr := Expression("{schema}_{table}")
+	err := topicExpr.Validate()
+	require.Nil(t, err)
+	testCases := []struct {
+		row           *model.RowChangedEvent
+		expectedTopic string
+	}{
+		{
+			row: &model.RowChangedEvent{
+				Table: &model.TableName{
+					Schema: "test1",
+					Table:  "tb1",
+				},
+			},
+			expectedTopic: "test1_tb1",
+		},
+		{
+			row: &model.RowChangedEvent{
+				Table: &model.TableName{
+					Schema: "test1",
+					Table:  "tb2",
+				},
+			},
+			expectedTopic: "test1_tb2",
+		},
+		{
+			row: &model.RowChangedEvent{
+				Table: &model.TableName{
+					Schema: "test2",
+					Table:  "tb1",
+				},
+			},
+			expectedTopic: "test2_tb1",
+		},
+		{
+			row: &model.RowChangedEvent{
+				Table: &model.TableName{
+					Schema: "test2",
+					Table:  "tb2",
+				},
+			},
+			expectedTopic: "test2_tb2",
+		},
+	}
+	p := NewDynamicTopicDispatcher("cdctest", topicExpr)
+	for _, tc := range testCases {
+		require.Equal(t, tc.expectedTopic, p.DispatchRowChangedEvent(tc.row))
+	}
+}
+
+func TestDynamicTopicDispatcherForDDL(t *testing.T) {
+	t.Parallel()
+
+	defaultTopic := "cdctest"
+	topicExpr := Expression("{schema}_{table}")
+	testCases := []struct {
+		ddl           *model.DDLEvent
+		expectedTopic string
+	}{
+		{
+			ddl: &model.DDLEvent{
+				TableInfo: &model.SimpleTableInfo{
+					Schema: "test",
+				},
+				Type: timodel.ActionCreateSchema,
+			},
+			expectedTopic: defaultTopic,
+		},
+		{
+			ddl: &model.DDLEvent{
+				TableInfo: &model.SimpleTableInfo{
+					Schema: "test",
+				},
+				Type: timodel.ActionDropSchema,
+			},
+			expectedTopic: defaultTopic,
+		},
+		{
+			ddl: &model.DDLEvent{
+				TableInfo: &model.SimpleTableInfo{
+					Schema: "cdc",
+					Table:  "person",
+				},
+				Query: "CREATE TABLE person(id int, name varchar(32), primary key(id))",
+				Type:  timodel.ActionCreateTable,
+			},
+			expectedTopic: "cdc_person",
+		},
+		{
+			ddl: &model.DDLEvent{
+				TableInfo: &model.SimpleTableInfo{
+					Schema: "cdc",
+					Table:  "person",
+				},
+				Query: "DROP TABLE person",
+				Type:  timodel.ActionDropTable,
+			},
+			expectedTopic: "cdc_person",
+		},
+		{
+			ddl: &model.DDLEvent{
+				TableInfo: &model.SimpleTableInfo{
+					Schema: "cdc",
+					Table:  "person",
+				},
+				Query: "ALTER TABLE cdc.person ADD COLUMN age int",
+				Type:  timodel.ActionAddColumn,
+			},
+			expectedTopic: "cdc_person",
+		},
+		{
+			ddl: &model.DDLEvent{
+				TableInfo: &model.SimpleTableInfo{
+					Schema: "",
+					Table:  "",
+				},
+			},
+			expectedTopic: defaultTopic,
+		},
+	}
+
+	p := NewDynamicTopicDispatcher(defaultTopic, topicExpr)
+	for _, tc := range testCases {
+		require.Equal(t, tc.expectedTopic, p.DispatchDDLEvent(tc.ddl))
+	}
+}
diff --git a/pkg/cmd/util/helper_test.go b/pkg/cmd/util/helper_test.go
index 6aee8a019..f0f1d2571 100644
--- a/pkg/cmd/util/helper_test.go
+++ b/pkg/cmd/util/helper_test.go
@@ -186,8 +186,8 @@ func (s *utilsSuite) TestAndWriteExampleReplicaTOML(c *check.C) {
 	})
 	c.Assert(cfg.Sink, check.DeepEquals, &config.SinkConfig{
 		DispatchRules: []*config.DispatchRule{
-			{Dispatcher: "ts", Matcher: []string{"test1.*", "test2.*"}},
-			{Dispatcher: "rowid", Matcher: []string{"test3.*", "test4.*"}},
+			{PartitionRule: "ts", TopicRule: "hello_{schema}", Matcher: []string{"test1.*", "test2.*"}},
+			{PartitionRule: "rowid", TopicRule: "{schema}_world", Matcher: []string{"test3.*", "test4.*"}},
 		},
 		ColumnSelectors: []*config.ColumnSelector{
 			{Matcher: []string{"test1.*", "test2.*"}, Columns: []string{"column1", "column2"}},
diff --git a/pkg/config/replica_config_test.go b/pkg/config/replica_config_test.go
index 1fd6c93fc..75003eb85 100644
--- a/pkg/config/replica_config_test.go
+++ b/pkg/config/replica_config_test.go
@@ -77,9 +77,9 @@ func TestReplicaConfigOutDated(t *testing.T) {
 	conf.Mounter.WorkerNum = 3
 	conf.Sink.Protocol = "open-protocol"
 	conf.Sink.DispatchRules = []*DispatchRule{
-		{Matcher: []string{"a.b"}, Dispatcher: "r1"},
-		{Matcher: []string{"a.c"}, Dispatcher: "r2"},
-		{Matcher: []string{"a.d"}, Dispatcher: "r2"},
+		{Matcher: []string{"a.b"}, PartitionRule: "r1"},
+		{Matcher: []string{"a.c"}, PartitionRule: "r2"},
+		{Matcher: []string{"a.d"}, PartitionRule: "r2"},
 	}
 	require.Equal(t, conf, conf2)
 }