Feature: Handle read only columns

CHANGELOG.md

@@ -2,6 +2,11 @@
 
 All notable changes to this project will be documented in this file.
 
+## [1.4.0 - 2026-05-18]
+
+### Added
+- Support generated/readonly columns by @plisandro, @driv3r, @grodowski in #437
+
 ## [1.3.1 - 2026-04-15]
 
 ### Changed

Gemfile

@@ -18,3 +18,4 @@ group :development do
 end
 
 gem "mutex_m", "~> 0.3.0"
+gem "logger"

Gemfile.lock

@@ -8,6 +8,7 @@ GEM
       reline (>= 0.6.0)
     coderay (1.1.3)
     io-console (0.8.2)
+    logger (1.7.0)
     method_source (1.1.0)
     minitest (5.27.0)
     minitest-fail-fast (0.1.0)
@@ -42,6 +43,7 @@ PLATFORMS
   ruby
 
 DEPENDENCIES
+  logger
   minitest
   minitest-fail-fast (~> 0.1.0)
   minitest-hooks

Makefile

@@ -1,5 +1,5 @@
 # Variables to be built into the binary
-VERSION         := 1.3.1
+VERSION         := 1.4.0
 
 # This variable can be overwritten by the caller
 DATETIME        ?= $(shell date -u +%Y%m%d%H%M%S)

cursor.go

@@ -268,12 +268,7 @@ func (c *Cursor) Fetch(db SqlPreparer) (batch *RowBatch, paginationKeypos Pagina
 		}
 	}
 
-	batch = &RowBatch{
-		values:             batchData,
-		paginationKeyIndex: paginationKeyIndex,
-		table:              c.Table,
-		columns:            columns,
-	}
+	batch = NewRowBatchWithColumns(c.Table, batchData, columns, paginationKeyIndex)
 
 	logger.Debugf("found %d rows", batch.Size())
 

data_iterator.go

@@ -122,13 +122,8 @@ func (d *DataIterator) Run(tables []*TableSchema) {
 							rows[i] = rowData[:len(rowData)-1]
 						}
 
-						batch = &RowBatch{
-							values:             rows,
-							paginationKeyIndex: batch.PaginationKeyIndex(),
-							table:              table,
-							fingerprints:       fingerprints,
-							columns:            batch.columns[:len(batch.columns)-1],
-						}
+						batch = NewRowBatchWithColumns(table, rows, batch.columns[:len(batch.columns)-1], batch.PaginationKeyIndex())
+						batch.fingerprints = fingerprints
 					}
 
 					for _, listener := range d.batchListeners {

dml_events.go

@@ -42,14 +42,14 @@ type RowData []interface{}
 // https://github.com/Shopify/ghostferry/issues/165.
 //
 // In summary:
-// - This code receives values from both go-sql-driver/mysql and
-//   go-mysql-org/go-mysql.
-// - go-sql-driver/mysql gives us int64 for signed integer, and uint64 in a byte
-//   slice for unsigned integer.
-// - go-mysql-org/go-mysql gives us int64 for signed integer, and uint64 for
-//   unsigned integer.
-// - We currently make this function deal with both cases. In the future we can
-//   investigate alternative solutions.
+//   - This code receives values from both go-sql-driver/mysql and
+//     go-mysql-org/go-mysql.
+//   - go-sql-driver/mysql gives us int64 for signed integer, and uint64 in a byte
+//     slice for unsigned integer.
+//   - go-mysql-org/go-mysql gives us int64 for signed integer, and uint64 for
+//     unsigned integer.
+//   - We currently make this function deal with both cases. In the future we can
+//     investigate alternative solutions.
 func (r RowData) GetUint64(colIdx int) (uint64, error) {
 	u64, ok := Uint64Value(r[colIdx])
 	if ok {
@@ -168,14 +168,15 @@ func (e *BinlogInsertEvent) NewValues() RowData {
 }
 
 func (e *BinlogInsertEvent) AsSQLString(schemaName, tableName string) (string, error) {
-	if err := verifyValuesHasTheSameLengthAsColumns(e.table, e.newValues); err != nil {
+	filteredNewValues, err := e.table.FilterGeneratedColumnsOnRowData(e.newValues)
+	if err != nil {
 		return "", err
 	}
 
 	query := "INSERT IGNORE INTO " +
 		QuotedTableNameFromString(schemaName, tableName) +
 		" (" + strings.Join(quotedColumnNames(e.table), ",") + ")" +
-		" VALUES (" + buildStringListForValues(e.table.Columns, e.newValues) + ")"
+		" VALUES (" + buildStringListForValues(e.table, filteredNewValues) + ")"
 
 	return query, nil
 }
@@ -227,8 +228,8 @@ func (e *BinlogUpdateEvent) AsSQLString(schemaName, tableName string) (string, e
 	}
 
 	query := "UPDATE " + QuotedTableNameFromString(schemaName, tableName) +
-		" SET " + buildStringMapForSet(e.table.Columns, e.newValues) +
-		" WHERE " + buildStringMapForWhere(e.table.Columns, e.oldValues)
+		" SET " + buildStringMapForSet(e.table, e.newValues) +
+		" WHERE " + buildStringMapForWhere(e.table, e.oldValues)
 
 	return query, nil
 }
@@ -269,7 +270,7 @@ func (e *BinlogDeleteEvent) AsSQLString(schemaName, tableName string) (string, e
 	}
 
 	query := "DELETE FROM " + QuotedTableNameFromString(schemaName, tableName) +
-		" WHERE " + buildStringMapForWhere(e.table.Columns, e.oldValues)
+		" WHERE " + buildStringMapForWhere(e.table, e.oldValues)
 
 	return query, nil
 }
@@ -281,32 +282,39 @@ 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 _, row := range rowsEvent.Rows {
-		if len(row) != len(table.Columns) {
+	for i, rawRow := range rowsEvent.Rows {
+		if len(rawRow) != 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(row),
+				len(rawRow),
 			)
 		}
-		for i, col := range table.Columns {
+
+		// 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 {
 			if col.IsUnsigned {
-				switch v := row[i].(type) {
+				switch v := rawRow[j].(type) {
 				case int64:
-					row[i] = uint64(v)
+					rawRow[j] = uint64(v)
 				case int32:
-					row[i] = uint32(v)
+					rawRow[j] = uint32(v)
 				case int16:
-					row[i] = uint16(v)
+					rawRow[j] = uint16(v)
 				case int8:
-					row[i] = uint8(v)
+					rawRow[j] = uint8(v)
 				case int:
-					row[i] = uint(v)
+					rawRow[j] = uint(v)
 				}
 			}
 		}
+		rowsEvent.Rows[i] = rawRow
 	}
 
 	timestamp := time.Unix(int64(ev.Header.Timestamp), 0)
@@ -324,9 +332,9 @@ func NewBinlogDMLEvents(table *TableSchema, ev *replication.BinlogEvent, pos, re
 }
 
 func quotedColumnNames(table *TableSchema) []string {
-	cols := make([]string, len(table.Columns))
-	for i, column := range table.Columns {
-		cols[i] = QuoteField(column.Name)
+	cols := make([]string, 0, len(table.Columns))
+	for _, name := range table.NonGeneratedColumnNames() {
+		cols = append(cols, QuoteField(name))
 	}
 
 	return cols
@@ -347,53 +355,71 @@ func verifyValuesHasTheSameLengthAsColumns(table *TableSchema, values ...RowData
 	return nil
 }
 
-func buildStringListForValues(columns []schema.TableColumn, values []interface{}) string {
+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, value := range values {
-		if i > 0 {
+		if len(buffer) > 0 {
 			buffer = append(buffer, ',')
 		}
 
-		buffer = appendEscapedValue(buffer, value, columns[i])
+		buffer = appendEscapedValue(buffer, value, nonGenerated[i])
 	}
 
 	return string(buffer)
 }
 
-func buildStringMapForWhere(columns []schema.TableColumn, values []interface{}) string {
+func buildStringMapForWhere(table *TableSchema, values []interface{}) string {
 	var buffer []byte
 
 	for i, value := range values {
-		if i > 0 {
+		if table.IsColumnIndexGenerated(i) {
+			continue
+		}
+		if len(buffer) > 0 {
 			buffer = append(buffer, " AND "...)
 		}
 
-		buffer = append(buffer, QuoteField(columns[i].Name)...)
+		buffer = append(buffer, QuoteField(table.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, columns[i])
+			buffer = appendEscapedValue(buffer, value, table.Columns[i])
 		}
 	}
 
 	return string(buffer)
 }
 
-func buildStringMapForSet(columns []schema.TableColumn, values []interface{}) string {
+func buildStringMapForSet(table *TableSchema, values []interface{}) string {
 	var buffer []byte
 
 	for i, value := range values {
-		if i > 0 {
+		if table.IsColumnIndexGenerated(i) {
+			continue
+		}
+		if len(buffer) > 0 {
 			buffer = append(buffer, ',')
 		}
 
-		buffer = append(buffer, QuoteField(columns[i].Name)...)
+		buffer = append(buffer, QuoteField(table.Columns[i].Name)...)
 		buffer = append(buffer, '=')
-		buffer = appendEscapedValue(buffer, value, columns[i])
+		buffer = appendEscapedValue(buffer, value, table.Columns[i])
 	}
 
 	return string(buffer)
@@ -504,10 +530,10 @@ func Int64Value(value interface{}) (int64, bool) {
 //
 // This is specifically mentioned in the the below link:
 //
-//    When BINARY values are stored, they are right-padded with the pad value
-//    to the specified length. The pad value is 0x00 (the zero byte). Values
-//    are right-padded with 0x00 for inserts, and no trailing bytes are removed
-//    for retrievals.
+//	When BINARY values are stored, they are right-padded with the pad value
+//	to the specified length. The pad value is 0x00 (the zero byte). Values
+//	are right-padded with 0x00 for inserts, and no trailing bytes are removed
+//	for retrievals.
 //
 // ref: https://dev.mysql.com/doc/refman/5.7/en/binary-varbinary.html
 func appendEscapedString(buffer []byte, value string, rightPadToLengthWithZeroBytes int) []byte {

ferry.go

@@ -902,7 +902,12 @@ func (f *Ferry) FlushBinlogAndStopStreaming() {
 func (f *Ferry) StopTargetVerifier() {
 	if !f.Config.SkipTargetVerification {
 		f.TargetVerifier.BinlogStreamer.FlushAndStop()
-		f.targetVerifierWg.Wait()
+		// 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()
+		}
 	}
 }
 

iterative_verifier.go

@@ -562,15 +562,18 @@ func (v *IterativeVerifier) tableIsIgnored(table *TableSchema) bool {
 
 func (v *IterativeVerifier) columnsToVerify(table *TableSchema) []schema.TableColumn {
 	ignoredColsSet, containsIgnoredColumns := v.IgnoredColumns[table.Name]
-	if !containsIgnoredColumns {
-		return table.Columns
-	}
 
+	// 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.
 	var columns []schema.TableColumn
 	for _, column := range table.Columns {
-		if _, isIgnored := ignoredColsSet[column.Name]; !isIgnored {
-			columns = append(columns, column)
+		if containsIgnoredColumns {
+			if _, isIgnored := ignoredColsSet[column.Name]; isIgnored {
+				continue
+			}
 		}
+		columns = append(columns, column)
 	}
 
 	return columns