Feature: Handle read only columns - #437 followup

CHANGELOG.md

@@ -2,10 +2,41 @@
 
 All notable changes to this project will be documented in this file.
 
-## [1.4.0 - 2026-05-18]
+## [1.4.0 - Unreleased]
 
 ### Added
-- Support generated/readonly columns by @plisandro, @driv3r, @grodowski in #437
+
+- Support MySQL generated columns (`VIRTUAL` and `STORED`) by @plisandro, @driv3r, @grodowski in #437.
+  Ghostferry no longer writes to generated columns: they are excluded from the column list of
+  every `INSERT` and from the `SET` clause of every replayed `UPDATE`, so the target recomputes
+  them from its own column definitions. They remain in `WHERE` clauses and in verification
+  fingerprints.
+
+### Changed
+
+- Generated columns are included in verification fingerprints, so a target whose generated column
+  definitions differ from the source's is reported as a mismatch. A specific column can be excluded
+  with `IgnoredColumnsForVerification`.
+
+- A `VIRTUAL` generated column is accepted as an explicitly configured pagination key when it is
+  `NOT NULL` and has a visible single-column `UNIQUE` index. A table whose columns are *all*
+  generated is rejected when schemas are loaded. Unsupported table shapes fail at startup with an
+  explanatory error rather than part-way through a move.
+
+### Fixed
+
+- `StopTargetVerifier` no longer panics when the ferry stops before `Run` starts the target
+  verifier. `targetVerifierWg` is held by value, so `Wait` on a verifier that never started is
+  a no-op. This affects embedders that defer `StopTargetVerifier` around `Run`.
+
+- `RowBatch.AsSQLQuery` returns an error instead of panicking when every selected column is
+  generated.
+
+### API
+
+New exported surface for embedders: `NewRowBatchWithColumns`, `IsColumnGenerated`,
+`TableSchema.IsColumnIndexGenerated`, `TableSchema.IsColumnNameGenerated`,
+`TableSchema.ColumnsCount`, `NoNonGeneratedColumnsError` and `VirtualPaginationKeyError`.
 
 ## [1.3.1 - 2026-04-15]
 

dml_events.go

@@ -168,15 +168,14 @@ func (e *BinlogInsertEvent) NewValues() RowData {
 }
 
 func (e *BinlogInsertEvent) AsSQLString(schemaName, tableName string) (string, error) {
-	filteredNewValues, err := e.table.FilterGeneratedColumnsOnRowData(e.newValues)
-	if err != nil {
+	if err := verifyValuesHasTheSameLengthAsColumns(e.table, e.newValues); err != nil {
 		return "", err
 	}
 
 	query := "INSERT IGNORE INTO " +
 		QuotedTableNameFromString(schemaName, tableName) +
 		" (" + strings.Join(quotedColumnNames(e.table), ",") + ")" +
-		" VALUES (" + buildStringListForValues(e.table, filteredNewValues) + ")"
+		" VALUES (" + buildStringListForValues(e.table, e.newValues) + ")"
 
 	return query, nil
 }
@@ -229,7 +228,7 @@ func (e *BinlogUpdateEvent) AsSQLString(schemaName, tableName string) (string, e
 
 	query := "UPDATE " + QuotedTableNameFromString(schemaName, tableName) +
 		" SET " + buildStringMapForSet(e.table, e.newValues) +
-		" WHERE " + buildStringMapForWhere(e.table, e.oldValues)
+		" WHERE " + buildStringMapForWhere(e.table.Columns, e.oldValues)
 
 	return query, nil
 }
@@ -270,7 +269,7 @@ func (e *BinlogDeleteEvent) AsSQLString(schemaName, tableName string) (string, e
 	}
 
 	query := "DELETE FROM " + QuotedTableNameFromString(schemaName, tableName) +
-		" WHERE " + buildStringMapForWhere(e.table, e.oldValues)
+		" WHERE " + buildStringMapForWhere(e.table.Columns, e.oldValues)
 
 	return query, nil
 }
@@ -282,39 +281,37 @@ func (e *BinlogDeleteEvent) PaginationKey() (string, error) {
 func NewBinlogDMLEvents(table *TableSchema, ev *replication.BinlogEvent, pos, resumablePos mysql.Position, query []byte) ([]DMLEvent, error) {
 	rowsEvent := ev.Event.(*replication.RowsEvent)
 
-	for i, rawRow := range rowsEvent.Rows {
-		if len(rawRow) != len(table.Columns) {
+	for _, row := range rowsEvent.Rows {
+		if len(row) != len(table.Columns) {
 			return nil, fmt.Errorf(
 				"table %s.%s has %d columns but event has %d columns instead",
 				table.Schema,
 				table.Name,
 				len(table.Columns),
-				len(rawRow),
+				len(row),
 			)
 		}
 
-		// Normalize signed-to-unsigned integer values in place using
-		// full-schema column indexes.  go-mysql always decodes rows to the
-		// full column width (RowsEvent.decodeImage allocates make([]any,
-		// ColumnCount) and leaves omitted positions as nil), so rawRow is
-		// always len(table.Columns) here and indexing is safe.
-		for j, col := range table.Columns {
+		// Normalise signed-to-unsigned integer values in place.  Generated
+		// columns must be normalised too — their values reach the WHERE
+		// clause of replayed UPDATEs and DELETEs, and skipping them would
+		// emit a negative value there that matches nothing.
+		for i, col := range table.Columns {
 			if col.IsUnsigned {
-				switch v := rawRow[j].(type) {
+				switch v := row[i].(type) {
 				case int64:
-					rawRow[j] = uint64(v)
+					row[i] = uint64(v)
 				case int32:
-					rawRow[j] = uint32(v)
+					row[i] = uint32(v)
 				case int16:
-					rawRow[j] = uint16(v)
+					row[i] = uint16(v)
 				case int8:
-					rawRow[j] = uint8(v)
+					row[i] = uint8(v)
 				case int:
-					rawRow[j] = uint(v)
+					row[i] = uint(v)
 				}
 			}
 		}
-		rowsEvent.Rows[i] = rawRow
 	}
 
 	timestamp := time.Unix(int64(ev.Header.Timestamp), 0)
@@ -331,10 +328,15 @@ func NewBinlogDMLEvents(table *TableSchema, ev *replication.BinlogEvent, pos, re
 	}
 }
 
+// Generated columns are excluded from the INSERT column list because MySQL
+// rejects assignment to them (see buildStringMapForSet).
 func quotedColumnNames(table *TableSchema) []string {
 	cols := make([]string, 0, len(table.Columns))
-	for _, name := range table.NonGeneratedColumnNames() {
-		cols = append(cols, QuoteField(name))
+	for i := range table.Columns {
+		if table.IsColumnIndexGenerated(i) {
+			continue
+		}
+		cols = append(cols, QuoteField(table.Columns[i].Name))
 	}
 
 	return cols
@@ -355,57 +357,61 @@ func verifyValuesHasTheSameLengthAsColumns(table *TableSchema, values ...RowData
 	return nil
 }
 
+// values is a full-width row in schema order, so a single index pairs each
+// value with the column metadata appendEscapedValue needs.  Filtering the row
+// first would split the two index spaces and misalign that pairing.
 func buildStringListForValues(table *TableSchema, values []interface{}) string {
 	var buffer []byte
 
-	// values contains only non-generated columns (already filtered by the
-	// caller via FilterGeneratedColumnsOnRowData).  Build a matching list of
-	// non-generated column descriptors so that value[i] is paired with the
-	// correct column metadata regardless of where generated columns sit in the
-	// full schema.
-	nonGenerated := make([]schema.TableColumn, 0, len(table.Columns))
-	for _, col := range table.Columns {
-		if !IsColumnGenerated(&col) {
-			nonGenerated = append(nonGenerated, col)
+	for i := range table.Columns {
+		if table.IsColumnIndexGenerated(i) {
+			continue
 		}
-	}
 
-	for i, value := range values {
 		if len(buffer) > 0 {
 			buffer = append(buffer, ',')
 		}
 
-		buffer = appendEscapedValue(buffer, value, nonGenerated[i])
+		buffer = appendEscapedValue(buffer, values[i], table.Columns[i])
 	}
 
 	return string(buffer)
 }
 
-func buildStringMapForWhere(table *TableSchema, values []interface{}) string {
+// The WHERE clause of a replayed UPDATE or DELETE keeps generated columns,
+// even though the SET clause and the INSERT column list must drop them.  The
+// asymmetry is deliberate: MySQL forbids assigning to a generated column, not
+// predicating on one, and dropping them here destroys rows on the target —
+// SQL `=` compares under the column's collation, so the remaining columns
+// need not identify the row.  A STORED generated column is also often the
+// primary key, and losing it from the predicate costs a table scan per event.
+// See test/go/generated_columns_test.go for the rows this loses.
+func buildStringMapForWhere(columns []schema.TableColumn, values []interface{}) string {
 	var buffer []byte
 
 	for i, value := range values {
-		if table.IsColumnIndexGenerated(i) {
-			continue
-		}
-		if len(buffer) > 0 {
+		if i > 0 {
 			buffer = append(buffer, " AND "...)
 		}
 
-		buffer = append(buffer, QuoteField(table.Columns[i].Name)...)
+		buffer = append(buffer, QuoteField(columns[i].Name)...)
 
 		if isNilValue(value) {
 			// "WHERE value = NULL" will never match rows.
 			buffer = append(buffer, " IS NULL"...)
 		} else {
 			buffer = append(buffer, '=')
-			buffer = appendEscapedValue(buffer, value, table.Columns[i])
+			buffer = appendEscapedValue(buffer, value, columns[i])
 		}
 	}
 
 	return string(buffer)
 }
 
+// Generated columns are excluded: MySQL rejects any assignment to one with
+// error 3105, VIRTUAL and STORED alike.  The target derives them from the
+// columns that are assigned.  See buildStringMapForWhere for why the WHERE
+// clause keeps them.
 func buildStringMapForSet(table *TableSchema, values []interface{}) string {
 	var buffer []byte
 

ferry.go

@@ -59,7 +59,7 @@ type Ferry struct {
 	BinlogStreamer *BinlogStreamer
 	BinlogWriter   *BinlogWriter
 
-	targetVerifierWg *sync.WaitGroup
+	targetVerifierWg sync.WaitGroup
 	TargetVerifier   *TargetVerifier
 
 	DataIterator *DataIterator
@@ -747,7 +747,6 @@ func (f *Ferry) Run() {
 	}()
 
 	if !f.Config.SkipTargetVerification {
-		f.targetVerifierWg = &sync.WaitGroup{}
 		f.targetVerifierWg.Add(1)
 		go func() {
 			defer f.targetVerifierWg.Done()
@@ -902,12 +901,7 @@ func (f *Ferry) FlushBinlogAndStopStreaming() {
 func (f *Ferry) StopTargetVerifier() {
 	if !f.Config.SkipTargetVerification {
 		f.TargetVerifier.BinlogStreamer.FlushAndStop()
-		// targetVerifierWg is only allocated inside Run().  If the ferry exits
-		// before Run() is reached (e.g. due to an earlier error), the pointer
-		// is still nil and calling Wait() would panic.
-		if f.targetVerifierWg != nil {
-			f.targetVerifierWg.Wait()
-		}
+		f.targetVerifierWg.Wait()
 	}
 }
 

iterative_verifier.go

@@ -563,17 +563,18 @@ func (v *IterativeVerifier) tableIsIgnored(table *TableSchema) bool {
 func (v *IterativeVerifier) columnsToVerify(table *TableSchema) []schema.TableColumn {
 	ignoredColsSet, containsIgnoredColumns := v.IgnoredColumns[table.Name]
 
-	// Generated columns (VIRTUAL / STORED) are intentionally included so that
-	// any divergence in computed output between source and target is caught.
-	// Explicitly ignored columns still take priority over this inclusion.
+	// Generated columns are deliberately verified, so that divergence in
+	// computed output between source and target is caught.  Only an explicit
+	// ignore removes a column from this list.
+	if !containsIgnoredColumns {
+		return table.Columns
+	}
+
 	var columns []schema.TableColumn
 	for _, column := range table.Columns {
-		if containsIgnoredColumns {
-			if _, isIgnored := ignoredColsSet[column.Name]; isIgnored {
-				continue
-			}
+		if _, isIgnored := ignoredColsSet[column.Name]; !isIgnored {
+			columns = append(columns, column)
 		}
-		columns = append(columns, column)
 	}
 
 	return columns

row_batch.go

@@ -2,6 +2,7 @@ package ghostferry
 
 import (
 	"encoding/json"
+	"fmt"
 	"strings"
 )
 
@@ -84,24 +85,41 @@ func (e *RowBatch) Fingerprints() map[string][]byte {
 }
 
 func (e *RowBatch) AsSQLQuery(schemaName, tableName string) (string, []interface{}, error) {
-	if err := verifyValuesHasTheSameLengthAsColumns(e.table, e.values...); err != nil {
-		return "", nil, err
+	for _, row := range e.values {
+		if len(e.columns) != len(row) {
+			return "", nil, fmt.Errorf(
+				"table %s.%s has %d selected columns but row has %d values",
+				e.table.Schema,
+				e.table.Name,
+				len(e.columns),
+				len(row),
+			)
+		}
 	}
 
-	// Build the INSERT column list from e.columns — the actual query-result
-	// order — skipping generated columns by precomputed index.
-	//
-	// We must NOT use table.NonGeneratedColumnNames() here because that
-	// always returns schema order.  When the SELECT query returns columns in a
-	// different order (for example, the sharding copy filter uses
-	//   SELECT * FROM t JOIN (SELECT id …) AS batch USING(id)
-	// which moves 'id' to the front), the column names and row values would
-	// be misaligned, corrupting every row written to the target.
+	// The INSERT column list must follow e.columns — the order the SELECT
+	// returned — not schema order.  The two differ under the sharding copy
+	// filter, whose JOIN ... USING moves the join column to the front; naming
+	// columns in schema order against result-ordered values silently writes
+	// every value into the wrong column (the gh-285 corruption pattern).
 	insertColumns := make([]string, 0, len(e.nonGeneratedColumnIdxs))
 	for _, i := range e.nonGeneratedColumnIdxs {
 		insertColumns = append(insertColumns, e.columns[i])
 	}
 
+	// LoadTables refuses a table with no writable columns, but a CopyFilter
+	// narrowing ColumnsToSelect, or an embedder populating Ferry.Tables
+	// directly, can still arrive here with nothing to write.  Without this
+	// check, strings.Repeat below panics on a negative count.
+	if len(insertColumns) == 0 {
+		return "", nil, fmt.Errorf(
+			"table %s.%s has no columns to write: every selected column (%v) is a generated column",
+			e.table.Schema,
+			e.table.Name,
+			e.columns,
+		)
+	}
+
 	valuesStr := "(" + strings.Repeat("?,", len(insertColumns)-1) + "?)"
 	valuesStr = strings.Repeat(valuesStr+",", len(e.values)-1) + valuesStr