move dispatcher validation

Author: a-cong

cdc/sink/dispatcher/dispatcher.go

@@ -16,19 +16,23 @@ package dispatcher
 import (
 	"fmt"
 	"regexp"
+
+	"github.com/pingcap/tiflow/pkg/config"
 )
 
 var (
 	// schemaRE is used to match substring '{schema}' in expressions
 	schemaRE = regexp.MustCompile(`\{schema\}`)
 	// tableRE is used to match substring '{table}' in expressions
 	tableRE = regexp.MustCompile(`\{table\}`)
-	// validExprRE validates schema/table routing expressions
-	// Expression can contain {schema}, {table}, or both, with valid separators
-	// Valid characters: A-Za-z0-9\._\-
-	validExprRE = regexp.MustCompile(`^[A-Za-z0-9\._\-]*(\{schema\})?[A-Za-z0-9\._\-]*(\{table\})?[A-Za-z0-9\._\-]*$`)
 )
 
+// ValidateExpression validates a schema or table routing expression.
+// This is a convenience wrapper around config.ValidateRoutingExpression.
+func ValidateExpression(expr string) error {
+	return config.ValidateRoutingExpression(expr)
+}
+
 // Dispatcher is an abstraction for routing schema and table names.
 // It determines the target schema and table name for a given source schema/table.
 type Dispatcher interface {
@@ -87,20 +91,3 @@ func (d *DynamicSchemaDispatcher) substituteExpression(expr, sourceSchema, sourc
 func (d *DynamicSchemaDispatcher) String() string {
 	return fmt.Sprintf("dynamic(schema:%s,table:%s)", d.schemaExpr, d.tableExpr)
 }
-
-// ValidateExpression validates a schema or table routing expression.
-// Expression format: [prefix][{schema}][middle][{table}][suffix]
-// prefix/middle/suffix are optional and should match [A-Za-z0-9\._\-]*
-func ValidateExpression(expr string) error {
-	if expr == "" {
-		// Empty is valid - means use source schema/table
-		return nil
-	}
-
-	if !validExprRE.MatchString(expr) {
-		return fmt.Errorf("invalid schema/table routing expression: %s. "+
-			"Expression must match pattern: [prefix][{schema}][middle][{table}][suffix]", expr)
-	}
-
-	return nil
-}

pkg/config/replica_config.go

@@ -325,28 +325,21 @@ func (c *ReplicaConfig) ValidateAndAdjust(sinkURI *url.URL) error { // check sin
 
 	// Validate sink routing configuration (dispatcher approach)
 	// Check if any dispatch rules have schema/table routing configured
-	hasSchemaRouting := false
 	for _, rule := range c.Sink.DispatchRules {
-		if rule.SchemaRule != "" || rule.TableRule != "" {
-			hasSchemaRouting = true
-			break
+		if rule.SchemaRule == "" && rule.TableRule == "" {
+			continue
 		}
-	}
-	if hasSchemaRouting {
 		// Schema routing is only supported for MySQL/TiDB sinks
 		if !sink.IsMySQLCompatibleScheme(sinkURI.Scheme) {
 			return cerror.ErrInvalidReplicaConfig.GenWithStackByArgs(
 				"sink routing (via dispatch rules) is only supported for MySQL/TiDB sinks")
 		}
-		// Validate by attempting to create the sink router
-		// This will validate the expressions and matcher patterns
-		// We don't need to keep the router, just validate it can be created
-		if _, err := func() (interface{}, error) {
-			// Import the dispatcher package dynamically to avoid import cycle
-			// The actual validation happens in dispatcher.NewSinkRouter
-			return nil, nil
-		}(); err != nil {
-			return err
+		// Validate the schema and table expressions
+		if err := ValidateRoutingExpression(rule.SchemaRule); err != nil {
+			return cerror.WrapError(cerror.ErrInvalidReplicaConfig, err)
+		}
+		if err := ValidateRoutingExpression(rule.TableRule); err != nil {
+			return cerror.WrapError(cerror.ErrInvalidReplicaConfig, err)
 		}
 	}
 

pkg/config/replica_config_test.go

@@ -383,6 +383,62 @@ func TestValidateAndAdjustLargeMessageHandle(t *testing.T) {
 	require.Equal(t, compression.None, cfg.Sink.KafkaConfig.LargeMessageHandle.LargeMessageHandleCompression)
 }
 
+func TestValidateAndAdjustSinkRouting(t *testing.T) {
+	t.Parallel()
+
+	// Test valid routing rules on MySQL sink
+	cfg := GetDefaultReplicaConfig()
+	cfg.Sink.DispatchRules = []*DispatchRule{
+		{Matcher: []string{"db1.*"}, SchemaRule: "target_db", TableRule: "{table}"},
+	}
+	mysqlURL, err := url.Parse("mysql://root:123@localhost:3306/")
+	require.NoError(t, err)
+	require.NoError(t, cfg.ValidateAndAdjust(mysqlURL))
+
+	// Test valid routing rules with transformations
+	cfg = GetDefaultReplicaConfig()
+	cfg.Sink.DispatchRules = []*DispatchRule{
+		{Matcher: []string{"db1.*"}, SchemaRule: "backup_{schema}", TableRule: "{table}_archive"},
+	}
+	require.NoError(t, cfg.ValidateAndAdjust(mysqlURL))
+
+	// Test invalid SchemaRule expression
+	cfg = GetDefaultReplicaConfig()
+	cfg.Sink.DispatchRules = []*DispatchRule{
+		{Matcher: []string{"db1.*"}, SchemaRule: "{invalid}", TableRule: "{table}"},
+	}
+	err = cfg.ValidateAndAdjust(mysqlURL)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "invalid schema/table routing expression")
+
+	// Test invalid TableRule expression with space
+	cfg = GetDefaultReplicaConfig()
+	cfg.Sink.DispatchRules = []*DispatchRule{
+		{Matcher: []string{"db1.*"}, SchemaRule: "{schema}", TableRule: "invalid space"},
+	}
+	err = cfg.ValidateAndAdjust(mysqlURL)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "invalid schema/table routing expression")
+
+	// Test routing rules on non-MySQL sink (should fail)
+	cfg = GetDefaultReplicaConfig()
+	cfg.Sink.DispatchRules = []*DispatchRule{
+		{Matcher: []string{"db1.*"}, SchemaRule: "target_db", TableRule: "{table}"},
+	}
+	kafkaURL, err := url.Parse("kafka://localhost:9092/topic?protocol=open-protocol")
+	require.NoError(t, err)
+	err = cfg.ValidateAndAdjust(kafkaURL)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), "sink routing (via dispatch rules) is only supported for MySQL/TiDB sinks")
+
+	// Test dispatch rules without routing (TopicRule only) on Kafka - should pass
+	cfg = GetDefaultReplicaConfig()
+	cfg.Sink.DispatchRules = []*DispatchRule{
+		{Matcher: []string{"db1.*"}, TopicRule: "topic_{schema}_{table}"},
+	}
+	require.NoError(t, cfg.ValidateAndAdjust(kafkaURL))
+}
+
 func TestMaskSensitiveData(t *testing.T) {
 	config := ReplicaConfig{
 		Sink:       nil,

pkg/config/sink.go

@@ -16,6 +16,7 @@ package config
 import (
 	"fmt"
 	"net/url"
+	"regexp"
 	"strconv"
 	"strings"
 	"time"
@@ -408,6 +409,28 @@ type DispatchRule struct {
 	TableRule string `toml:"table" json:"table"`
 }
 
+// validRoutingExprRE validates schema/table routing expressions.
+// Expression can contain {schema}, {table}, or both, with valid separators.
+// Valid characters: A-Za-z0-9\._\-
+var validRoutingExprRE = regexp.MustCompile(`^[A-Za-z0-9\._\-]*(\{schema\})?[A-Za-z0-9\._\-]*(\{table\})?[A-Za-z0-9\._\-]*$`)
+
+// ValidateRoutingExpression validates a schema or table routing expression.
+// Expression format: [prefix][{schema}][middle][{table}][suffix]
+// prefix/middle/suffix are optional and should match [A-Za-z0-9\._\-]*
+func ValidateRoutingExpression(expr string) error {
+	if expr == "" {
+		// Empty is valid - means use source schema/table
+		return nil
+	}
+
+	if !validRoutingExprRE.MatchString(expr) {
+		return fmt.Errorf("invalid schema/table routing expression: %s. "+
+			"Expression must match pattern: [prefix][{schema}][middle][{table}][suffix]", expr)
+	}
+
+	return nil
+}
+
 // ColumnSelector represents a column selector for a table.
 type ColumnSelector struct {
 	Matcher []string `toml:"matcher" json:"matcher"`