syncer(dm): centralize row image layout mapping

dm/syncer/data_validator.go

@@ -858,27 +858,17 @@ func (v *DataValidator) genValidateTableInfo(sourceTable *filter.Table, columnCo
 // tableInfoForVisibleColumnCount returns the source TableInfo layout that matches
 // a binlog row image with columnCount visible columns.
 func tableInfoForVisibleColumnCount(tableInfo *model.TableInfo, columnCount int) (*model.TableInfo, bool) {
-	visibleCount := 0
-	stripColumnCount := len(tableInfo.Columns)
-	for i, col := range tableInfo.Columns {
-		if col.Hidden {
-			continue
-		}
-		visibleCount++
-		if visibleCount > columnCount {
-			stripColumnCount = i
-			break
-		}
-	}
-	if visibleCount < columnCount {
+	layout := sqlmodel.NewRowImageLayoutFromColumns(tableInfo.Columns, nil)
+	sourceColumnCount, ok := layout.SourceColumnCountForVisibleColumnCount(columnCount)
+	if !ok {
 		return nil, false
 	}
-	if visibleCount == columnCount {
+	if sourceColumnCount == len(tableInfo.Columns) {
 		return tableInfo, true
 	}
 
 	clone := tableInfo.Clone()
-	clone.Columns = clone.Columns[:stripColumnCount]
+	clone.Columns = clone.Columns[:sourceColumnCount]
 	return clone, true
 }
 

dm/syncer/expr_filter_group.go

@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
+	"github.com/pingcap/tiflow/pkg/sqlmodel"
 	"go.uber.org/zap"
 )
 
@@ -193,13 +194,10 @@ func rowForExpressionFilter(
 
 	values := row
 	if len(row) != len(upstreamCols) {
-		visibleCols := visibleColumns(upstreamCols)
-		if len(row) != len(visibleCols) {
-			return chunk.Row{}, terror.ErrSyncerUnitDMLColumnNotMatch.Generate(len(visibleCols), len(row))
-		}
-		fullValues := make([]interface{}, len(upstreamCols))
-		for i, col := range visibleCols {
-			fullValues[col.Offset] = row[i]
+		layout := sqlmodel.NewRowImageLayoutFromColumns(upstreamCols, nil)
+		fullValues, ok := layout.FullValues(row)
+		if !ok {
+			return chunk.Row{}, terror.ErrSyncerUnitDMLColumnNotMatch.Generate(layout.VisibleColumnCount(), len(row))
 		}
 		values = fullValues
 	}

dm/syncer/validate_worker.go

@@ -320,7 +320,7 @@ func (vw *validateWorker) batchValidateRowChanges(rows []*rowValidationJob, dele
 	firstRow := rows[0].row
 	cond := &Cond{
 		TargetTbl: firstRow.TargetTableID(),
-		Columns:   visibleColumns(firstRow.SourceTableInfo().Columns),
+		Columns:   sqlmodel.VisibleColumns(firstRow.SourceTableInfo().Columns),
 		PK:        firstRow.UniqueNotNullIdx(),
 		PkValues:  pkValues,
 	}
@@ -453,18 +453,6 @@ func (vw *validateWorker) resetErrorRows() {
 	vw.errorRows = make([]*validateFailedRow, 0)
 }
 
-// TODO(joechenrh): Share visible-row/source-column offset mapping with sqlmodel
-// instead of maintaining validator-specific hidden column helpers.
-func visibleColumns(columns []*model.ColumnInfo) []*model.ColumnInfo {
-	ret := make([]*model.ColumnInfo, 0, len(columns))
-	for _, col := range columns {
-		if !col.Hidden {
-			ret = append(ret, col)
-		}
-	}
-	return ret
-}
-
 func (vw *validateWorker) newJobAdded(job *rowValidationJob) {
 	tp := job.Tp
 	vw.pendingRowCounts[tp]++

pkg/sqlmodel/row_change.go

@@ -184,13 +184,7 @@ func (r *RowChange) TargetTableID() string {
 // TiDB TableInfo contains some internal columns like expression index, they
 // are not included in this count.
 func (r *RowChange) ColumnCount() int {
-	c := 0
-	for _, col := range r.sourceTableInfo.Columns {
-		if !col.Hidden {
-			c++
-		}
-	}
-	return c
+	return VisibleColumnCount(r.sourceTableInfo.Columns)
 }
 
 // SourceTableInfo returns the TableInfo of source table.
@@ -242,15 +236,13 @@ func (r *RowChange) lazyInitWhereHandle() {
 // dmlRowMapping returns DML column/value mapping without initializing the full
 // where handle. CDC insert/update paths can use this without paying the WHERE
 // handle build cost, while DM paths reuse the injected cached handle.
-func (r *RowChange) dmlRowMapping() (rowValueMapper, []*timodel.ColumnInfo) {
+func (r *RowChange) dmlRowMapping() (RowImageLayout, []*timodel.ColumnInfo) {
 	if r.whereHandle != nil {
-		return r.whereHandle.rowMapper, r.whereHandle.writableColumns
+		return r.whereHandle.rowMapper, r.whereHandle.rowMapper.WritableColumns()
 	}
 
-	// Keep this fallback mapping consistent with GetWhereHandle until the row
-	// image mapping is centralized.
-	rowMapper := newRowValueMapper(r.sourceTableInfo.Columns)
-	return rowMapper, writableSourceColumns(rowMapper.visibleColumns, r.targetTableInfo.Columns)
+	layout := NewRowImageLayout(r.sourceTableInfo, r.targetTableInfo)
+	return layout, layout.WritableColumns()
 }
 
 // whereColumnsAndValues returns columns and values to identify the row, to form

pkg/sqlmodel/row_image_layout.go

@@ -0,0 +1,173 @@
+// Copyright 2026 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 sqlmodel
+
+import "github.com/pingcap/tidb/pkg/meta/model"
+
+// RowImageLayout describes how a binlog row image maps to source table columns.
+type RowImageLayout struct {
+	columns                     []*model.ColumnInfo
+	visibleColumns              []*model.ColumnInfo
+	visibleOffsetByColumnOffset []int
+	writableColumns             []*model.ColumnInfo
+}
+
+// NewRowImageLayout creates a RowImageLayout from source and target TableInfo.
+func NewRowImageLayout(source, target *model.TableInfo) RowImageLayout {
+	return NewRowImageLayoutFromColumns(source.Columns, target.Columns)
+}
+
+// NewRowImageLayoutFromColumns creates a RowImageLayout from source and target columns.
+func NewRowImageLayoutFromColumns(sourceColumns, targetColumns []*model.ColumnInfo) RowImageLayout {
+	visibleColumns := VisibleColumns(sourceColumns)
+	visibleOffsetByColumnOffset := make([]int, len(sourceColumns))
+	for i := range visibleOffsetByColumnOffset {
+		visibleOffsetByColumnOffset[i] = -1
+	}
+	for i, column := range visibleColumns {
+		visibleOffsetByColumnOffset[column.Offset] = i
+	}
+
+	layout := RowImageLayout{
+		columns:                     sourceColumns,
+		visibleColumns:              visibleColumns,
+		visibleOffsetByColumnOffset: visibleOffsetByColumnOffset,
+	}
+	if targetColumns != nil {
+		layout.writableColumns = writableSourceColumns(visibleColumns, targetColumns)
+	}
+	return layout
+}
+
+// VisibleColumns returns the visible columns from the given table columns.
+func VisibleColumns(columns []*model.ColumnInfo) []*model.ColumnInfo {
+	ret := make([]*model.ColumnInfo, 0, len(columns))
+	for _, col := range columns {
+		if !col.Hidden {
+			ret = append(ret, col)
+		}
+	}
+	return ret
+}
+
+// VisibleColumnCount returns the number of visible columns.
+func VisibleColumnCount(columns []*model.ColumnInfo) int {
+	count := 0
+	for _, col := range columns {
+		if !col.Hidden {
+			count++
+		}
+	}
+	return count
+}
+
+// VisibleColumns returns the visible source columns in the row image.
+func (l RowImageLayout) VisibleColumns() []*model.ColumnInfo {
+	return l.visibleColumns
+}
+
+// VisibleColumnCount returns the visible source column count in the row image.
+func (l RowImageLayout) VisibleColumnCount() int {
+	return len(l.visibleColumns)
+}
+
+// WritableColumns returns visible source columns that can be written to the target table.
+func (l RowImageLayout) WritableColumns() []*model.ColumnInfo {
+	return l.writableColumns
+}
+
+// FullValues expands a visible-only row image to the source table column layout.
+func (l RowImageLayout) FullValues(row []any) ([]any, bool) {
+	if len(row) == len(l.columns) {
+		return row, true
+	}
+	if len(row) != len(l.visibleColumns) {
+		return nil, false
+	}
+
+	fullValues := make([]any, len(l.columns))
+	for i, col := range l.visibleColumns {
+		fullValues[col.Offset] = row[i]
+	}
+	return fullValues, true
+}
+
+// SourceColumnCountForVisibleColumnCount returns the source-column prefix width
+// that contains exactly columnCount visible columns.
+func (l RowImageLayout) SourceColumnCountForVisibleColumnCount(columnCount int) (int, bool) {
+	visibleCount := 0
+	sourceColumnCount := len(l.columns)
+	for i, col := range l.columns {
+		if col.Hidden {
+			continue
+		}
+		visibleCount++
+		if visibleCount > columnCount {
+			sourceColumnCount = i
+			break
+		}
+	}
+	if visibleCount < columnCount {
+		return 0, false
+	}
+	return sourceColumnCount, true
+}
+
+func (l RowImageLayout) isFullValues(values []any) bool {
+	return len(values) == len(l.columns)
+}
+
+func (l RowImageLayout) columnsForValues(values []any) []*model.ColumnInfo {
+	if l.isFullValues(values) {
+		return l.columns
+	}
+	return l.visibleColumns
+}
+
+func (l RowImageLayout) columnsAndValuesByIndex(
+	indexInfo *model.IndexInfo,
+	values []any,
+) ([]*model.ColumnInfo, []any) {
+	cols := make([]*model.ColumnInfo, 0, len(indexInfo.Columns))
+	vals := make([]any, 0, len(indexInfo.Columns))
+	for _, column := range indexInfo.Columns {
+		offset := l.valueOffset(column.Offset, values)
+		cols = append(cols, l.columns[column.Offset])
+		vals = append(vals, values[offset])
+	}
+	return cols, vals
+}
+
+func (l RowImageLayout) valuesByIndex(indexInfo *model.IndexInfo, values []any) []any {
+	ret := make([]any, 0, len(indexInfo.Columns))
+	if values == nil {
+		return ret
+	}
+	for _, column := range indexInfo.Columns {
+		offset := l.valueOffset(column.Offset, values)
+		ret = append(ret, values[offset])
+	}
+	return ret
+}
+
+func (l RowImageLayout) valueOffset(columnOffset int, values []any) int {
+	if l.isFullValues(values) {
+		return columnOffset
+	}
+	return l.visibleOffsetByColumnOffset[columnOffset]
+}
+
+func (l RowImageLayout) valueByOffset(columnOffset int, values []any) any {
+	return values[l.valueOffset(columnOffset, values)]
+}

pkg/sqlmodel/row_image_layout_test.go

@@ -0,0 +1,60 @@
+// Copyright 2026 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 sqlmodel
+
+import (
+	"testing"
+
+	timodel "github.com/pingcap/tidb/pkg/meta/model"
+	"github.com/pingcap/tiflow/pkg/util/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestRowImageLayoutWithInterleavedHiddenColumn(t *testing.T) {
+	t.Parallel()
+
+	tableInfo := mockTableInfo(t, "CREATE TABLE t ("+
+		"id INT PRIMARY KEY, a INT, b INT, UNIQUE KEY idx_expr ((a + b)))")
+	hidden := testutil.HiddenColumnName(t, tableInfo)
+	testutil.ReorderColumnsByName(t, tableInfo, "id", "a", hidden, "b")
+
+	layout := NewRowImageLayout(tableInfo, tableInfo)
+	require.Equal(t, 3, layout.VisibleColumnCount())
+	require.Equal(t, []string{"id", "a", "b"}, columnNames(layout.VisibleColumns()))
+	require.Equal(t, 2, layout.valueOffset(3, []any{1, 2, 3}))
+	require.Equal(t, 3, layout.valueByOffset(3, []any{1, 2, 3}))
+
+	fullValues, ok := layout.FullValues([]any{1, 2, 3})
+	require.True(t, ok)
+	require.Equal(t, []any{1, 2, nil, 3}, fullValues)
+
+	sourceColumnCount, ok := layout.SourceColumnCountForVisibleColumnCount(2)
+	require.True(t, ok)
+	require.Equal(t, 3, sourceColumnCount)
+
+	sourceColumnCount, ok = layout.SourceColumnCountForVisibleColumnCount(3)
+	require.True(t, ok)
+	require.Equal(t, len(tableInfo.Columns), sourceColumnCount)
+
+	_, ok = layout.SourceColumnCountForVisibleColumnCount(4)
+	require.False(t, ok)
+}
+
+func columnNames(columns []*timodel.ColumnInfo) []string {
+	names := make([]string, 0, len(columns))
+	for _, column := range columns {
+		names = append(names, column.Name.L)
+	}
+	return names
+}