diff --git a/go/vt/vttablet/tabletmanager/vdiff/framework_test.go b/go/vt/vttablet/tabletmanager/vdiff/framework_test.go index 80f5062328f..3ff0afbf498 100644 --- a/go/vt/vttablet/tabletmanager/vdiff/framework_test.go +++ b/go/vt/vttablet/tabletmanager/vdiff/framework_test.go @@ -23,6 +23,7 @@ import ( "io" "os" "regexp" + "slices" "strings" "sync" "testing" @@ -112,6 +113,16 @@ var ( Name: "nopkwithpke", Columns: []string{"c1", "c2", "c3"}, Fields: sqltypes.MakeTestFields("c1|c2|c3", "int64|int64|int64"), + }, { + // t2 is a cross-table MoveTables source: its physical PK is c0, + // which is renamed to the target's c1 by the filter + // "select c0 as c1, c2 from t2". getSourcePKCols resolves the + // source schema from the FROM table (t2) and matches the + // physical PK c0 via its underlying ColName. + Name: "t2", + Columns: []string{"c0", "c2"}, + PrimaryKeyColumns: []string{"c0"}, + Fields: sqltypes.MakeTestFields("c0|c2", "int64|int64"), }, }, } @@ -124,6 +135,7 @@ var ( "datze": 5, "nopk": 6, "nopkwithpke": 7, + "t2": 8, } ) @@ -435,19 +447,21 @@ func (dbc *realDBClient) SupportsCapability(capability capabilities.FlavorCapabi type fakeTMClient struct { tmclient.TabletManagerClient - schema *tabletmanagerdatapb.SchemaDefinition - vrQueries map[int]map[string]*querypb.QueryResult - waitpos map[int]string - vrpos map[int]string - pos map[int]string + schema *tabletmanagerdatapb.SchemaDefinition + vrQueries map[int]map[string]*querypb.QueryResult + waitpos map[int]string + vrpos map[int]string + pos map[int]string + pkeResults map[string]*sqltypes.Result } func newFakeTMClient() *fakeTMClient { return &fakeTMClient{ - vrQueries: make(map[int]map[string]*querypb.QueryResult), - waitpos: make(map[int]string), - vrpos: make(map[int]string), - pos: make(map[int]string), + vrQueries: make(map[int]map[string]*querypb.QueryResult), + waitpos: make(map[int]string), + vrpos: make(map[int]string), + pos: make(map[int]string), + pkeResults: make(map[string]*sqltypes.Result), } } @@ -455,7 +469,28 @@ func newFakeTMClient() *fakeTMClient { func (tmc *fakeTMClient) Close() {} func (tmc *fakeTMClient) GetSchema(ctx context.Context, tablet *topodatapb.Tablet, request *tabletmanagerdatapb.GetSchemaRequest) (*tabletmanagerdatapb.SchemaDefinition, error) { - return tmc.schema, nil + if len(request.Tables) == 0 { + return tmc.schema, nil + } + filtered := &tabletmanagerdatapb.SchemaDefinition{ + TableDefinitions: make([]*tabletmanagerdatapb.TableDefinition, 0), + } + for _, td := range tmc.schema.TableDefinitions { + if slices.Contains(request.Tables, td.Name) { + filtered.TableDefinitions = append(filtered.TableDefinitions, td) + } + } + return filtered, nil +} + +func (tmc *fakeTMClient) ExecuteFetchAsApp(ctx context.Context, tablet *topodatapb.Tablet, usePool bool, req *tabletmanagerdatapb.ExecuteFetchAsAppRequest) (*querypb.QueryResult, error) { + query := string(req.Query) + for tableName, result := range tmc.pkeResults { + if strings.Contains(query, tableName) { + return sqltypes.ResultToProto3(result), nil + } + } + return sqltypes.ResultToProto3(noResults), nil } // setVRResults allows you to specify VReplicationExec queries and their results. You can specify @@ -564,6 +599,10 @@ func newTestVDiffEnv(t *testing.T) *testVDiffEnv { vdiffenv.vre.Open(t.Context()) vdiffenv.tmc.schema = testSchema + vdiffenv.tmc.pkeResults["nopkwithpke"] = sqltypes.MakeTestResult( + sqltypes.MakeTestFields("column_name|index_name", "varchar|varchar"), + "c3|c3", + ) // We need to add t1, which we use for a full VDiff in TestVDiff, to // the schema engine with the PK val. st := &schema.Table{ diff --git a/go/vt/vttablet/tabletmanager/vdiff/schema.go b/go/vt/vttablet/tabletmanager/vdiff/schema.go index f87bd9cd970..27991b0534f 100644 --- a/go/vt/vttablet/tabletmanager/vdiff/schema.go +++ b/go/vt/vttablet/tabletmanager/vdiff/schema.go @@ -64,6 +64,7 @@ const ( sqlUpdateTableState = "update _vt.vdiff_table set state = %a where vdiff_id = %a and table_name = %a" sqlUpdateTableStateAndReport = "update _vt.vdiff_table set state = %a, rows_compared = %a, report = %a where vdiff_id = %a and table_name = %a" sqlUpdateTableMismatch = "update _vt.vdiff_table set mismatch = true where vdiff_id = %a and table_name = %a" + sqlClearTableMismatch = "update _vt.vdiff_table set mismatch = false where vdiff_id = %a and table_name = %a" sqlGetIncompleteTables = "select table_name as table_name from _vt.vdiff_table where vdiff_id = %a and state != 'completed' order by table_name" ) diff --git a/go/vt/vttablet/tabletmanager/vdiff/source_pk_cols_ordering_test.go b/go/vt/vttablet/tabletmanager/vdiff/source_pk_cols_ordering_test.go new file mode 100644 index 00000000000..677481bf185 --- /dev/null +++ b/go/vt/vttablet/tabletmanager/vdiff/source_pk_cols_ordering_test.go @@ -0,0 +1,372 @@ +/* +Copyright 2026 The Vitess Authors. + +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, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package vdiff + +import ( + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "vitess.io/vitess/go/sqltypes" + "vitess.io/vitess/go/vt/sqlparser" + + tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata" +) + +// TestSourcePKColsOrdering verifies that getSourcePKCols maps PK columns +// against the source query's SELECT expression order (the actual row layout), +// not against td.table.Columns (column ordinal position). +// +// Before the fix, the function mapped PK columns against td.table.Columns, +// producing indices in ordinal order. When the source query reorders columns +// (e.g., "select c2, c1 from t1"), lastPKFromRow would index into the wrong +// position, corrupting the resume checkpoint. +func TestSourcePKColsOrdering(t *testing.T) { + tvde := newTestVDiffEnv(t) + defer tvde.close() + + ct := tvde.createController(t, 1) + + testCases := []struct { + name string + // sourceTable is the physical table the source query reads from + // (the table in the FROM clause). getSourcePKCols resolves the source + // schema from this name, which can differ from the target table name. + sourceTable string + columns []string + primaryKeyColumns []string + sourceQuery string + wantSourcePkCols []int + }{ + { + name: "columns in natural order", + sourceTable: "t", + columns: []string{"c1", "c2"}, + primaryKeyColumns: []string{"c1"}, + sourceQuery: "select c1, c2 from t order by c1 asc", + wantSourcePkCols: []int{0}, + }, + { + name: "columns reordered in select", + sourceTable: "t", + columns: []string{"c1", "c2"}, + primaryKeyColumns: []string{"c1"}, + sourceQuery: "select c2, c1 from t order by c1 asc", + wantSourcePkCols: []int{1}, + }, + { + name: "composite pk reordered in select", + sourceTable: "t", + columns: []string{"a", "b", "c"}, + primaryKeyColumns: []string{"c", "a"}, + sourceQuery: "select a, b, c from t order by c asc, a asc", + wantSourcePkCols: []int{2, 0}, + }, + { + name: "composite pk with select reorder", + sourceTable: "t", + columns: []string{"a", "b", "c", "d"}, + primaryKeyColumns: []string{"b", "d"}, + sourceQuery: "select d, c, b, a from t order by b asc, d asc", + wantSourcePkCols: []int{2, 0}, + }, + { + // Cross-table MoveTables filter: the source table is t2 and its + // physical PK is c0, renamed to the target column c1 in the SELECT. + // The source PK c0 is matched via its underlying ColName. + name: "renamed physical source column resolves from source table", + sourceTable: "t2", + columns: []string{"c0", "c2"}, + primaryKeyColumns: []string{"c0"}, + sourceQuery: "select c0 as c1, c2 from t2 order by c1 asc", + wantSourcePkCols: []int{0}, + }, + { + name: "pk matches ordinal order", + sourceTable: "t", + columns: []string{"a", "b", "c"}, + primaryKeyColumns: []string{"a", "b"}, + sourceQuery: "select a, b, c from t order by a asc, b asc", + wantSourcePkCols: []int{0, 1}, + }, + { + name: "column swap alias does not shadow real pk", + sourceTable: "t", + columns: []string{"a", "b"}, + primaryKeyColumns: []string{"a"}, + sourceQuery: "select b as a, a as b from t order by a asc", + wantSourcePkCols: []int{1}, + }, + { + name: "column swap composite pk prefers colname over alias", + sourceTable: "t", + columns: []string{"a", "b", "c"}, + primaryKeyColumns: []string{"a", "b"}, + sourceQuery: "select b as a, a as b, c from t order by a asc, b asc", + wantSourcePkCols: []int{1, 0}, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + types := make([]string, len(tc.columns)) + for i := range types { + types[i] = "varbinary" + } + fieldTypes := strings.Join(types, "|") + fields := strings.Join(tc.columns, "|") + + // The source schema is keyed by the physical source table name so + // that getSourcePKCols resolves it from the query's FROM clause. + sourceTable := &tabletmanagerdatapb.TableDefinition{ + Name: tc.sourceTable, + Columns: tc.columns, + PrimaryKeyColumns: tc.primaryKeyColumns, + Fields: sqltypes.MakeTestFields(fields, fieldTypes), + } + + tvde.tmc.schema = &tabletmanagerdatapb.SchemaDefinition{ + TableDefinitions: []*tabletmanagerdatapb.TableDefinition{sourceTable}, + } + + td := &tableDiffer{ + wd: &workflowDiffer{ + ct: ct, + }, + table: sourceTable, + tablePlan: &tablePlan{ + table: sourceTable, + sourceQuery: tc.sourceQuery, + }, + } + + err := td.getSourcePKCols() + require.NoError(t, err) + + assert.Equal(t, tc.wantSourcePkCols, td.tablePlan.sourcePkCols, + "sourcePkCols should reflect PK column positions in the source SELECT expression list") + }) + } +} + +// TestSourcePKSelectIndices tests the extracted sourcePKSelectIndices function +// directly with parsed queries, without needing topo/tablet infrastructure. +func TestSourcePKSelectIndices(t *testing.T) { + testCases := []struct { + name string + sourceQuery string + pkColumns []string + wantIndices []int + // wantErr is for the present-but-non-physical case (a PK name is + // projected only via a computed/unrelated expression): fail closed. + wantErr bool + // wantNotProjected is for the entirely-absent case (a PK column is not + // projected at all): no error, allProjected == false, and no partial + // index slice is produced. + wantNotProjected bool + }{ + { + name: "natural order single PK", + sourceQuery: "select c1, c2 from t1 order by c1 asc", + pkColumns: []string{"c1"}, + wantIndices: []int{0}, + }, + { + name: "reordered columns single PK", + sourceQuery: "select c2, c1 from t1 order by c1 asc", + pkColumns: []string{"c1"}, + wantIndices: []int{1}, + }, + { + name: "composite PK natural order", + sourceQuery: "select c1, c2 from multipk order by c1 asc, c2 asc", + pkColumns: []string{"c1", "c2"}, + wantIndices: []int{0, 1}, + }, + { + name: "composite PK columns reordered in select", + sourceQuery: "select c2, c1 from multipk order by c1 asc, c2 asc", + pkColumns: []string{"c1", "c2"}, + wantIndices: []int{1, 0}, + }, + { + name: "composite PK 4 columns fully reversed", + sourceQuery: "select d, c, b, a from t order by b asc, d asc", + pkColumns: []string{"b", "d"}, + wantIndices: []int{2, 0}, + }, + { + // Cross-table MoveTables filter: the physical source column c0 is + // renamed to the target column c1. The source PK is the physical + // column c0, which we match via the underlying ColName. + name: "renamed physical source column (cross-table MoveTables filter)", + sourceQuery: "select c0 as c1, c2 from t2 order by c1 asc", + pkColumns: []string{"c0"}, + wantIndices: []int{0}, + }, + { + // A CONVERT(col USING charset) rename must unwrap to the inner + // physical source column, mirroring the row streamer planner. + name: "convert using rename unwraps to source column", + sourceQuery: "select convert(c1 using utf8mb4) as c2, c3 from t order by c2 asc", + pkColumns: []string{"c1"}, + wantIndices: []int{0}, + }, + { + // A nested CONVERT (e.g. wrapping a CAST) is NOT a direct column + // rename, so it is not treated as a physical PK column and fails + // closed. Here c1 is entirely absent otherwise, so the whole key is + // reported as not projected. + name: "nested convert using is not a physical column", + sourceQuery: "select convert(cast(c1 as char) using utf8mb4) as c2, c3 from t order by c2 asc", + pkColumns: []string{"c1"}, + wantNotProjected: true, + }, + { + // A computed expression wrapped in CONVERT that is aliased back to + // the PK name must NOT satisfy the PK lookup: it is a derived value, + // not the physical column. Since the alias names the PK column, this + // is the present-but-non-physical case and must fail closed. + name: "computed convert aliased to PK name fails closed", + sourceQuery: "select convert(concat(c1, 'x') using utf8mb4) as c1, c2 from t order by c1 asc", + pkColumns: []string{"c1"}, + wantErr: true, + }, + { + name: "function expression with alias", + sourceQuery: "select c1, c2, count(*) as c3, sum(c4) as c4 from t group by c1 order by c1 asc", + pkColumns: []string{"c1"}, + wantIndices: []int{0}, + }, + { + name: "PK at last position", + sourceQuery: "select a, b, c, id from t order by id asc", + pkColumns: []string{"id"}, + wantIndices: []int{3}, + }, + { + name: "case insensitive match", + sourceQuery: "select ID, Name from t order by ID asc", + pkColumns: []string{"id"}, + wantIndices: []int{0}, + }, + { + // A PK column entirely absent from the SELECT list is a valid + // subset-projection filter: no error, and allProjected is false so + // the caller does not build a partial source key. + name: "PK entirely absent from select list is not projected", + sourceQuery: "select a, b from t order by a asc", + pkColumns: []string{"missing_col"}, + wantNotProjected: true, + }, + { + // Mirrors the customer CI case: composite source PK (cid, typ) with + // a filter that projects only cid. typ is entirely absent, so the + // whole key is treated as not projected (never a partial [0] slice). + name: "composite PK with one column absent is not projected", + sourceQuery: "select cid, name from customer order by cid asc", + pkColumns: []string{"cid", "typ"}, + wantNotProjected: true, + }, + { + name: "in_keyrange filter preserves column positions", + sourceQuery: "select c1, c2 from t1 where in_keyrange('-80') order by c1 asc", + pkColumns: []string{"c1"}, + wantIndices: []int{0}, + }, + { + name: "three PKs scattered across wide select", + sourceQuery: "select a, b, c, d, e, f from t order by b asc, d asc, f asc", + pkColumns: []string{"b", "d", "f"}, + wantIndices: []int{1, 3, 5}, + }, + { + // The source PKs are the physical columns src_a and src_b, matched + // via their underlying ColNames even though they are renamed. + name: "multi-column cross-table filter matches physical columns", + sourceQuery: "select src_a as id, src_b as name, src_c as value from source_t order by id asc", + pkColumns: []string{"src_a", "src_b"}, + wantIndices: []int{0, 1}, + }, + { + // A computed alias must not satisfy a source PK lookup: the row + // streamer resumes using the physical PK value, so persisting a + // derived value would skip/repeat rows. Fail closed instead. + name: "computed alias does not satisfy source PK (fails closed)", + sourceQuery: "select a + b as id, c from t order by id asc", + pkColumns: []string{"id"}, + wantErr: true, + }, + { + // An alias mapping an unrelated physical column to the source PK + // name must not match; only the real physical PK column may. + name: "unrelated column aliased to source PK name fails closed", + sourceQuery: "select other_col as id, c from t order by id asc", + pkColumns: []string{"id"}, + wantErr: true, + }, + { + // Invariant guard: buildTablePlan must expand "*" into explicit + // columns before sourcePKSelectIndices runs. A StarExpr reaching this + // function means a caller violated that invariant, so we fail loud + // rather than silently treating PK columns as not projected. + name: "unexpanded star fails loud (invariant guard)", + sourceQuery: "select * from t order by a asc", + pkColumns: []string{"a"}, + wantErr: true, + }, + { + name: "column swap alias does not shadow real PK", + sourceQuery: "select b as a, a as b from t order by a asc", + pkColumns: []string{"a"}, + wantIndices: []int{1}, + }, + { + name: "column swap composite PK prefers ColName over alias", + sourceQuery: "select b as a, a as b, c from t order by a asc, b asc", + pkColumns: []string{"a", "b"}, + wantIndices: []int{1, 0}, + }, + } + + parser := sqlparser.NewTestParser() + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + statement, err := parser.Parse(tc.sourceQuery) + require.NoError(t, err) + sourceSelect, ok := statement.(*sqlparser.Select) + require.True(t, ok) + + indices, allProjected, err := sourcePKSelectIndices(sourceSelect, tc.pkColumns) + if tc.wantErr { + require.Error(t, err) + return + } + require.NoError(t, err) + if tc.wantNotProjected { + assert.False(t, allProjected) + assert.Empty(t, indices, "must not return a partial index slice when a PK column is absent") + return + } + assert.True(t, allProjected) + assert.Equal(t, tc.wantIndices, indices) + }) + } +} diff --git a/go/vt/vttablet/tabletmanager/vdiff/table_differ.go b/go/vt/vttablet/tabletmanager/vdiff/table_differ.go index 7a5ad1f79ea..9987741548d 100644 --- a/go/vt/vttablet/tabletmanager/vdiff/table_differ.go +++ b/go/vt/vttablet/tabletmanager/vdiff/table_differ.go @@ -550,7 +550,18 @@ func (td *tableDiffer) diff(ctx context.Context, coreOpts *tabletmanagerdatapb.V curState := cs.Named().Row() mismatch := curState.AsBool("mismatch", false) dr := &DiffReport{} - if rpt := curState.AsBytes("report", []byte("{}")); json.Valid(rpt) { + if td.tablePlan.sourceCheckpointUnavailable { + // This table has no resumable checkpoint and restarts from the beginning + // on every run (see getSourcePKCols). Carrying over the persisted partial + // report or mismatch flag would double-count rows and duplicate mismatch + // samples across restarts, so we start fresh instead. Also clear the + // persisted mismatch bit so a mismatch recorded by a discarded partial + // attempt does not stick after a clean full-table pass. + mismatch = false + if err = clearTableMismatch(dbClient, td.wd.ct.id, td.table.Name); err != nil { + return nil, err + } + } else if rpt := curState.AsBytes("report", []byte("{}")); json.Valid(rpt) { if err = json.Unmarshal(rpt, dr); err != nil { return nil, err } @@ -766,7 +777,27 @@ func (td *tableDiffer) updateTableProgress(dbClient binlogplayer.DBClient, dr *D return err } - if lastRow == nil { + switch { + case td.tablePlan.sourceCheckpointUnavailable: + // The source PK cannot be represented as a resumable checkpoint (see + // getSourcePKCols). Persist progress but explicitly clear lastpk (to NULL, + // which also discards any stale value written before this fix) and leave + // the in-memory retry PKs unset. Any resume then restarts the whole table + // from the beginning for both streams, which avoids the false + // ExtraRowsSource that a source-only restart against a resumed target + // would produce. + query, err = sqlparser.ParseAndBind(sqlUpdateTableProgress, + sqltypes.Int64BindVariable(dr.ProcessedRows), + sqltypes.NullBindVariable, + sqltypes.StringBindVariable(string(rpt)), + sqltypes.Int64BindVariable(td.wd.ct.id), + sqltypes.StringBindVariable(td.table.Name), + ) + if err != nil { + return err + } + case lastRow == nil: + // No rows were processed, so there is nothing to checkpoint. query, err = sqlparser.ParseAndBind(sqlUpdateTableNoProgress, sqltypes.Int64BindVariable(dr.ProcessedRows), sqltypes.StringBindVariable(string(rpt)), @@ -776,7 +807,7 @@ func (td *tableDiffer) updateTableProgress(dbClient binlogplayer.DBClient, dr *D if err != nil { return err } - } else { + default: lastPK := td.lastPKFromRow(lastRow) if td.wd.opts.CoreOptions.MaxDiffSeconds > 0 { // Update the in-memory lastPK as well so that we can restart the table @@ -872,6 +903,24 @@ func updateTableMismatch(dbClient binlogplayer.DBClient, vdiffID int64, table st return nil } +// clearTableMismatch resets the persisted mismatch bit for a table. It is used +// when a table with no resumable checkpoint restarts from the beginning, so a +// mismatch recorded by a discarded partial attempt does not stick after a clean +// full-table pass. +func clearTableMismatch(dbClient binlogplayer.DBClient, vdiffID int64, table string) error { + query, err := sqlparser.ParseAndBind(sqlClearTableMismatch, + sqltypes.Int64BindVariable(vdiffID), + sqltypes.StringBindVariable(table), + ) + if err != nil { + return err + } + if _, err = dbClient.ExecuteFetch(query, 1); err != nil { + return err + } + return nil +} + func (td *tableDiffer) lastPKFromRow(row []sqltypes.Value) *tabletmanagerdatapb.VDiffTableLastPK { buildQR := func(pkCols []int) *querypb.QueryResult { pkColCnt := len(pkCols) @@ -948,6 +997,27 @@ func (td *tableDiffer) getSourcePKCols() error { ctx, cancel := context.WithTimeout(td.wd.ct.vde.ctx, topo.RemoteOperationTimeout*3) defer cancel() + // Parse the source query first. We need it for two reasons: + // 1. The physical source table can differ from the target table name + // (td.table.Name) for cross-table MoveTables filters, e.g. a filter of + // "select ... from t2" for target table t1. The schema, PK-equivalent, + // and PK column lookups must all use the real source table. + // 2. We map PK columns against the SELECT expression order (the actual row + // layout) rather than td.table.Columns (column ordinal position), + // because filters can reorder columns (e.g. "select c2, c1 from t1"). + statement, err := td.wd.ct.vde.parser.Parse(td.tablePlan.sourceQuery) + if err != nil { + return vterrors.Wrapf(err, "failed to parse source query for table %s", td.table.Name) + } + sourceSelect, ok := statement.(*sqlparser.Select) + if !ok { + return vterrors.Errorf(vtrpcpb.Code_INTERNAL, "unexpected statement type for source query of table %s", td.table.Name) + } + sourceTableName, err := sourceTableNameFromSelect(sourceSelect) + if err != nil { + return vterrors.Wrapf(err, "failed to determine source table for target table %s", td.table.Name) + } + // We use the first sourceShard as all of them should have the same schema. if len(td.wd.ct.sources) == 0 { return vterrors.Errorf(vtrpcpb.Code_FAILED_PRECONDITION, "no source shards found in %s keyspace", @@ -971,16 +1041,16 @@ func (td *tableDiffer) getSourcePKCols() error { td.wd.ct.sourceKeyspace, sourceShardName) } sourceSchema, err := td.wd.ct.tmc.GetSchema(ctx, sourceTablet.Tablet, &tabletmanagerdatapb.GetSchemaRequest{ - Tables: []string{td.table.Name}, + Tables: []string{sourceTableName}, }) if err != nil { return vterrors.Wrapf(err, "failed to get the schema for table %s from source tablet %s", - td.table.Name, topoproto.TabletAliasString(sourceTablet.Alias)) + sourceTableName, topoproto.TabletAliasString(sourceTablet.Alias)) } if len(sourceSchema.TableDefinitions) == 0 { // The table no longer exists on the source. Any rows that exist on the target will be // reported as extra rows. - log.Warn(fmt.Sprintf("The %s table was not found on source tablet %s during VDiff for the %s workflow; any rows on the target will be reported as extra", td.table.Name, topoproto.TabletAliasString(sourceTablet.Alias), td.wd.ct.workflow)) + log.Warn(fmt.Sprintf("The %s table was not found on source tablet %s during VDiff for the %s workflow; any rows on the target will be reported as extra", sourceTableName, topoproto.TabletAliasString(sourceTablet.Alias), td.wd.ct.workflow)) return nil } sourceTable := sourceSchema.TableDefinitions[0] @@ -993,39 +1063,190 @@ func (td *tableDiffer) getSourcePKCols() error { }) if err != nil { return nil, vterrors.Wrapf(err, "failed to query the %s source tablet in order to get a primary key equivalent for the %s table", - topoproto.TabletAliasString(sourceTablet.Alias), td.table.Name) + topoproto.TabletAliasString(sourceTablet.Alias), sourceTableName) } return sqltypes.Proto3ToResult(res), nil } - pkeCols, _, err := mysqlctl.GetPrimaryKeyEquivalentColumns(ctx, executeFetch, sourceTablet.DbName(), td.table.Name) + pkeCols, _, err := mysqlctl.GetPrimaryKeyEquivalentColumns(ctx, executeFetch, sourceTablet.DbName(), sourceTableName) if err != nil { return vterrors.Wrapf(err, "failed to get a primary key equivalent for the %s table from source tablet %s", - td.table.Name, topoproto.TabletAliasString(sourceTablet.Alias)) + sourceTableName, topoproto.TabletAliasString(sourceTablet.Alias)) } if len(pkeCols) > 0 { - log.Info(fmt.Sprintf("Using primary key equivalent columns %+v for table %s in vdiff %s", pkeCols, td.table.Name, td.wd.ct.uuid)) + log.Info(fmt.Sprintf("Using primary key equivalent columns %+v for table %s in vdiff %s", pkeCols, sourceTableName, td.wd.ct.uuid)) sourceTable.PrimaryKeyColumns = pkeCols } else { // We use every column together as a substitute PK. - log.Info(fmt.Sprintf("Using all columns as a substitute primary key for table %s in vdiff %s", td.table.Name, td.wd.ct.uuid)) - sourceTable.PrimaryKeyColumns = append(sourceTable.PrimaryKeyColumns, td.table.Columns...) + log.Info(fmt.Sprintf("Using all columns as a substitute primary key for table %s in vdiff %s", sourceTableName, td.wd.ct.uuid)) + sourceTable.PrimaryKeyColumns = append(sourceTable.PrimaryKeyColumns, sourceTable.Columns...) } } - sourcePKColumns := make(map[string]struct{}, len(sourceTable.PrimaryKeyColumns)) - td.tablePlan.sourcePkCols = make([]int, 0, len(sourceTable.PrimaryKeyColumns)) - for _, pkc := range sourceTable.PrimaryKeyColumns { - sourcePKColumns[pkc] = struct{}{} + // Map each source PK column to its position in the SELECT expression order, + // which is the actual layout of the streamed rows. + indices, allProjected, err := sourcePKSelectIndices(sourceSelect, sourceTable.PrimaryKeyColumns) + if err != nil { + return vterrors.Wrapf(err, "table %s", sourceTableName) } - for i, pkc := range td.table.Columns { - if _, ok := sourcePKColumns[pkc]; ok { - td.tablePlan.sourcePkCols = append(td.tablePlan.sourcePkCols, i) - } + if !allProjected { + // At least one source PK column is not projected by the filter's SELECT + // list at all (e.g. source PK (cid, typ) with a filter of + // "select cid, name from customer"). We cannot build a resumable source + // checkpoint for such a subset-projection (or cross-table) filter: the + // row streamer always resumes on the full source PK and rejects a lastpk + // whose length does not match the source table's PK column count (see + // buildSelect in rowstreamer.go). + // + // Persisting only a target checkpoint (leaving the source key nil or + // empty) is not safe either: on resume the target would resume mid-table + // while the source restarts from the beginning, so the merge loop would + // classify every source row before the target's resume point as + // ExtraRowsSource, producing a false mismatch on every retry. + // + // So we record an explicit "source checkpoint unavailable" state. This + // makes updateTableProgress clear any persisted lastpk and leave the + // in-memory retry PKs unset, so any resume (auto-retry or manual resume) + // restarts the whole table from the beginning for both the source and + // target streams. That is correct (no false extra-row reports). Because + // such a table cannot be checkpointed it also cannot honor + // --max-diff-duration: if it exceeds that bound the differ stops with a + // non-retryable error rather than overriding the operator's bound (see + // differ). + td.tablePlan.sourceCheckpointUnavailable = true + // Discard any checkpoint that buildPlan already loaded from the database + // via getTableLastPK (e.g. a stale, possibly wrong-length lastpk written + // before this fix and then resumed after an upgrade). Leaving it set would + // resume the streams on that stale key instead of restarting the table. + td.lastSourcePK = nil + td.lastTargetPK = nil + return nil } + td.tablePlan.sourcePkCols = indices return nil } +// sourceTableNameFromSelect returns the physical source table referenced by the +// VReplication filter's SELECT. This can differ from the VDiff target table name +// for cross-table MoveTables filters (e.g. "select ... from t2" for target t1), +// and must be used for all source schema/PK lookups. +func sourceTableNameFromSelect(sourceSelect *sqlparser.Select) (string, error) { + if len(sourceSelect.From) != 1 { + return "", vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "unsupported source query, expected a single table in the FROM clause: %s", sqlparser.String(sourceSelect)) + } + aliased, ok := sourceSelect.From[0].(*sqlparser.AliasedTableExpr) + if !ok { + return "", vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "unsupported source query, expected a simple table reference: %s", sqlparser.String(sourceSelect)) + } + tableName := sqlparser.GetTableName(aliased.Expr) + if tableName.IsEmpty() { + return "", vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "unsupported source query, could not resolve the source table: %s", sqlparser.String(sourceSelect)) + } + return tableName.String(), nil +} + +// sourcePKSelectIndices maps each source PK column to its position in the SELECT +// expression list (the physical row layout the row streamer produces). The +// indices are returned in PK definition order. +// +// Because the schema lookup uses the real source table (see +// sourceTableNameFromSelect), a projected source PK column appears in the SELECT +// list as a direct reference to that physical column. We resolve each PK against +// the underlying column of the expression: +// - a plain ColName ("select c1, c2 ..." or reordered "select c2, c1 ..."), +// - a renamed column ("select source_id as target_id ...", underlying ColName), or +// - a CONVERT(col USING charset) rename, unwrapping to the inner ColName, +// mirroring how the row streamer's planner resolves the physical column. +// +// A source PK column can be missing in two distinct ways, handled differently: +// +// - Present-but-non-physical: an expression whose output column is named after +// the PK column (via an alias) but that does not resolve to the physical PK +// column, e.g. "select a + b as id" or "select other_col as id". This fails +// closed with an error. The row streamer resumes using the physical source +// PK value, so persisting a derived value as the source checkpoint would +// skip or repeat rows on resume. +// +// - Entirely absent: the PK column is not projected at all, e.g. source PK +// (cid, typ) with a filter of "select cid, name from customer". This is a +// valid subset-projection filter, so we return allProjected == false and no +// error. The caller must then treat the source checkpoint as unavailable and +// persist no resumable checkpoint (restarting the whole table on resume) +// rather than building a partial source key. We never return a partial-length +// index slice: the row streamer always resumes on the full source PK and +// rejects a lastpk whose length does not match the table's PK column count +// (see buildSelect in rowstreamer.go). +// +// allProjected is true only when every source PK column resolved to a physical +// SELECT column, in which case indices holds all of their SELECT-order indices. +func sourcePKSelectIndices(sourceSelect *sqlparser.Select, pkColumns []string) (indices []int, allProjected bool, err error) { + indices = make([]int, 0, len(pkColumns)) + for _, pkc := range pkColumns { + physicalIdx := -1 + aliasedButNotPhysical := false + for i, selExpr := range sourceSelect.SelectExprs.Exprs { + // Invariant: buildTablePlan expands "*" into explicit columns before + // this runs, so the SELECT list must contain only AliasedExprs. If a + // StarExpr ever reaches here it means a caller passed an unexpanded + // query; fail loud rather than silently treating PK columns as not + // projected (which would corrupt resume checkpoints). We do NOT expand + // the star here; that logic belongs in buildTablePlan. + if _, isStar := selExpr.(*sqlparser.StarExpr); isStar { + return nil, false, vterrors.Errorf(vtrpcpb.Code_INTERNAL, "unexpected '*' in vdiff source query SELECT list; expected columns to be expanded by buildTablePlan: %s", sqlparser.String(sourceSelect)) + } + aliasedExpr, ok := selExpr.(*sqlparser.AliasedExpr) + if !ok { + continue + } + // A physical match always wins, even if it appears after an alias + // that shadows the PK name. This preserves correct resolution for + // queries like "select b as a, a as b" (PK a -> the real a). + if colName, ok := underlyingSourceColumn(aliasedExpr.Expr); ok && strings.EqualFold(pkc, colName) { + physicalIdx = i + break + } + if !aliasedExpr.As.IsEmpty() && strings.EqualFold(pkc, aliasedExpr.As.String()) { + aliasedButNotPhysical = true + } + } + if physicalIdx >= 0 { + indices = append(indices, physicalIdx) + continue + } + if aliasedButNotPhysical { + // Present-but-non-physical: fail closed. + return nil, false, vterrors.Errorf(vtrpcpb.Code_INVALID_ARGUMENT, "source PK column %s is projected only via a non-physical expression in the source query SELECT list", pkc) + } + // Entirely absent: report to the caller instead of returning a partial + // (and therefore unusable) source key. + return nil, false, nil + } + return indices, true, nil +} + +// underlyingSourceColumn returns the name of the physical source column that a +// SELECT expression reads from, and whether the expression resolves to one. It +// handles plain column references and CONVERT(col USING charset) renames (used +// for charset conversions where the AS is the renamed target column). Computed +// expressions, functions, and literals do not resolve to a physical column and +// return ok == false. +func underlyingSourceColumn(expr sqlparser.Expr) (string, bool) { + switch e := expr.(type) { + case *sqlparser.ColName: + return e.Name.String(), true + case *sqlparser.ConvertUsingExpr: + // Only a direct column rename like "convert(c1 using utf8mb4) as c2" + // is a physical source column. Anything wrapped in a computation + // (concat, cast, arithmetic, etc.) is NOT a physical column and must + // fail closed so we never persist a derived value as the source + // checkpoint. Mirrors the fail-closed contract for computed aliases. + if inner, ok := e.Expr.(*sqlparser.ColName); ok { + return inner.Name.String(), true + } + } + return "", false +} + func getColumnNameForSelectExpr(selectExpression sqlparser.SelectExpr) (string, error) { aliasedExpr := selectExpression.(*sqlparser.AliasedExpr) expr := aliasedExpr.Expr diff --git a/go/vt/vttablet/tabletmanager/vdiff/table_differ_test.go b/go/vt/vttablet/tabletmanager/vdiff/table_differ_test.go index a782311e8c6..9d8b67c1381 100644 --- a/go/vt/vttablet/tabletmanager/vdiff/table_differ_test.go +++ b/go/vt/vttablet/tabletmanager/vdiff/table_differ_test.go @@ -25,6 +25,7 @@ import ( "vitess.io/vitess/go/sqltypes" "vitess.io/vitess/go/stats" "vitess.io/vitess/go/vt/binlog/binlogplayer" + "vitess.io/vitess/go/vt/sqlparser" querypb "vitess.io/vitess/go/vt/proto/query" tabletmanagerdatapb "vitess.io/vitess/go/vt/proto/tabletmanagerdata" @@ -153,7 +154,8 @@ func TestGetSourcePKCols_TableDroppedOnSource(t *testing.T) { }, table: table, tablePlan: &tablePlan{ - table: table, + table: table, + sourceQuery: "select c1, c2 from dropped_table order by c1 asc", }, } @@ -161,3 +163,231 @@ func TestGetSourcePKCols_TableDroppedOnSource(t *testing.T) { require.NoError(t, err) require.Nil(t, td.tablePlan.sourcePkCols) } + +// TestGetSourcePKCols_ComputedAliasFailsClosed verifies that when a source PK +// column is not projected as a physical column in the source query (only a +// computed value aliased to the PK name is present), getSourcePKCols fails +// closed rather than persisting the derived value as the source checkpoint. +// The row streamer resumes using the physical source PK, so accepting the +// computed value would skip or repeat rows on resume. +func TestGetSourcePKCols_ComputedAliasFailsClosed(t *testing.T) { + tvde := newTestVDiffEnv(t) + defer tvde.close() + + ct := tvde.createController(t, 1) + + // The source table's physical PK is "textcol", but the source query only + // projects a computed expression "a + b" aliased to "textcol". + sourceTable := &tabletmanagerdatapb.TableDefinition{ + Name: "pktext", + Columns: []string{"textcol", "c2"}, + PrimaryKeyColumns: []string{"textcol"}, + Fields: sqltypes.MakeTestFields("textcol|c2", "varchar|int64"), + } + tvde.tmc.schema = &tabletmanagerdatapb.SchemaDefinition{ + TableDefinitions: []*tabletmanagerdatapb.TableDefinition{sourceTable}, + } + + td := &tableDiffer{ + wd: &workflowDiffer{ + ct: ct, + }, + table: sourceTable, + tablePlan: &tablePlan{ + table: sourceTable, + sourceQuery: "select c2, a + b as textcol from pktext order by textcol asc", + }, + } + + err := td.getSourcePKCols() + require.Error(t, err) + require.ErrorContains(t, err, "source PK column textcol is projected only via a non-physical expression") + require.Nil(t, td.tablePlan.sourcePkCols) +} + +// TestGetSourcePKCols_SubsetProjectionUnavailable mirrors the customer +// materialize CI regression: the source table has a composite PK (cid, typ) but +// the filter projects only a subset that omits the trailing PK column typ +// ("select cid, name from customer"). getSourcePKCols must NOT fail closed for +// this valid subset-projection filter, and it must NOT build a partial source +// key. Instead it flags the source checkpoint as unavailable so that no +// resumable checkpoint is persisted and the whole table restarts on resume. +func TestGetSourcePKCols_SubsetProjectionUnavailable(t *testing.T) { + tvde := newTestVDiffEnv(t) + defer tvde.close() + + ct := tvde.createController(t, 1) + + sourceTable := &tabletmanagerdatapb.TableDefinition{ + Name: "customer", + Columns: []string{"cid", "name", "typ"}, + PrimaryKeyColumns: []string{"cid", "typ"}, + Fields: sqltypes.MakeTestFields("cid|name|typ", "int64|varchar|varchar"), + } + tvde.tmc.schema = &tabletmanagerdatapb.SchemaDefinition{ + TableDefinitions: []*tabletmanagerdatapb.TableDefinition{sourceTable}, + } + + td := &tableDiffer{ + wd: &workflowDiffer{ + ct: ct, + }, + table: sourceTable, + tablePlan: &tablePlan{ + table: sourceTable, + sourceQuery: "select cid, name from customer order by cid asc", + // The target has cid as its (single) PK at SELECT index 0. + pkCols: []int{0}, + }, + } + + err := td.getSourcePKCols() + require.NoError(t, err) + // The source checkpoint is flagged unavailable and no partial source key is + // built. + require.True(t, td.tablePlan.sourceCheckpointUnavailable) + require.Empty(t, td.tablePlan.sourcePkCols) +} + +// TestGetSourcePKCols_ResumeCheckpointReorderedPK is a resume regression test. +// For a composite-PK table whose source query reorders the PK columns +// (SELECT layout differs from PK definition order), it confirms that the +// sourcePkCols indices point at the correct SELECT positions so that the +// persisted source lastpk pairs each PK value with the right column. Before +// the fix, the indices were in column-ordinal order and lastPKFromRow built a +// corrupted source checkpoint, causing false ExtraRowsSource on every resume. +func TestGetSourcePKCols_ResumeCheckpointReorderedPK(t *testing.T) { + tvde := newTestVDiffEnv(t) + defer tvde.close() + + ct := tvde.createController(t, 1) + + // Physical source table "t" has columns a,b,c with composite PK (c, a). + // The source query projects them in a different order: b, c, a. + // So source PK "c" is at SELECT index 1 and "a" is at SELECT index 2. + // Fields are in DDL order (a,b,c), matching what GetSchema returns. + sourceTable := &tabletmanagerdatapb.TableDefinition{ + Name: "t", + Columns: []string{"a", "b", "c"}, + PrimaryKeyColumns: []string{"c", "a"}, + Fields: sqltypes.MakeTestFields("a|b|c", "int64|int64|int64"), + } + tvde.tmc.schema = &tabletmanagerdatapb.SchemaDefinition{ + TableDefinitions: []*tabletmanagerdatapb.TableDefinition{sourceTable}, + } + + td := &tableDiffer{ + wd: &workflowDiffer{ + ct: ct, + }, + table: sourceTable, + tablePlan: &tablePlan{ + table: sourceTable, + sourceQuery: "select b, c, a from t order by c asc, a asc", + // Target PK differs from source PK order so that lastPKFromRow + // persists a distinct Source checkpoint. + pkCols: []int{2, 1}, + }, + } + + err := td.getSourcePKCols() + require.NoError(t, err) + require.Equal(t, []int{1, 2}, td.tablePlan.sourcePkCols) + + // A streamed row [b=10, c=20, a=30] must serialize the source checkpoint + // as PK (c, a) = (20, 30), i.e. taking SELECT indices [1, 2], not the + // ordinal-order indices [2, 0] that the buggy code would have produced. + row := []sqltypes.Value{sqltypes.NewInt64(10), sqltypes.NewInt64(20), sqltypes.NewInt64(30)} + lastPK := td.lastPKFromRow(row) + require.NotNil(t, lastPK.Source) + sourceResult := sqltypes.Proto3ToResult(lastPK.Source) + require.Len(t, sourceResult.Rows, 1) + require.Equal(t, "20", sourceResult.Rows[0][0].ToString(), "first source PK value should be column c") + require.Equal(t, "30", sourceResult.Rows[0][1].ToString(), "second source PK value should be column a") + // Note: the checkpoint field-type metadata is still looked up via colIndex + // against table.Fields (DDL order), which is the pre-existing known + // limitation documented in the PR; it affects the target pkCols path + // equally and is out of scope here. This test pins the value indices, which + // is what this change fixes. +} + +// TestGetSourcePKCols_SubsetProjectionPersistReloadResume is the persist/reload/ +// resume regression test for the subset-projection case (source PK (cid, typ), +// target PK cid, filter "select cid, name from customer"). It exercises the full +// lifecycle where the failure would otherwise occur: +// +// 1. Persist: updateTableProgress runs after rows have been processed. +// 2. Reload: getTableLastPK reads the persisted state back on resume. +// +// Because the source checkpoint is unavailable for this filter, no lastpk may be +// persisted: a target-only checkpoint would resume the target mid-table while +// the source restarts from the beginning, so the merge loop would report every +// earlier source row as ExtraRowsSource. The fix persists no lastpk (an explicit +// "source checkpoint unavailable" state), so getTableLastPK returns nil on resume +// and the whole table restarts from the beginning for both the source and target +// streams. +func TestGetSourcePKCols_SubsetProjectionPersistReloadResume(t *testing.T) { + tvde := newTestVDiffEnv(t) + defer tvde.close() + + ct := tvde.createController(t, 1) + + sourceTable := &tabletmanagerdatapb.TableDefinition{ + Name: "customer", + Columns: []string{"cid", "name", "typ"}, + PrimaryKeyColumns: []string{"cid", "typ"}, + Fields: sqltypes.MakeTestFields("cid|name|typ", "int64|varchar|varchar"), + } + tvde.tmc.schema = &tabletmanagerdatapb.SchemaDefinition{ + TableDefinitions: []*tabletmanagerdatapb.TableDefinition{sourceTable}, + } + + wd := &workflowDiffer{ct: ct} + td := &tableDiffer{ + wd: wd, + table: sourceTable, + tablePlan: &tablePlan{ + table: sourceTable, + sourceQuery: "select cid, name from customer order by cid asc", + // The target has cid as its (single) PK at SELECT index 0. + pkCols: []int{0}, + }, + } + + require.NoError(t, td.getSourcePKCols()) + require.True(t, td.tablePlan.sourceCheckpointUnavailable) + + // --- Persist: even with a processed row, updateTableProgress must explicitly + // clear lastpk (to NULL) for a source-checkpoint-unavailable table, so no + // resumable (and unsafe) checkpoint remains, including any stale value from + // before this fix. + persistClient := binlogplayer.NewMockDBClient(t) + persistClient.ExpectRequestRE( + `^update _vt\.vdiff_table set rows_compared = 100, lastpk = null, report = '.*' where vdiff_id = 1 and table_name = 'customer'$`, + &sqltypes.Result{}, nil) + dr := &DiffReport{TableName: sourceTable.Name, ProcessedRows: 100} + row := []sqltypes.Value{sqltypes.NewInt64(42), sqltypes.NewVarChar("acme")} + require.NoError(t, td.updateTableProgress(persistClient, dr, row)) + // The in-memory retry PKs must remain unset so a same-process + // max-diff-duration restart also restarts both streams from the beginning. + require.Nil(t, td.lastSourcePK) + require.Nil(t, td.lastTargetPK) + + // --- Reload: with no lastpk persisted, getTableLastPK returns nil, so on + // resume both td.lastSourcePK and td.lastTargetPK stay nil and the whole + // table restarts from the beginning for both streams. + reloadClient := binlogplayer.NewMockDBClient(t) + getQuery, err := sqlparser.ParseAndBind(sqlGetVDiffTable, + sqltypes.Int64BindVariable(ct.id), + sqltypes.StringBindVariable(sourceTable.Name), + ) + require.NoError(t, err) + reloadClient.ExpectRequest(getQuery, sqltypes.MakeTestResult( + sqltypes.MakeTestFields("lastpk|mismatch|report", "varbinary|int64|varbinary"), + "|0|", // empty lastpk + ), nil) + + reloaded, err := wd.getTableLastPK(reloadClient, sourceTable.Name) + require.NoError(t, err) + require.Nil(t, reloaded, "no lastpk persisted, so resume must restart the whole table for both streams") +} diff --git a/go/vt/vttablet/tabletmanager/vdiff/table_plan.go b/go/vt/vttablet/tabletmanager/vdiff/table_plan.go index 4cbd3c5496e..8321fa30602 100644 --- a/go/vt/vttablet/tabletmanager/vdiff/table_plan.go +++ b/go/vt/vttablet/tabletmanager/vdiff/table_plan.go @@ -59,6 +59,16 @@ type tablePlan struct { // and target have different PK columns. sourcePkCols []int + // sourceCheckpointUnavailable is set when a source PK column cannot be + // represented in the source query's SELECT list (a subset-projection or + // cross-table filter that omits part of the source PK, e.g. source PK + // (cid, typ) with "select cid, name from customer"). We then cannot build a + // resumable source checkpoint, so no lastpk is persisted for the table and + // any resume (auto-retry, max-diff-duration restart, or manual resume) + // restarts the whole table from the beginning for both the source and target + // streams. See getSourcePKCols and updateTableProgress. + sourceCheckpointUnavailable bool + // selectPks is the list of pk columns as they appear in the select clause for the diff. selectPks []int dbName string diff --git a/go/vt/vttablet/tabletmanager/vdiff/workflow_differ.go b/go/vt/vttablet/tabletmanager/vdiff/workflow_differ.go index 681b4de3125..18eba631109 100644 --- a/go/vt/vttablet/tabletmanager/vdiff/workflow_differ.go +++ b/go/vt/vttablet/tabletmanager/vdiff/workflow_differ.go @@ -28,6 +28,7 @@ import ( "google.golang.org/protobuf/encoding/prototext" "vitess.io/vitess/go/mysql/collations" + "vitess.io/vitess/go/mysql/sqlerror" "vitess.io/vitess/go/sqltypes" "vitess.io/vitess/go/vt/binlog/binlogplayer" "vitess.io/vitess/go/vt/key" @@ -264,6 +265,21 @@ func (wd *workflowDiffer) diffTable(ctx context.Context, dbClient binlogplayer.D if !errors.Is(diffErr, ErrMaxDiffDurationExceeded) { // We only want to retry if we hit the max-diff-duration return diffErr } + if td.tablePlan.sourceCheckpointUnavailable { + // This table cannot be checkpointed because its filter does not + // project the full source primary key (see getSourcePKCols), so it + // cannot be resumed and every retry would restart from the beginning + // and hit the same timeout. We do not override the operator's + // --max-diff-duration bound (long-lived snapshots have real + // operational cost), so we stop here. The failure is wrapped as a + // MySQL ERNotSupportedYet so that its "(errno 1235)" suffix survives + // being persisted as a string and rebuilt by retryVDiffs: that makes + // IsEphemeralError classify it as non-ephemeral, so the engine does + // not auto-retry it forever. + return sqlerror.NewSQLError(sqlerror.ERNotSupportedYet, sqlerror.SSClientError, + fmt.Sprintf("table %s exceeded the configured --max-diff-duration and cannot be resumed because its filter does not project the full source primary key; increase or unset --max-diff-duration so it can complete within a single window", + td.table.Name)) + } } log.Info(fmt.Sprintf("Table diff done on table %s for vdiff %s with report: %+v", td.table.Name, wd.ct.uuid, diffReport)) diff --git a/go/vt/vttablet/tabletmanager/vdiff/workflow_differ_test.go b/go/vt/vttablet/tabletmanager/vdiff/workflow_differ_test.go index 82fd0c44300..779edd646b7 100644 --- a/go/vt/vttablet/tabletmanager/vdiff/workflow_differ_test.go +++ b/go/vt/vttablet/tabletmanager/vdiff/workflow_differ_test.go @@ -515,7 +515,7 @@ func TestBuildPlanSuccess(t *testing.T) { compareCols: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), false, "c2"}, {1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c1"}}, comparePKs: []compareColInfo{{1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c1"}}, pkCols: []int{1}, - sourcePkCols: []int{0}, + sourcePkCols: []int{1}, selectPks: []int{1}, orderBy: sqlparser.OrderBy{&sqlparser.Order{ Expr: &sqlparser.ColName{Name: sqlparser.NewIdentifierCI("c1")}, @@ -580,7 +580,7 @@ func TestBuildPlanSuccess(t *testing.T) { compareCols: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), false, "textcol"}, {1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c1"}}, comparePKs: []compareColInfo{{1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c1"}}, pkCols: []int{1}, - sourcePkCols: []int{0}, + sourcePkCols: []int{1}, selectPks: []int{1}, orderBy: sqlparser.OrderBy{&sqlparser.Order{ Expr: &sqlparser.ColName{Name: sqlparser.NewIdentifierCI("c1")}, @@ -602,7 +602,7 @@ func TestBuildPlanSuccess(t *testing.T) { compareCols: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "textcol"}, {1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), false, "c2"}}, comparePKs: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "textcol"}}, pkCols: []int{0}, - sourcePkCols: []int{}, + sourcePkCols: []int{0}, selectPks: []int{0}, orderBy: sqlparser.OrderBy{&sqlparser.Order{ Expr: &sqlparser.ColName{Name: sqlparser.NewIdentifierCI("textcol")}, @@ -624,7 +624,7 @@ func TestBuildPlanSuccess(t *testing.T) { compareCols: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), false, "c2"}, {1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "textcol"}}, comparePKs: []compareColInfo{{1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "textcol"}}, pkCols: []int{1}, - sourcePkCols: []int{}, + sourcePkCols: []int{1}, selectPks: []int{1}, orderBy: sqlparser.OrderBy{&sqlparser.Order{ Expr: &sqlparser.ColName{Name: sqlparser.NewIdentifierCI("textcol")}, @@ -646,7 +646,7 @@ func TestBuildPlanSuccess(t *testing.T) { compareCols: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c1"}, {1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c2"}, {2, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c3"}}, comparePKs: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c1"}, {1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c2"}, {2, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c3"}}, pkCols: []int{0, 1, 2}, - sourcePkCols: []int{0}, + sourcePkCols: []int{0, 1, 2}, selectPks: []int{0, 1, 2}, orderBy: sqlparser.OrderBy{ &sqlparser.Order{ @@ -678,7 +678,7 @@ func TestBuildPlanSuccess(t *testing.T) { compareCols: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), false, "c1"}, {1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), false, "c2"}, {2, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c3"}}, comparePKs: []compareColInfo{{2, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c3"}}, pkCols: []int{2}, - sourcePkCols: []int{0}, + sourcePkCols: []int{2}, selectPks: []int{2}, orderBy: sqlparser.OrderBy{ &sqlparser.Order{ @@ -687,28 +687,6 @@ func TestBuildPlanSuccess(t *testing.T) { }, }, }, - }, { - // Text column as expression. - input: &binlogdatapb.Rule{ - Match: "pktext", - Filter: "select c2, a+b as textcol from pktext", - }, - table: "pktext", - tablePlan: &tablePlan{ - dbName: vdiffDBName, - table: testSchema.TableDefinitions[tableDefMap["pktext"]], - sourceQuery: "select c2, a + b as textcol from pktext order by textcol asc", - targetQuery: "select c2, textcol from pktext order by textcol asc", - compareCols: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), false, "c2"}, {1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "textcol"}}, - comparePKs: []compareColInfo{{1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "textcol"}}, - pkCols: []int{1}, - sourcePkCols: []int{}, - selectPks: []int{1}, - orderBy: sqlparser.OrderBy{&sqlparser.Order{ - Expr: &sqlparser.ColName{Name: sqlparser.NewIdentifierCI("textcol")}, - Direction: sqlparser.AscOrder, - }}, - }, }, { // Multiple PK columns. input: &binlogdatapb.Rule{ @@ -723,7 +701,7 @@ func TestBuildPlanSuccess(t *testing.T) { compareCols: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c1"}, {1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c2"}}, comparePKs: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c1"}, {1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "c2"}}, pkCols: []int{0, 1}, - sourcePkCols: []int{0}, + sourcePkCols: []int{0, 1}, selectPks: []int{0, 1}, orderBy: sqlparser.OrderBy{ &sqlparser.Order{ @@ -909,7 +887,7 @@ func TestBuildPlanSuccess(t *testing.T) { compareCols: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "id"}, {1, collations.MySQL8().LookupByName(sqltypes.NULL.String()), false, "dt"}}, comparePKs: []compareColInfo{{0, collations.MySQL8().LookupByName(sqltypes.NULL.String()), true, "id"}}, pkCols: []int{0}, - sourcePkCols: []int{}, + sourcePkCols: []int{0}, selectPks: []int{0}, orderBy: sqlparser.OrderBy{&sqlparser.Order{ Expr: &sqlparser.ColName{Name: sqlparser.NewIdentifierCI("id")},