VDiff: save a sample for every drained extra row so reconciliation can match them (#20855)

Signed-off-by: Matthias Crauwels <matthias.crauwels@planetscale.com>
Signed-off-by: Matt Lord <mattalord@gmail.com>
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: Matt Lord <mattalord@gmail.com>

Author: 3 people

go/vt/vttablet/tabletmanager/vdiff/primitive_executor.go

@@ -89,19 +89,3 @@ func (pe *primitiveExecutor) next() ([]sqltypes.Value, error) {
 	pe.rows = pe.rows[1:]
 	return row, nil
 }
-
-// drain fastforward's a shard to process (and ignore) everything from its results stream and return a count of the
-// discarded rows.
-func (pe *primitiveExecutor) drain(ctx context.Context) (int64, error) {
-	var count int64
-	for {
-		row, err := pe.next()
-		if err != nil {
-			return 0, err
-		}
-		if row == nil {
-			return count, nil
-		}
-		count++
-	}
-}

go/vt/vttablet/tabletmanager/vdiff/report.go

@@ -63,6 +63,13 @@ type DiffMismatch struct {
 type RowDiff struct {
 	Row   map[string]string `json:"Row,omitempty"`
 	Query string            `json:"Query,omitempty"`
+	// LosslessValues is set when the sample contains all of the row's column
+	// values without truncation, meaning it can be used to prove that two
+	// rows are identical during extra-row reconciliation. The marker is
+	// deliberately affirmative: samples that are lossy (only-pks, truncated
+	// values) -- or that were persisted by an older binary and reloaded on
+	// resume -- lack it and are excluded from reconciliation.
+	LosslessValues bool `json:"LosslessValues,omitempty"`
 }
 
 func (td *tableDiffer) genRowDiff(queryStmt string, row []sqltypes.Value, opts *tabletmanagerdatapb.VDiffReportOptions) (*RowDiff, error) {
@@ -81,13 +88,15 @@ func (td *tableDiffer) genRowDiff(queryStmt string, row []sqltypes.Value, opts *
 		rd.Query = td.genDebugQueryDiff(sel, row, opts.GetOnlyPks())
 	}
 
+	truncated := false
 	addVal := func(index int, truncateAt int) error {
 		buf := sqlparser.NewTrackedBuffer(nil)
 		sel.SelectExprs.Exprs[index].Format(buf)
 		col := buf.String()
 		// Let's truncate if it's really worth it to avoid losing
 		// value for a few chars.
 		if truncateAt > 0 && row[index].Len() >= truncateAt+len(truncatedNotation)+20 {
+			truncated = true
 			if row[index].IsBinary() {
 				rb, err := row[index].ToBytes()
 				if err != nil { // Should never happen
@@ -126,6 +135,9 @@ func (td *tableDiffer) genRowDiff(queryStmt string, row []sqltypes.Value, opts *
 	}
 
 	if opts.GetOnlyPks() {
+		// A PK-only sample is still lossless when the PK columns cover the
+		// entire projection, since PK values are never truncated.
+		rd.LosslessValues = len(pks) == len(sel.SelectExprs.Exprs)
 		return rd, nil
 	}
 
@@ -138,6 +150,10 @@ func (td *tableDiffer) genRowDiff(queryStmt string, row []sqltypes.Value, opts *
 		}
 	}
 
+	// The sample contains all of the row's column values (this point is not
+	// reached with only-pks); it is lossless if none of them were truncated.
+	rd.LosslessValues = !truncated
+
 	return rd, nil
 }
 

go/vt/vttablet/tabletmanager/vdiff/report_test.go

@@ -66,6 +66,7 @@ func TestGenRowDiff(t *testing.T) {
 			},
 			reportOptions: &tabletmanagerdatapb.VDiffReportOptions{},
 			want: &RowDiff{
+				LosslessValues: true,
 				Row: map[string]string{ // The two PK cols should be first
 					// mysql> select hex("hi4");
 					// +------------+
@@ -131,6 +132,7 @@ func TestGenRowDiff(t *testing.T) {
 				DebugQuery: true,
 			},
 			want: &RowDiff{
+				LosslessValues: true,
 				Row: map[string]string{
 					"c1": "1",
 					"c2": "2",

go/vt/vttablet/tabletmanager/vdiff/table_differ.go

@@ -581,7 +581,14 @@ func (td *tableDiffer) diff(ctx context.Context, coreOpts *tabletmanagerdatapb.V
 	maxReportSampleRows := reportOpts.GetMaxSampleRows()
 
 	for {
-		lastProcessedRow = sourceRow
+		// Only advance the persisted position when the previous iteration
+		// consumed the held source row (advanceSource still holds that
+		// iteration's decision here). After an extra-target-row iteration the
+		// held source row has not been processed yet, and recording it as
+		// lastpk would make a resumed diff skip it permanently.
+		if advanceSource {
+			lastProcessedRow = sourceRow
+		}
 
 		select {
 		case <-ctx.Done():
@@ -629,35 +636,60 @@ func (td *tableDiffer) diff(ctx context.Context, coreOpts *tabletmanagerdatapb.V
 		advanceSource = true
 		advanceTarget = true
 		if sourceRow == nil {
-			diffRow, err := td.genRowDiff(td.tablePlan.sourceQuery, targetRow, reportOpts)
-			if err != nil {
-				return nil, vterrors.Wrap(err, "unexpected error generating diff")
-			}
-			dr.ExtraRowsTargetDiffs = append(dr.ExtraRowsTargetDiffs, diffRow)
-
-			// Drain target, update count.
-			count, err := targetExecutor.drain(ctx)
-			if err != nil {
-				return nil, err
+			// No more rows from the source; drain the remaining target rows,
+			// saving a sample for each one (up to maxExtraRowsToCompare) so that
+			// reconcileExtraRows can match them against any extra source rows.
+			// Counting drained rows without saving a sample makes them impossible
+			// to reconcile, producing false positive extra rows in the report.
+			// The drained rows are merged into the report only after the full
+			// drain succeeds: they are beyond the persisted lastpk, so partially
+			// counted rows would be counted again when a failed diff is resumed.
+			drainedRows := int64(0)
+			var drainedDiffs []*RowDiff
+			for targetRow != nil {
+				if dr.ExtraRowsTarget+drainedRows < maxExtraRowsToCompare {
+					diffRow, err := td.genRowDiff(td.tablePlan.targetQuery, targetRow, reportOpts)
+					if err != nil {
+						return nil, vterrors.Wrap(err, "unexpected error generating diff")
+					}
+					drainedDiffs = append(drainedDiffs, diffRow)
+				}
+				drainedRows++
+				targetRow, err = targetExecutor.next()
+				if err != nil {
+					return nil, err
+				}
 			}
-			dr.ExtraRowsTarget += 1 + count
-			dr.ProcessedRows += 1 + count
+			dr.ExtraRowsTarget += drainedRows
+			dr.ProcessedRows += drainedRows
+			dr.ExtraRowsTargetDiffs = append(dr.ExtraRowsTargetDiffs, drainedDiffs...)
 			return dr, nil
 		}
 		if targetRow == nil {
-			// No more rows from the target but we know we have more rows from
-			// source, so drain them and update the counts.
-			diffRow, err := td.genRowDiff(td.tablePlan.sourceQuery, sourceRow, reportOpts)
-			if err != nil {
-				return nil, vterrors.Wrap(err, "unexpected error generating diff")
-			}
-			dr.ExtraRowsSourceDiffs = append(dr.ExtraRowsSourceDiffs, diffRow)
-			count, err := sourceExecutor.drain(ctx)
-			if err != nil {
-				return nil, err
+			// No more rows from the target; drain the remaining source rows,
+			// saving a sample for each one (up to maxExtraRowsToCompare) so that
+			// reconcileExtraRows can match them against any extra target rows.
+			// As above, the drained rows are merged into the report only after
+			// the full drain succeeds.
+			drainedRows := int64(0)
+			var drainedDiffs []*RowDiff
+			for sourceRow != nil {
+				if dr.ExtraRowsSource+drainedRows < maxExtraRowsToCompare {
+					diffRow, err := td.genRowDiff(td.tablePlan.sourceQuery, sourceRow, reportOpts)
+					if err != nil {
+						return nil, vterrors.Wrap(err, "unexpected error generating diff")
+					}
+					drainedDiffs = append(drainedDiffs, diffRow)
+				}
+				drainedRows++
+				sourceRow, err = sourceExecutor.next()
+				if err != nil {
+					return nil, err
+				}
 			}
-			dr.ExtraRowsSource += 1 + count
-			dr.ProcessedRows += 1 + count
+			dr.ExtraRowsSource += drainedRows
+			dr.ProcessedRows += drainedRows
+			dr.ExtraRowsSourceDiffs = append(dr.ExtraRowsSourceDiffs, drainedDiffs...)
 			return dr, nil
 		}