Repository: pingcap/tiflow (go)

Issue:
make the case-sensitive works on the whole dispatchers configuration
### Is your feature request related to a problem?

At the moment, there is one configuration named as  [case-sensitive](https://docs.pingcap.com/tidb/stable/ticdc-changefeed-config/#case-sensitive), which controls the database name and table name specified in the dispatcher part.

Let's take the following as an example, the `columns` specified the columns used to calculate the partition. 

It may be hard for the user to specify columns with case if there are a lot of tables. For example, table's may have different columns name to indicate the same thing, such as `ID`, `Id`, `id`, etc.
```
[sink]
dispatchers = [
    {matcher = ['test2.*'], partition = "columns", columns = ["id", "a"]},
]
```

This should also takes effect on the column-selector configurations
```
[sink]
column-selectors = [
    {matcher = ['test.t1'], columns = ['a', 'b']},
    {matcher = ['test.*'], columns = ["*", "!b"]},
    {matcher = ['test1.t1'], columns = ['column*', '!column1']},
    {matcher = ['test3.t'], columns = ["column?", "!column1"]},
]
```

### Describe the feature you'd like

the `case-sensitive` field, should also take affect on the columns configurations.

### Describe alternatives you've considered

_No response_

### Teachability, Documentation, Adoption, Migration Strategy

After this feature is implemented, should also update the document to mention this.

Pull request description:
sink(ticdc): calculate partition by the orignal column name
<!--
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: close #12103

### What is changed and how it works?
This pr https://github.com/pingcap/tiflow/pull/12132 makes the column and index names case-insensitive in dispatcher. But it also miscalculates the partition with the lowercase column name.
The `ColumnsDispatcher` and `IndexValueDispatcher` now correctly use the original column names for hashing, ensuring consistent and accurate data partitioning.

### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration 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`.
```


Patch:
diff --git a/cdc/model/schema_storage.go b/cdc/model/schema_storage.go
index dad065a01..78ebd1820 100644
--- a/cdc/model/schema_storage.go
+++ b/cdc/model/schema_storage.go
@@ -458,7 +458,7 @@ func (ti *TableInfo) IndexByName(name string) ([]string, []int, bool) {
 	names := make([]string, 0, len(index.Columns))
 	offset := make([]int, 0, len(index.Columns))
 	for _, col := range index.Columns {
-		names = append(names, col.Name.L)
+		names = append(names, col.Name.O)
 		offset = append(offset, col.Offset)
 	}
 	return names, offset, true
diff --git a/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go b/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
index 164a1896f..abaab5327 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
@@ -15,7 +15,6 @@ package partition
 
 import (
 	"strconv"
-	"strings"
 	"sync"
 
 	"github.com/pingcap/log"
@@ -70,7 +69,8 @@ func (r *ColumnsDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent,
 		if col == nil {
 			continue
 		}
-		r.hasher.Write([]byte(strings.ToLower(r.Columns[idx])), []byte(model.ColumnValueString(col.Value)))
+		colInfo := row.TableInfo.ForceGetColumnInfo(col.ColumnID)
+		r.hasher.Write([]byte(colInfo.Name.O), []byte(model.ColumnValueString(col.Value)))
 	}
 
 	sum32 := r.hasher.Sum32()
diff --git a/cdc/sink/dmlsink/mq/dispatcher/partition/index_value.go b/cdc/sink/dmlsink/mq/dispatcher/partition/index_value.go
index c581a4355..272d21829 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/partition/index_value.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/partition/index_value.go
@@ -15,7 +15,6 @@ package partition
 
 import (
 	"strconv"
-	"strings"
 	"sync"
 
 	"github.com/pingcap/log"
@@ -62,7 +61,7 @@ func (r *IndexValueDispatcher) DispatchRowChangedEvent(row *model.RowChangedEven
 				continue
 			}
 			if tableInfo.ForceGetColumnFlagType(col.ColumnID).IsHandleKey() {
-				r.hasher.Write([]byte(strings.ToLower(tableInfo.ForceGetColumnName(col.ColumnID))), []byte(model.ColumnValueString(col.Value)))
+				r.hasher.Write([]byte(tableInfo.ForceGetColumnName(col.ColumnID)), []byte(model.ColumnValueString(col.Value)))
 			}
 		}
 	} else {


Test patch:
diff --git a/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go b/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go
index a0aa46c6e..52adf2adc 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/partition/columns_test.go
@@ -56,8 +56,19 @@ func TestColumnsDispatcher(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, int32(15), index)
 
+	idx := index
 	p = NewColumnsDispatcher([]string{"COL2", "Col1"})
 	index, _, err = p.DispatchRowChangedEvent(event, 16)
 	require.NoError(t, err)
-	require.Equal(t, int32(15), index)
+	require.Equal(t, idx, index)
+
+	event.TableInfo.Columns = []*timodel.ColumnInfo{
+		{ID: 1, Name: pmodel.NewCIStr("COL2"), Offset: 1, FieldType: *types.NewFieldType(mysql.TypeLong)},
+		{ID: 2, Name: pmodel.NewCIStr("Col1"), Offset: 0, FieldType: *types.NewFieldType(mysql.TypeLong)},
+		{ID: 3, Name: pmodel.NewCIStr("col3"), Offset: 2, FieldType: *types.NewFieldType(mysql.TypeLong)},
+	}
+	p = NewColumnsDispatcher([]string{"col2", "col1"})
+	index, _, err = p.DispatchRowChangedEvent(event, 16)
+	require.NoError(t, err)
+	require.Equal(t, int32(5), index)
 }
diff --git a/cdc/sink/dmlsink/mq/dispatcher/partition/index_value_test.go b/cdc/sink/dmlsink/mq/dispatcher/partition/index_value_test.go
index 36b5edbec..b5724ba0a 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/partition/index_value_test.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/partition/index_value_test.go
@@ -151,7 +151,9 @@ func TestIndexValueDispatcherWithIndexName(t *testing.T) {
 		Name:       pmodel.NewCIStr("t1"),
 		PKIsHandle: true,
 		Columns: []*timodel.ColumnInfo{
-			{ID: 1, Name: pmodel.NewCIStr("A"), FieldType: *types.NewFieldType(mysql.TypeLong)},
+			{ID: 1, Name: pmodel.NewCIStr("col2"), Offset: 1, FieldType: *types.NewFieldType(mysql.TypeLong)},
+			{ID: 2, Name: pmodel.NewCIStr("col1"), Offset: 0, FieldType: *types.NewFieldType(mysql.TypeLong)},
+			{ID: 3, Name: pmodel.NewCIStr("col3"), Offset: 2, FieldType: *types.NewFieldType(mysql.TypeLong)},
 		},
 		Indices: []*timodel.IndexInfo{
 			{
@@ -159,7 +161,10 @@ func TestIndexValueDispatcherWithIndexName(t *testing.T) {
 				Name:    pmodel.NewCIStr("index1"),
 				Columns: []*timodel.IndexColumn{
 					{
-						Name: pmodel.NewCIStr("A"),
+						Name: pmodel.NewCIStr("col2"), Offset: 1,
+					},
+					{
+						Name: pmodel.NewCIStr("col1"), Offset: 0,
 					},
 				},
 			},
@@ -171,6 +176,8 @@ func TestIndexValueDispatcherWithIndexName(t *testing.T) {
 		TableInfo: tableInfo,
 		Columns: []*model.ColumnData{
 			{ColumnID: 1, Value: 11},
+			{ColumnID: 2, Value: 22},
+			{ColumnID: 3, Value: 33},
 		},
 	}
 
@@ -181,15 +188,16 @@ func TestIndexValueDispatcherWithIndexName(t *testing.T) {
 	p = NewIndexValueDispatcher("index1")
 	index, _, err := p.DispatchRowChangedEvent(event, 16)
 	require.NoError(t, err)
-	require.Equal(t, int32(2), index)
+	require.Equal(t, int32(15), index)
 
+	idx := index
 	p = NewIndexValueDispatcher("INDEX1")
 	index, _, err = p.DispatchRowChangedEvent(event, 16)
 	require.NoError(t, err)
-	require.Equal(t, int32(2), index)
+	require.Equal(t, idx, index)
 
 	p = NewIndexValueDispatcher("")
-	index, _, err = p.DispatchRowChangedEvent(event, 3)
+	index, _, err = p.DispatchRowChangedEvent(event, 16)
 	require.NoError(t, err)
-	require.Equal(t, int32(0), index)
+	require.Equal(t, idx, index)
 }