diff --git a/CHANGELOG.md b/CHANGELOG.md index 05e6d7ba..c16b919c 100644 --- a/CHANGELOG.md +++ b/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] diff --git a/dml_events.go b/dml_events.go index 7070a3ec..632082a7 100644 --- a/dml_events.go +++ b/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 diff --git a/ferry.go b/ferry.go index a5092cee..338fcf77 100644 --- a/ferry.go +++ b/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() } } diff --git a/iterative_verifier.go b/iterative_verifier.go index 8c3fff8d..f05f1933 100644 --- a/iterative_verifier.go +++ b/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 diff --git a/row_batch.go b/row_batch.go index b15735b1..b6efa2e9 100644 --- a/row_batch.go +++ b/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 diff --git a/table_schema_cache.go b/table_schema_cache.go index 91f87ef9..b5cafc0a 100644 --- a/table_schema_cache.go +++ b/table_schema_cache.go @@ -45,7 +45,7 @@ type TableSchema struct { rowMd5Query string } -// IsColumnGenerated evaluates whether a go_myslq.schema.TableColumn is generated or not. +// IsColumnGenerated evaluates whether a go-mysql schema.TableColumn is generated or not. func IsColumnGenerated(tc *schema.TableColumn) bool { return tc.IsVirtual || tc.IsStored } @@ -55,23 +55,23 @@ func (t *TableSchema) IsColumnIndexGenerated(idx int) bool { return IsColumnGenerated(&t.Columns[idx]) } -// Evaluates whether a TableSchema column is generated, by name. +// IsColumnNameGenerated evaluates whether a TableSchema column is generated, by name. func (t *TableSchema) IsColumnNameGenerated(name string) bool { - for _, col := range t.Columns { - if name == col.Name && IsColumnGenerated(&col) { - return true + for i := range t.Columns { + if name == t.Columns[i].Name { + return IsColumnGenerated(&t.Columns[i]) } } return false } -// Returns a count of total, generated and non-generated columns for a TableSchema. +// ColumnsCount returns a count of total, generated and non-generated columns for a TableSchema. func (t *TableSchema) ColumnsCount() (int, int, int) { var generated int - for _, col := range t.Columns { - if IsColumnGenerated(&col) { + for i := range t.Columns { + if IsColumnGenerated(&t.Columns[i]) { generated += 1 } } @@ -79,55 +79,6 @@ func (t *TableSchema) ColumnsCount() (int, int, int) { return len(t.Columns), generated, len(t.Columns) - generated } -// Returns a list of all non-generated column names for a TableSchema, in schema order. -func (t *TableSchema) NonGeneratedColumnNames() []string { - res := make([]string, 0, len(t.Columns)) - - for _, col := range t.Columns { - if IsColumnGenerated(&col) { - continue - } - res = append(res, col.Name) - } - - return res -} - -// FilterGeneratedColumnsOnRowData takes a row (as slice of RowData elements) and returns -// a copy with elements for generated columns removed. -func (t *TableSchema) FilterGeneratedColumnsOnRowData(row []interface{}) ([]interface{}, error) { - columnsCount, _, nonGeneratedColumnsCount := t.ColumnsCount() - - if len(row) != columnsCount { - return nil, fmt.Errorf( - "table %s.%s has %d columns but row has %d columns instead", - t.Schema, - t.Name, - columnsCount, - len(row), - ) - } - - res := make([]interface{}, 0, len(row)) - for i, val := range row { - if t.IsColumnIndexGenerated(i) { - continue - } - res = append(res, val) - } - - if len(res) != nonGeneratedColumnsCount { - return nil, fmt.Errorf( - "table %s.%s has %d updatable columns but processed row has %d updatable columns instead", - t.Schema, - t.Name, - nonGeneratedColumnsCount, - len(res), - ) - } - return res, nil -} - // This query returns the MD5 hash for a row on this table. This query is valid // for both the source and the target shard. // @@ -137,11 +88,6 @@ func (t *TableSchema) FilterGeneratedColumnsOnRowData(row []interface{}) ([]inte // Any columns specified in IgnoredColumnsForVerification are excluded from the // checksum and the raw data will not be returned. // -// Generated columns (VIRTUAL and STORED) are included in the checksum so that -// any divergence in computed output between source and target is also caught. -// An operator can still opt out of checking a specific generated column by -// adding it to IgnoredColumnsForVerification. -// // Note that the MD5 hash should consists of at least 1 column: the paginationKey column. // This is to say that there should never be a case where the MD5 hash is // derived from an empty string. @@ -242,17 +188,15 @@ func MaxPaginationKeys(db *sql.DB, tables []*TableSchema, logger Logger) (map[*T return tablesWithData, emptyTables, nil } -// removeInvisibleIndeces removes all invisible idx references from a go_mysql.schema.Table. +// removeInvisibleIndexes removes all invisible index references from a go-mysql schema.Table. func removeInvisibleIndexes(ts *schema.Table) { - j := 0 - for i, index := range ts.Indexes { - if !index.Visible { - continue + visibleIndexes := make([]*schema.Index, 0, len(ts.Indexes)) + for _, index := range ts.Indexes { + if index.Visible { + visibleIndexes = append(visibleIndexes, index) } - ts.Indexes[j] = ts.Indexes[i] - j++ } - ts.Indexes = ts.Indexes[:j] + ts.Indexes = visibleIndexes } func LoadTables(db *sql.DB, tableFilter TableFilter, columnCompressionConfig ColumnCompressionConfig, columnIgnoreConfig ColumnIgnoreConfig, forceIndexConfig ForceIndexConfig, cascadingPaginationColumnConfig *CascadingPaginationColumnConfig) (TableSchemaCache, error) { @@ -293,7 +237,6 @@ func LoadTables(db *sql.DB, tableFilter TableFilter, columnCompressionConfig Col return tableSchemaCache, err } - // filter out unwanted indeces and columns removeInvisibleIndexes(tableSchema) tableSchemas = append(tableSchemas, &TableSchema{ @@ -314,7 +257,16 @@ func LoadTables(db *sql.DB, tableFilter TableFilter, columnCompressionConfig Col tableLog := dbLog.WithField("table", tableName) tableLog.Debug("caching table schema") - paginationKeyColumn, paginationKeyIndex, err := tableSchema.paginationKeyColumn(cascadingPaginationColumnConfig) + // MySQL permits a table whose every column is generated, but + // Ghostferry would have nothing it is allowed to write. Refusing it + // here beats failing part-way through a move. + if _, _, nonGeneratedColumnsCount := tableSchema.ColumnsCount(); nonGeneratedColumnsCount == 0 { + err := NoNonGeneratedColumnsError(tableSchema.Schema, tableSchema.Name) + tableLog.WithError(err).Error("invalid table") + return tableSchemaCache, err + } + + paginationKeyColumn, paginationKeyIndex, err := tableSchema.paginationKeyColumn(db, cascadingPaginationColumnConfig) if err != nil { logger.WithError(err).Error("invalid table") return tableSchemaCache, err @@ -360,12 +312,17 @@ func NonBinaryCollationError(schema, table, paginationKey, collation string) err return fmt.Errorf("Pagination Key `%s` for %s has non-binary collation '%s'. Binary columns (BINARY, VARBINARY) or string columns with binary collation (e.g., utf8mb4_bin) are required to ensure consistent ordering between MySQL and Ghostferry", paginationKey, QuotedTableNameFromString(schema, table), collation) } +// NoNonGeneratedColumnsError exported to facilitate black box testing +func NoNonGeneratedColumnsError(schema, table string) error { + return fmt.Errorf("%s has no columns that Ghostferry can write: every column is a generated column, and MySQL rejects assignment to those, so there would be nothing to send to the target. Exclude this table from the move", QuotedTableNameFromString(schema, table)) +} + // VirtualPaginationKeyError exported to facilitate black box testing func VirtualPaginationKeyError(schema, table, paginationKey string) error { - return fmt.Errorf("Pagination Key `%s` for %s is a VIRTUAL generated column. VIRTUAL columns are not stored on disk, so their values are unavailable during data iteration. Use a real column or a STORED generated column as the Pagination Key instead", paginationKey, QuotedTableNameFromString(schema, table)) + return fmt.Errorf("Pagination Key `%s` for %s is a VIRTUAL generated column that is not declared NOT NULL with a visible single-column UNIQUE index. Add those constraints, or use a real column or STORED generated column as the Pagination Key instead", paginationKey, QuotedTableNameFromString(schema, table)) } -func (t *TableSchema) paginationKeyColumn(cascadingPaginationColumnConfig *CascadingPaginationColumnConfig) (*schema.TableColumn, int, error) { +func (t *TableSchema) paginationKeyColumn(db *sql.DB, cascadingPaginationColumnConfig *CascadingPaginationColumnConfig) (*schema.TableColumn, int, error) { var err error var paginationKeyColumn *schema.TableColumn var paginationKeyIndex int @@ -386,11 +343,32 @@ func (t *TableSchema) paginationKeyColumn(cascadingPaginationColumnConfig *Casca } if paginationKeyColumn != nil { - // VIRTUAL generated columns are not stored on disk and cannot be used for - // data iteration. STORED generated columns are physically stored and are - // safe to use. if paginationKeyColumn.IsVirtual { - return nil, -1, VirtualPaginationKeyError(t.Schema, t.Name, paginationKeyColumn.Name) + hasUniqueIndex := false + for _, index := range t.Indexes { + if index.NoneUnique == 0 && len(index.Columns) == 1 && index.Columns[0] == paginationKeyColumn.Name { + hasUniqueIndex = true + break + } + } + + if !hasUniqueIndex { + return nil, -1, VirtualPaginationKeyError(t.Schema, t.Name, paginationKeyColumn.Name) + } + + var isNullable string + err = db.QueryRow( + "SELECT IS_NULLABLE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ? AND COLUMN_NAME = ?", + t.Schema, + t.Name, + paginationKeyColumn.Name, + ).Scan(&isNullable) + if err != nil { + return nil, -1, err + } + if strings.EqualFold(isNullable, "YES") { + return nil, -1, VirtualPaginationKeyError(t.Schema, t.Name, paginationKeyColumn.Name) + } } isNumber := paginationKeyColumn.Type == schema.TYPE_NUMBER || paginationKeyColumn.Type == schema.TYPE_MEDIUM_INT diff --git a/test/go/dml_events_test.go b/test/go/dml_events_test.go index 535f60d7..50a4ae5e 100644 --- a/test/go/dml_events_test.go +++ b/test/go/dml_events_test.go @@ -95,7 +95,7 @@ func (this *DMLEventsTestSuite) TestBinlogInsertEventWithWrongColumnsReturnsErro _, err = dmlEvents[0].AsSQLString(this.targetTable.Schema, this.targetTable.Name) this.Require().NotNil(err) - this.Require().Contains(err.Error(), "test_table has 3 columns but row has 1 column") + this.Require().Contains(err.Error(), "test_table has 3 columns but event has 1 column") } func (this *DMLEventsTestSuite) TestBinlogInsertEventMetadata() { @@ -390,18 +390,14 @@ func (this *DMLEventsTestSuite) TestNoRowsQueryEvent() { this.Require().Equal("", annotation) } -// TestNewBinlogDMLEventsUnsignedConversionWithGeneratedColumn exercises two bugs -// that arise when a virtual column sits before an unsigned integer column. -// -// Bug 1 (panic / index out of range): the second commit of the PR compacts the -// raw row by removing generated column values BEFORE applying the unsigned-integer -// normalisation loop. The loop still iterates over table.Columns (full length), -// so row[i] for the unsigned column indexes into the shortened slice and panics. -// -// Bug 2 (wrong value): if the panic is suppressed or the generated column happens -// to be last, the unsigned normalisation is applied to the wrong row position, -// leaving int8(-1) serialised as -1 instead of the correct uint8(255). +// TestNewBinlogDMLEventsUnsignedConversionWithGeneratedColumn pins unsigned +// normalisation to full-schema indexes. Compacting the row before +// normalising would read the wrong positions, leaving int8(-1) serialised as +// -1 rather than uint8(255). Filtering happens at SQL construction, never +// before. func (this *DMLEventsTestSuite) TestNewBinlogDMLEventsUnsignedConversionWithGeneratedColumn() { + // 'gen' must sit BEFORE 'u8': the index spaces only diverge after the + // first generated column, so with 'gen' last the test detects nothing. columns := []schema.TableColumn{ {Name: "id"}, {Name: "gen", IsVirtual: true}, @@ -436,16 +432,14 @@ func (this *DMLEventsTestSuite) TestNewBinlogDMLEventsUnsignedConversionWithGene ) } -// TestBinlogInsertEventGeneratedColumnBeforeJSONPreservesJSONCasting exercises -// the metadata misalignment in buildStringListForValues introduced by the PR. -// -// AsSQLString filters generated column values out of the row before passing the -// shortened slice to buildStringListForValues. That function then uses the loop -// counter i to index table.Columns, so the JSON column's value (at position 0 -// in the filtered slice) is looked up against table.Columns[0] — the virtual -// column — which has no JSON type. As a result the value is emitted as a plain -// escaped string instead of CAST(... AS JSON). +// TestBinlogInsertEventGeneratedColumnBeforeJSONPreservesJSONCasting pins the +// pairing between a value and the column metadata used to escape it. Pairing +// a filtered row against the full schema would escape this JSON payload with +// the virtual column's metadata and emit a plain quoted string instead of +// CAST(... AS JSON). func (this *DMLEventsTestSuite) TestBinlogInsertEventGeneratedColumnBeforeJSONPreservesJSONCasting() { + // 'gen' must sit BEFORE 'payload': the index spaces only diverge after the + // first generated column, so with 'gen' last the test detects nothing. columns := []schema.TableColumn{ {Name: "gen", IsVirtual: true}, {Name: "payload", Type: schema.TYPE_JSON}, @@ -476,10 +470,10 @@ func (this *DMLEventsTestSuite) TestBinlogInsertEventGeneratedColumnBeforeJSONPr ) } -// TestBinlogUpdateEventExcludesGeneratedColumnFromSetAndWhere verifies that -// UPDATE events for tables with virtual columns emit SET and WHERE clauses that -// reference only real (non-generated) columns. -func (this *DMLEventsTestSuite) TestBinlogUpdateEventExcludesGeneratedColumnFromSetAndWhere() { +// TestBinlogUpdateEventExcludesGeneratedColumnFromSetOnly pins the asymmetry +// of a replayed UPDATE: SET must omit the generated column, WHERE must keep +// it. See buildStringMapForWhere in dml_events.go. +func (this *DMLEventsTestSuite) TestBinlogUpdateEventExcludesGeneratedColumnFromSetOnly() { columns := []schema.TableColumn{ {Name: "id"}, {Name: "gen", IsVirtual: true}, @@ -509,14 +503,15 @@ func (this *DMLEventsTestSuite) TestBinlogUpdateEventExcludesGeneratedColumnFrom q, err := dmlEvents[0].AsSQLString("test_schema", "test_table") this.Require().Nil(err) this.Require().Equal( - "UPDATE `test_schema`.`test_table` SET `id`=1000,`data`='new_data' WHERE `id`=1000 AND `data`='old_data'", + "UPDATE `test_schema`.`test_table` SET `id`=1000,`data`='new_data' WHERE `id`=1000 AND `gen`='gen_old' AND `data`='old_data'", q, ) } -// TestBinlogDeleteEventExcludesStoredGeneratedColumnFromWhere verifies that -// DELETE events skip both VIRTUAL and STORED generated columns in the WHERE clause. -func (this *DMLEventsTestSuite) TestBinlogDeleteEventExcludesStoredGeneratedColumnFromWhere() { +// TestBinlogDeleteEventKeepsStoredGeneratedColumnInWhere pins the DELETE +// predicate. A STORED generated column may be the primary key, so dropping +// it from the WHERE clause can delete rows never deleted on the source. +func (this *DMLEventsTestSuite) TestBinlogDeleteEventKeepsStoredGeneratedColumnInWhere() { columns := []schema.TableColumn{ {Name: "id"}, {Name: "data"}, @@ -543,7 +538,7 @@ func (this *DMLEventsTestSuite) TestBinlogDeleteEventExcludesStoredGeneratedColu q, err := dmlEvents[0].AsSQLString("test_schema", "test_table") this.Require().Nil(err) this.Require().Equal( - "DELETE FROM `test_schema`.`test_table` WHERE `id`=1000 AND `data`='hello'", + "DELETE FROM `test_schema`.`test_table` WHERE `id`=1000 AND `data`='hello' AND `summary`='abc123'", q, ) } diff --git a/test/go/generated_columns_test.go b/test/go/generated_columns_test.go new file mode 100644 index 00000000..f5b2a704 --- /dev/null +++ b/test/go/generated_columns_test.go @@ -0,0 +1,538 @@ +package test + +import ( + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/Shopify/ghostferry" + sql "github.com/Shopify/ghostferry/sqlwrapper" + "github.com/Shopify/ghostferry/testhelpers" + + "github.com/stretchr/testify/suite" +) + +// GeneratedColumnsTestSuite exercises generated-column handling end to end +// against real MySQL servers: real DDL, real binlog row images, and the +// generated SQL executed against a real target. The hand-built TableSchema +// tests elsewhere in this package test our logic; these test our premises — +// that MySQL sets the flags, that the values survive the binlog, and that a +// real target accepts the SQL we emit. +type GeneratedColumnsTestSuite struct { + *testhelpers.GhostferryUnitTestSuite + + sourceDB *sql.DB +} + +func (this *GeneratedColumnsTestSuite) SetupTest() { + this.GhostferryUnitTestSuite.SetupTest() + + // The binlog streamer needs its own connection: it hands it to the + // replication client for the lifetime of the stream. + testFerry := testhelpers.NewTestFerry() + sourceConfig, err := testFerry.Source.MySQLConfig() + this.Require().Nil(err) + + this.sourceDB, err = sql.Open("mysql", sourceConfig.FormatDSN(), testFerry.Source.Marginalia) + this.Require().Nil(err) + + _, err = this.Ferry.SourceDB.Exec(fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`", testhelpers.TestSchemaName)) + this.Require().Nil(err) + _, err = this.Ferry.TargetDB.Exec(fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`", testhelpers.TestSchemaName)) + this.Require().Nil(err) +} + +func (this *GeneratedColumnsTestSuite) TearDownTest() { + if this.sourceDB != nil { + this.sourceDB.Close() + } + this.GhostferryUnitTestSuite.TearDownTest() +} + +func (this *GeneratedColumnsTestSuite) createOnBothSides(ddl string) { + _, err := this.Ferry.SourceDB.Exec(ddl) + this.Require().Nil(err) + _, err = this.Ferry.TargetDB.Exec(ddl) + this.Require().Nil(err) +} + +func (this *GeneratedColumnsTestSuite) execOnBothSides(query string, args ...interface{}) { + _, err := this.Ferry.SourceDB.Exec(query, args...) + this.Require().Nil(err) + _, err = this.Ferry.TargetDB.Exec(query, args...) + this.Require().Nil(err) +} + +func (this *GeneratedColumnsTestSuite) loadTable(tableName string) *ghostferry.TableSchema { + cache := this.loadTables() + table := cache.Get(testhelpers.TestSchemaName, tableName) + this.Require().NotNil(table) + return table +} + +func (this *GeneratedColumnsTestSuite) loadTables() ghostferry.TableSchemaCache { + cache, err := ghostferry.LoadTables( + this.Ferry.SourceDB, + &testhelpers.TestTableFilter{ + DbsFunc: testhelpers.DbApplicabilityFilter([]string{testhelpers.TestSchemaName}), + TablesFunc: nil, + }, + nil, nil, nil, nil, + ) + this.Require().Nil(err) + return cache +} + +// captureBinlogEvents streams the source binlog while mutate() runs, and +// returns the DMLEvents Ghostferry decodes from it — row images produced by +// MySQL, not by us. +func (this *GeneratedColumnsTestSuite) captureBinlogEvents(mutate func()) []ghostferry.DMLEvent { + testFerry := testhelpers.NewTestFerry() + streamer := &ghostferry.BinlogStreamer{ + DB: this.sourceDB, + DBConfig: testFerry.Config.Source, + MyServerId: testFerry.Config.MyServerId, + ErrorHandler: testFerry.ErrorHandler, + TableSchema: this.loadTables(), + } + + _, err := streamer.ConnectBinlogStreamerToMysql() + this.Require().Nil(err) + + var captured []ghostferry.DMLEvent + streamer.AddEventListener(func(evs []ghostferry.DMLEvent) error { + captured = append(captured, evs...) + streamer.FlushAndStop() + return nil + }) + + done := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + streamer.Run() + }() + + mutate() + + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(30 * time.Second): + this.Require().FailNow("timed out waiting for binlog events") + } + + return captured +} + +// applyToTarget replays events against the target exactly as BinlogWriter does: +// render each event to SQL and execute it. +func (this *GeneratedColumnsTestSuite) applyToTarget(events []ghostferry.DMLEvent) { + for _, ev := range events { + stmt, err := ev.AsSQLString(ev.Database(), ev.Table()) + this.Require().Nil(err, "rendering event to SQL") + + _, err = this.Ferry.TargetDB.Exec(stmt) + this.Require().Nil(err, fmt.Sprintf("executing replayed statement on target: %s", stmt)) + } +} + +func (this *GeneratedColumnsTestSuite) sourceStrings(query string) []string { + return this.queryStrings(this.Ferry.SourceDB, query) +} + +func (this *GeneratedColumnsTestSuite) targetStrings(query string) []string { + return this.queryStrings(this.Ferry.TargetDB, query) +} + +func (this *GeneratedColumnsTestSuite) queryStrings(db *sql.DB, query string) []string { + rows, err := db.Query(query) + this.Require().Nil(err) + defer rows.Close() + + var out []string + for rows.Next() { + var v string + this.Require().Nil(rows.Scan(&v)) + out = append(out, v) + } + this.Require().Nil(rows.Err()) + return out +} + +// docsDDL is the content-addressed table from the PR #437 review: the primary +// key is a hash of the body, so nothing else can tell two rows apart. The +// collation is pinned so the test does not depend on the server default. +const docsDDL = "CREATE TABLE %s.docs (" + + "doc TEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL," + + "doc_hash BINARY(32) AS (UNHEX(SHA2(doc, 256))) STORED," + + "PRIMARY KEY (doc_hash))" + +func (this *GeneratedColumnsTestSuite) seedDocs() { + this.createOnBothSides(fmt.Sprintf(docsDDL, testhelpers.TestSchemaName)) + // Four rows, four distinct hashes, one equivalence class under `=` on + // `doc`: utf8mb4_unicode_ci is case-insensitive, accent-insensitive and + // PAD SPACE. + this.execOnBothSides(fmt.Sprintf( + "INSERT INTO %s.docs (doc) VALUES ('cafe'), ('caf\u00e9'), ('CAFE'), ('cafe ')", + testhelpers.TestSchemaName, + )) +} + +// assertDocsMatch asserts the target is indistinguishable from the source, +// catching over- and under-deletion in one order-independent assertion. +func (this *GeneratedColumnsTestSuite) assertDocsMatch() { + query := fmt.Sprintf( + "SELECT doc, HEX(doc_hash) AS h FROM %s.docs ORDER BY h", + testhelpers.TestSchemaName, + ) + testhelpers.AssertTwoQueriesHaveEqualResult(this.T(), this.Ferry, query, query) +} + +func (this *GeneratedColumnsTestSuite) targetDocCount() int { + return len(this.targetStrings(fmt.Sprintf("SELECT doc FROM %s.docs", testhelpers.TestSchemaName))) +} + +// sourceCount guards against a vacuous pass: if a mutation silently failed on +// the source, source and target would agree on the unchanged data. +func (this *GeneratedColumnsTestSuite) sourceCount(table string) int { + var n int + row := this.Ferry.SourceDB.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM %s.%s", testhelpers.TestSchemaName, table)) + this.Require().Nil(row.Scan(&n)) + return n +} + +// TestBinlogDeleteAffectsExactlyTheRowThatWasDeleted is the data-loss +// regression test for the WHERE clause. SQL `=` compares under the column's +// collation, so without `doc_hash` in the predicate a one-row delete on the +// source becomes a multi-row delete on the target. +func (this *GeneratedColumnsTestSuite) TestBinlogDeleteAffectsExactlyTheRowThatWasDeleted() { + this.seedDocs() + + events := this.captureBinlogEvents(func() { + _, err := this.Ferry.SourceDB.Exec(fmt.Sprintf( + "DELETE FROM %s.docs WHERE doc_hash = UNHEX(SHA2('cafe', 256))", + testhelpers.TestSchemaName, + )) + this.Require().Nil(err) + }) + this.Require().Equal(1, len(events)) + + this.applyToTarget(events) + + this.Require().Equal(3, this.targetDocCount(), + "replaying a one-row DELETE must not remove rows that merely compare equal under the collation") + this.assertDocsMatch() +} + +// TestBinlogDeleteOnCompositeKeyAffectsExactlyOneRow is the same failure in +// an ordinary-looking table. The exposure is not limited to "the primary key +// IS a generated column": it is that after removing every generated column, +// no remaining subset forms a unique key. Here `tenant`, `label` and +// `payload` are all in the WHERE clause and still fail to separate the rows. +func (this *GeneratedColumnsTestSuite) TestBinlogDeleteOnCompositeKeyAffectsExactlyOneRow() { + this.createOnBothSides(fmt.Sprintf( + "CREATE TABLE %s.composite ("+ + "label VARCHAR(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,"+ + "tenant INT NOT NULL,"+ + "payload INT NOT NULL,"+ + "label_hash BINARY(32) AS (UNHEX(SHA2(label, 256))) STORED,"+ + "PRIMARY KEY (label_hash),"+ + "UNIQUE KEY u (tenant, label_hash))", + testhelpers.TestSchemaName, + )) + this.execOnBothSides(fmt.Sprintf( + "INSERT INTO %s.composite (label, tenant, payload) VALUES ('abc', 1, 7), ('ABC', 1, 7)", + testhelpers.TestSchemaName, + )) + + events := this.captureBinlogEvents(func() { + _, err := this.Ferry.SourceDB.Exec(fmt.Sprintf( + "DELETE FROM %s.composite WHERE label_hash = UNHEX(SHA2('abc', 256))", + testhelpers.TestSchemaName, + )) + this.Require().Nil(err) + }) + this.Require().Equal(1, len(events)) + this.Require().Equal(1, this.sourceCount("composite"), + "test is broken: the source DELETE should have removed exactly one of the two rows") + + this.applyToTarget(events) + + query := fmt.Sprintf( + "SELECT label, tenant, payload, HEX(label_hash) AS h FROM %s.composite ORDER BY h", + testhelpers.TestSchemaName, + ) + testhelpers.AssertTwoQueriesHaveEqualResult(this.T(), this.Ferry, query, query) +} + +// TestBinlogUpdateAffectsExactlyTheRowThatWasUpdated is the UPDATE +// counterpart: the same over-match, but rewriting every matched row to the +// same body makes the generated primary keys collide, so it fails loudly +// rather than silently. +func (this *GeneratedColumnsTestSuite) TestBinlogUpdateAffectsExactlyTheRowThatWasUpdated() { + this.seedDocs() + + events := this.captureBinlogEvents(func() { + _, err := this.Ferry.SourceDB.Exec(fmt.Sprintf( + "UPDATE %s.docs SET doc = 'greetings' WHERE doc_hash = UNHEX(SHA2('cafe', 256))", + testhelpers.TestSchemaName, + )) + this.Require().Nil(err) + }) + this.Require().Equal(1, len(events)) + this.Require().Equal( + []string{"greetings"}, + this.sourceStrings(fmt.Sprintf("SELECT doc FROM %s.docs WHERE doc = 'greetings'", testhelpers.TestSchemaName)), + "test is broken: the source UPDATE should have rewritten exactly one row", + ) + + this.applyToTarget(events) + + this.Require().Equal(4, this.targetDocCount()) + this.assertDocsMatch() +} + +// TestStoredPaginationKeyIsUsedInBinlogWhereClause covers the operational +// consequence of dropping the key from the WHERE clause: every replayed +// statement would scan the table instead of seeking on the primary key. +func (this *GeneratedColumnsTestSuite) TestStoredPaginationKeyIsUsedInBinlogWhereClause() { + this.seedDocs() + + table := this.loadTable("docs") + this.Require().Equal("doc_hash", table.GetPaginationColumn().Name) + this.Require().True(table.GetPaginationColumn().IsStored, "MySQL must report doc_hash as STORED") + + events := this.captureBinlogEvents(func() { + _, err := this.Ferry.SourceDB.Exec(fmt.Sprintf( + "UPDATE %s.docs SET doc = 'greetings' WHERE doc_hash = UNHEX(SHA2('cafe', 256))", + testhelpers.TestSchemaName, + )) + this.Require().Nil(err) + }) + this.Require().Equal(1, len(events)) + + stmt, err := events[0].AsSQLString(events[0].Database(), events[0].Table()) + this.Require().Nil(err) + + // An exact prefix pins the whole SET clause and the start of WHERE in one + // go; only the raw SHA-256 bytes of the key are left unpinned. + this.Require().True( + strings.HasPrefix(stmt, + "UPDATE `gftest`.`docs` SET `doc`=_binary'greetings'"+ + " WHERE `doc`=_binary'cafe' AND `doc_hash`='"), + "the pagination key must be in the WHERE clause and absent from SET; got: "+stmt, + ) +} + +// itemsDDL places one VIRTUAL and one STORED generated column between +// ordinary columns, so index-space confusion shows up as a wrong value rather +// than a coincidentally correct one. +const itemsDDL = "CREATE TABLE %s.items (" + + "id BIGINT NOT NULL AUTO_INCREMENT," + + "body VARCHAR(64) NOT NULL," + + "body_len BIGINT AS (CHAR_LENGTH(body)) VIRTUAL," + + "note VARCHAR(64)," + + "body_upper VARCHAR(64) AS (UPPER(body)) STORED," + + "PRIMARY KEY (id))" + +func (this *GeneratedColumnsTestSuite) seedItems() { + this.createOnBothSides(fmt.Sprintf(itemsDDL, testhelpers.TestSchemaName)) +} + +func (this *GeneratedColumnsTestSuite) assertItemsMatch() { + query := fmt.Sprintf( + "SELECT id, body, body_len, note, body_upper FROM %s.items ORDER BY id", + testhelpers.TestSchemaName, + ) + testhelpers.AssertTwoQueriesHaveEqualResult(this.T(), this.Ferry, query, query) +} + +// TestBinlogInsertReplayLetsTargetComputeGeneratedColumns: the INSERT must +// not name the generated columns, and need not — the target computes +// identical values from the columns we do send. +func (this *GeneratedColumnsTestSuite) TestBinlogInsertReplayLetsTargetComputeGeneratedColumns() { + this.seedItems() + + events := this.captureBinlogEvents(func() { + _, err := this.Ferry.SourceDB.Exec(fmt.Sprintf( + "INSERT INTO %s.items (id, body, note) VALUES (1, 'hello', 'first')", + testhelpers.TestSchemaName, + )) + this.Require().Nil(err) + }) + this.Require().Equal(1, len(events)) + + this.applyToTarget(events) + this.assertItemsMatch() +} + +// TestBinlogUpdateReplayMatchesOnGeneratedColumnValues: the VIRTUAL and +// STORED values read from the source's row image match what the target +// computed for itself, so predicating on them finds the row. +func (this *GeneratedColumnsTestSuite) TestBinlogUpdateReplayMatchesOnGeneratedColumnValues() { + this.seedItems() + this.execOnBothSides(fmt.Sprintf( + "INSERT INTO %s.items (id, body, note) VALUES (1, 'hello', 'first'), (2, 'world', 'second')", + testhelpers.TestSchemaName, + )) + + events := this.captureBinlogEvents(func() { + _, err := this.Ferry.SourceDB.Exec(fmt.Sprintf( + "UPDATE %s.items SET body = 'goodbye', note = 'edited' WHERE id = 1", + testhelpers.TestSchemaName, + )) + this.Require().Nil(err) + }) + this.Require().Equal(1, len(events)) + + stmt, err := events[0].AsSQLString(events[0].Database(), events[0].Table()) + this.Require().Nil(err) + this.Require().Contains(stmt, "`body_len`=", "VIRTUAL value from the row image belongs in the WHERE clause") + this.Require().Contains(stmt, "`body_upper`=", "STORED value from the row image belongs in the WHERE clause") + + this.applyToTarget(events) + this.assertItemsMatch() +} + +// TestBinlogDeleteReplayMatchesOnGeneratedColumnValues is the DELETE +// counterpart: the WHERE clause includes both generated columns and must +// still remove the row on the target. +func (this *GeneratedColumnsTestSuite) TestBinlogDeleteReplayMatchesOnGeneratedColumnValues() { + this.seedItems() + this.execOnBothSides(fmt.Sprintf( + "INSERT INTO %s.items (id, body, note) VALUES (1, 'hello', 'first'), (2, 'world', 'second')", + testhelpers.TestSchemaName, + )) + + events := this.captureBinlogEvents(func() { + _, err := this.Ferry.SourceDB.Exec(fmt.Sprintf("DELETE FROM %s.items WHERE id = 1", testhelpers.TestSchemaName)) + this.Require().Nil(err) + }) + this.Require().Equal(1, len(events)) + this.Require().Equal(1, this.sourceCount("items"), "test is broken: the source DELETE removed the wrong number of rows") + + this.applyToTarget(events) + this.assertItemsMatch() + + this.Require().Equal( + []string{"world"}, + this.targetStrings(fmt.Sprintf("SELECT body FROM %s.items ORDER BY id", testhelpers.TestSchemaName)), + ) +} + +// TestMySQLRejectsAssignmentToGeneratedColumns pins the premise the INSERT +// and SET filtering rests on: error 3105, for VIRTUAL and STORED alike. +func (this *GeneratedColumnsTestSuite) TestMySQLRejectsAssignmentToGeneratedColumns() { + this.seedItems() + + _, err := this.Ferry.TargetDB.Exec(fmt.Sprintf( + "INSERT INTO %s.items (id, body, body_len) VALUES (1, 'hello', 5)", + testhelpers.TestSchemaName, + )) + this.Require().NotNil(err) + this.Require().Contains(err.Error(), "3105", "assigning a VIRTUAL generated column must be rejected") + + _, err = this.Ferry.TargetDB.Exec(fmt.Sprintf( + "INSERT INTO %s.items (id, body, body_upper) VALUES (1, 'hello', 'HELLO')", + testhelpers.TestSchemaName, + )) + this.Require().NotNil(err) + this.Require().Contains(err.Error(), "3105", "assigning a STORED generated column must be rejected") +} + +// TestUnsignedGeneratedColumnsAreNormalisedBeforeUse: go-mysql hands back an +// unsigned generated column as a negative signed integer, like any other +// unsigned column. Skipping generated columns during normalisation — to +// match the INSERT and SET filtering — would emit `WHERE v = -1` for a column +// holding 18446744073709551615, and every replayed UPDATE and DELETE for the +// table would quietly do nothing. +func (this *GeneratedColumnsTestSuite) TestUnsignedGeneratedColumnsAreNormalisedBeforeUse() { + this.createOnBothSides(fmt.Sprintf( + "CREATE TABLE %s.bignum ("+ + "id BIGINT NOT NULL,"+ + "base BIGINT UNSIGNED NOT NULL,"+ + "v_copy BIGINT UNSIGNED AS (base) VIRTUAL,"+ + "s_copy BIGINT UNSIGNED AS (base) STORED,"+ + "PRIMARY KEY (id))", + testhelpers.TestSchemaName, + )) + // Above math.MaxInt64, so a signed reading of the binlog value is negative. + this.execOnBothSides(fmt.Sprintf( + "INSERT INTO %s.bignum (id, base) VALUES (1, 18446744073709551615)", + testhelpers.TestSchemaName, + )) + + events := this.captureBinlogEvents(func() { + _, err := this.Ferry.SourceDB.Exec(fmt.Sprintf("DELETE FROM %s.bignum WHERE id = 1", testhelpers.TestSchemaName)) + this.Require().Nil(err) + }) + this.Require().Equal(1, len(events)) + + stmt, err := events[0].AsSQLString(events[0].Database(), events[0].Table()) + this.Require().Nil(err) + + this.Require().Equal( + "DELETE FROM `gftest`.`bignum` WHERE `id`=1 AND `base`=18446744073709551615"+ + " AND `v_copy`=18446744073709551615 AND `s_copy`=18446744073709551615", + stmt, + ) + + // And it has to actually match on the target, not merely look right. + this.applyToTarget(events) + this.Require().Empty(this.targetStrings(fmt.Sprintf("SELECT id FROM %s.bignum", testhelpers.TestSchemaName))) +} + +// copyEverythingToTarget runs Ghostferry's real copy path over every loaded +// table, with the inline verifier enforcing. Generated columns are part of +// the row fingerprint, so this checks the values the target computed, not +// just the columns we sent. +func (this *GeneratedColumnsTestSuite) copyEverythingToTarget() { + this.Ferry.Tables = this.loadTables() + err := this.Ferry.RunStandaloneDataCopy(this.Ferry.Tables.AsSlice()) + this.Require().Nil(err) +} + +// TestCopyTableWithStoredGeneratedPrimaryKey covers the copy path for the +// docs table. The copy path and the binlog path filter generated columns +// independently — one from query-result order, one from schema order — so +// passing on one says nothing about the other. +func (this *GeneratedColumnsTestSuite) TestCopyTableWithStoredGeneratedPrimaryKey() { + this.createOnBothSides(fmt.Sprintf(docsDDL, testhelpers.TestSchemaName)) + _, err := this.Ferry.SourceDB.Exec(fmt.Sprintf( + "INSERT INTO %s.docs (doc) VALUES ('cafe'), ('caf\u00e9'), ('CAFE'), ('cafe ')", + testhelpers.TestSchemaName, + )) + this.Require().Nil(err) + + this.copyEverythingToTarget() + + this.Require().Equal(4, this.targetDocCount()) + this.assertDocsMatch() +} + +// TestCopyTableWithGeneratedColumnsBetweenOrdinaryOnes: with the generated +// columns mid-schema, dropping the wrong position shifts every later value by +// one instead of coincidentally landing correctly. +func (this *GeneratedColumnsTestSuite) TestCopyTableWithGeneratedColumnsBetweenOrdinaryOnes() { + this.seedItems() + _, err := this.Ferry.SourceDB.Exec(fmt.Sprintf( + "INSERT INTO %s.items (id, body, note) VALUES (1, 'hello', 'first'), (2, 'world', 'second'), (3, 'third', NULL)", + testhelpers.TestSchemaName, + )) + this.Require().Nil(err) + + this.copyEverythingToTarget() + this.assertItemsMatch() +} + +func TestGeneratedColumnsTestSuite(t *testing.T) { + suite.Run(t, &GeneratedColumnsTestSuite{GhostferryUnitTestSuite: &testhelpers.GhostferryUnitTestSuite{}}) +} diff --git a/test/go/row_batch_test.go b/test/go/row_batch_test.go index 9513f820..7cf89989 100644 --- a/test/go/row_batch_test.go +++ b/test/go/row_batch_test.go @@ -73,20 +73,11 @@ func (this *RowBatchTestSuite) TestRowBatchGeneratesInsertQuery() { this.Require().Equal(expected, v1) } -// TestRowBatchReorderedColumnsGeneratesCorrectInsert is a regression test for -// the gh-285 corruption pattern. -// -// The sharding copy filter executes: -// -// SELECT * FROM t JOIN (SELECT id …) AS batch USING(id) -// -// MySQL's USING clause moves the join column to the front of the result set, -// so for a table with schema order (tenant_id, col1, id, d) the query returns -// columns in result order (id, tenant_id, col1, d). -// -// Before the fix, AsSQLQuery used table.NonGeneratedColumnNames() (schema -// order) for the INSERT column list while values were in result order — every -// row written to the target had its column values shifted to the wrong columns. +// TestRowBatchReorderedColumnsGeneratesCorrectInsert pins the INSERT column +// list to query-result order, never schema order. The sharding copy filter's +// JOIN ... USING moves the join column to the front of the result set, so +// schema-ordered names against result-ordered values would silently write +// every value into the wrong column (the gh-285 corruption pattern). func (this *RowBatchTestSuite) TestRowBatchReorderedColumnsGeneratesCorrectInsert() { // Schema order: tenant_id(0), col1(1), id(2), d(3) schemaColumns := []schema.TableColumn{ @@ -120,6 +111,20 @@ func (this *RowBatchTestSuite) TestRowBatchReorderedColumnsGeneratesCorrectInser this.Require().Equal([]interface{}{int64(2), int64(1), "z", "2021-01-01"}, args) } +func (this *RowBatchTestSuite) TestRowBatchSelectedColumnSubsetGeneratesCorrectInsert() { + selectedColumns := []string{"col1", "col3"} + rows := []ghostferry.RowData{{int64(1000), true}} + batch := ghostferry.NewRowBatchWithColumns(this.sourceTable, rows, selectedColumns, 0) + + query, args, err := batch.AsSQLQuery(this.targetTable.Schema, this.targetTable.Name) + this.Require().Nil(err) + this.Require().Equal( + "INSERT IGNORE INTO `target_schema`.`target_table` (`col1`,`col3`) VALUES (?,?)", + query, + ) + this.Require().Equal([]interface{}{int64(1000), true}, args) +} + // TestRowBatchReorderedColumnsWithGeneratedColumnFiltersCorrectly combines the // gh-285 reordering scenario with generated column filtering: the USING join // moves 'id' first, and a VIRTUAL column 'gen' must be excluded from the @@ -157,6 +162,37 @@ func (this *RowBatchTestSuite) TestRowBatchReorderedColumnsWithGeneratedColumnFi this.Require().Equal([]interface{}{int64(2), int64(1), "z"}, args) } +// TestRowBatchWithOnlyGeneratedColumnsReturnsError covers the case LoadTables +// cannot: a CopyFilter narrowing ColumnsToSelect, or an embedder populating +// Ferry.Tables directly, can reach AsSQLQuery with nothing writable. That +// must be an error, not a panic inside strings.Repeat. +func (this *RowBatchTestSuite) TestRowBatchWithOnlyGeneratedColumnsReturnsError() { + table := &ghostferry.TableSchema{ + Table: &schema.Table{ + Schema: "test_schema", + Name: "test_table", + Columns: []schema.TableColumn{ + {Name: "base"}, + {Name: "gen_a", IsStored: true}, + {Name: "gen_b", IsVirtual: true}, + }, + }, + } + + batch := ghostferry.NewRowBatchWithColumns( + table, + []ghostferry.RowData{{int64(1), int64(2)}}, + []string{"gen_a", "gen_b"}, + 0, + ) + + q, args, err := batch.AsSQLQuery("test_schema", "test_table") + this.Require().NotNil(err, "must report an error rather than panicking in strings.Repeat") + this.Require().Contains(err.Error(), "has no columns to write") + this.Require().Equal("", q) + this.Require().Nil(args) +} + func (this *RowBatchTestSuite) TestRowBatchWithWrongColumnsReturnsError() { vals := []ghostferry.RowData{ ghostferry.RowData{1000, []byte("val0"), true}, @@ -167,7 +203,7 @@ func (this *RowBatchTestSuite) TestRowBatchWithWrongColumnsReturnsError() { _, _, err := batch.AsSQLQuery(this.targetTable.Schema, this.targetTable.Name) this.Require().NotNil(err) - this.Require().Contains(err.Error(), "test_table has 3 columns but event has 1 column") + this.Require().Contains(err.Error(), "test_table has 3 selected columns but row has 1 value") } func (this *RowBatchTestSuite) TestRowBatchMetadata() { diff --git a/test/go/table_schema_cache_test.go b/test/go/table_schema_cache_test.go index a71a15e3..8c8b2e09 100644 --- a/test/go/table_schema_cache_test.go +++ b/test/go/table_schema_cache_test.go @@ -200,14 +200,10 @@ func (this *TableSchemaCacheTestSuite) TestLoadTablesRejectTablesWhenCascadingPa this.Require().EqualError(err, ghostferry.NonExistingPaginationKeyColumnError(testhelpers.TestSchemaName, table, paginationColumn).Error()) } -// TestLoadTablesRejectVirtualPaginationKey verifies that using a VIRTUAL -// generated column as the pagination key is rejected at load time. VIRTUAL -// columns are not stored on disk, so they cannot be used reliably for cursor -// iteration. STORED generated columns are fine and are not tested here. -// -// MySQL prevents VIRTUAL columns from being a PRIMARY KEY, so the only way to -// reach this code path in practice is via CascadingPaginationColumnConfig. -func (this *TableSchemaCacheTestSuite) TestLoadTablesRejectVirtualPaginationKey() { +// A VIRTUAL pagination key must be NOT NULL and have its own visible UNIQUE +// index. Without both, the configured column does not provide the uniqueness +// and index-backed ordering pagination requires. +func (this *TableSchemaCacheTestSuite) TestLoadTablesRejectNonUniqueVirtualPaginationKey() { table := "virtual_pagination_key" virtualColumn := "vlen" cascadingPaginationColumnConfig := &ghostferry.CascadingPaginationColumnConfig{ @@ -217,8 +213,8 @@ func (this *TableSchemaCacheTestSuite) TestLoadTablesRejectVirtualPaginationKey( } query := fmt.Sprintf( - "CREATE TABLE %s.%s (id bigint(20) NOT NULL AUTO_INCREMENT, data TEXT, %s BIGINT AS (LENGTH(data)) VIRTUAL, PRIMARY KEY(id))", - testhelpers.TestSchemaName, table, virtualColumn, + "CREATE TABLE %s.%s (id bigint(20) NOT NULL AUTO_INCREMENT, data TEXT, %s BIGINT AS (LENGTH(data)) VIRTUAL NOT NULL, PRIMARY KEY(id), KEY (%s))", + testhelpers.TestSchemaName, table, virtualColumn, virtualColumn, ) _, err := this.Ferry.SourceDB.Exec(query) this.Require().Nil(err) @@ -229,6 +225,87 @@ func (this *TableSchemaCacheTestSuite) TestLoadTablesRejectVirtualPaginationKey( this.Require().EqualError(err, ghostferry.VirtualPaginationKeyError(testhelpers.TestSchemaName, table, virtualColumn).Error()) } +func (this *TableSchemaCacheTestSuite) TestLoadTablesAllowsUniqueVirtualPaginationKey() { + table := "unique_virtual_pagination_key" + virtualColumn := "virtual_id" + cascadingPaginationColumnConfig := &ghostferry.CascadingPaginationColumnConfig{ + PerTable: map[string]map[string]string{ + testhelpers.TestSchemaName: {table: virtualColumn}, + }, + } + + query := fmt.Sprintf( + "CREATE TABLE %s.%s (id BIGINT NOT NULL, data TEXT, %s BIGINT AS (id + 1) VIRTUAL NOT NULL, PRIMARY KEY(id), UNIQUE KEY (%s))", + testhelpers.TestSchemaName, table, virtualColumn, virtualColumn, + ) + _, err := this.Ferry.SourceDB.Exec(query) + this.Require().Nil(err) + + this.assertLoadTablesWithCascadingPaginationColumnConfig(table, virtualColumn, cascadingPaginationColumnConfig) +} + +func (this *TableSchemaCacheTestSuite) TestLoadTablesRejectsVirtualPaginationKeyWithCompositeUniqueIndex() { + table := "composite_unique_virtual_pagination_key" + virtualColumn := "vlen" + cascadingPaginationColumnConfig := &ghostferry.CascadingPaginationColumnConfig{ + PerTable: map[string]map[string]string{ + testhelpers.TestSchemaName: {table: virtualColumn}, + }, + } + + query := fmt.Sprintf( + "CREATE TABLE %s.%s (id BIGINT NOT NULL, data TEXT, %s BIGINT AS (LENGTH(data)) VIRTUAL NOT NULL, PRIMARY KEY(id), UNIQUE KEY (%s, id))", + testhelpers.TestSchemaName, table, virtualColumn, virtualColumn, + ) + _, err := this.Ferry.SourceDB.Exec(query) + this.Require().Nil(err) + + _, err = ghostferry.LoadTables(this.Ferry.SourceDB, this.tableFilter, nil, nil, nil, cascadingPaginationColumnConfig) + this.Require().EqualError(err, ghostferry.VirtualPaginationKeyError(testhelpers.TestSchemaName, table, virtualColumn).Error()) +} + +func (this *TableSchemaCacheTestSuite) TestLoadTablesRejectsNullableUniqueVirtualPaginationKey() { + table := "nullable_unique_virtual_pagination_key" + virtualColumn := "virtual_id" + cascadingPaginationColumnConfig := &ghostferry.CascadingPaginationColumnConfig{ + PerTable: map[string]map[string]string{ + testhelpers.TestSchemaName: {table: virtualColumn}, + }, + } + + query := fmt.Sprintf( + "CREATE TABLE %s.%s (id BIGINT NOT NULL, data TEXT, %s BIGINT AS (NULLIF(id, 0)) VIRTUAL, PRIMARY KEY(id), UNIQUE KEY (%s))", + testhelpers.TestSchemaName, table, virtualColumn, virtualColumn, + ) + _, err := this.Ferry.SourceDB.Exec(query) + this.Require().Nil(err) + + _, err = ghostferry.LoadTables(this.Ferry.SourceDB, this.tableFilter, nil, nil, nil, cascadingPaginationColumnConfig) + this.Require().EqualError(err, ghostferry.VirtualPaginationKeyError(testhelpers.TestSchemaName, table, virtualColumn).Error()) +} + +// TestLoadTablesRejectTableWithOnlyGeneratedColumns: MySQL accepts a table +// whose every column is generated — the first two assertions prove it — but +// Ghostferry would have nothing to write. Refusing it at load time turns a +// mid-move failure into an actionable startup error. +func (this *TableSchemaCacheTestSuite) TestLoadTablesRejectTableWithOnlyGeneratedColumns() { + table := "all_generated" + + _, err := this.Ferry.SourceDB.Exec(fmt.Sprintf( + "CREATE TABLE %s.%s (a BIGINT AS (1) STORED, b BIGINT AS (2) STORED, PRIMARY KEY (a))", + testhelpers.TestSchemaName, table, + )) + this.Require().Nil(err, "MySQL is expected to accept a table whose columns are all generated") + + _, err = this.Ferry.SourceDB.Exec(fmt.Sprintf("INSERT INTO %s.%s () VALUES ()", testhelpers.TestSchemaName, table)) + this.Require().Nil(err, "and to accept rows into it") + + _, err = ghostferry.LoadTables(this.Ferry.SourceDB, this.tableFilter, nil, nil, nil, nil) + + this.Require().NotNil(err) + this.Require().EqualError(err, ghostferry.NoNonGeneratedColumnsError(testhelpers.TestSchemaName, table).Error()) +} + func (this *TableSchemaCacheTestSuite) TestLoadTablesWithPaginationKeyColumnFallback() { table := "pk_fallback_column_present" query := fmt.Sprintf("CREATE TABLE %s.%s (identity bigint(20) not null, data TEXT, primary key(identity))", testhelpers.TestSchemaName, table) diff --git a/test/integration/generated_columns_test.rb b/test/integration/generated_columns_test.rb new file mode 100644 index 00000000..c554a5d8 --- /dev/null +++ b/test/integration/generated_columns_test.rb @@ -0,0 +1,242 @@ +require "test_helper" + +# End-to-end coverage for tables whose row identity is carried by a generated +# column -- the shape where replaying a binlog event can affect more rows on +# the target than it did on the source. The common case, generated columns +# alongside an ordinary primary key, is exercised by the whole suite through +# the default table (see DbHelper#seed_random_data). +class GeneratedColumnsTest < GhostferryTestCase + # A content-addressed table: the primary key is a STORED generated column + # derived from the only other column (from the PR #437 review). + CONTENT_ADDRESSED_TABLE = "test_content_addressed" + CONTENT_ADDRESSED_FULL_TABLE_NAME = DbHelper.full_table_name(DbHelper::DEFAULT_DB, CONTENT_ADDRESSED_TABLE) + + # Same, plus a payload column carrying the *same* value in every sibling row. + CONTENT_ADDRESSED_PAYLOAD_TABLE = "test_content_addressed_payload" + CONTENT_ADDRESSED_PAYLOAD_FULL_TABLE_NAME = DbHelper.full_table_name(DbHelper::DEFAULT_DB, CONTENT_ADDRESSED_PAYLOAD_TABLE) + + # A shape much closer to a real merchant table: ordinary columns, a natural + # key, and a content hash that happens to be the primary key. + TENANT_LABELS_TABLE = "test_tenant_labels" + TENANT_LABELS_FULL_TABLE_NAME = DbHelper.full_table_name(DbHelper::DEFAULT_DB, TENANT_LABELS_TABLE) + + # Four values DISTINCT as bytes -- four hashes, four rows -- but EQUAL under + # utf8mb4_unicode_ci, which is case-insensitive, accent-insensitive and PAD + # SPACE. `WHERE doc = 'cafe'` is collation equality, so it selects all four; + # that is the whole point of the fixture. + SIBLING_DOCUMENTS = ["cafe", "café", "CAFE", "cafe "].freeze + + # The one we delete or update on the source in each test. + CHOSEN_DOCUMENT = "cafe" + + # A control outside the equivalence class: without it, wiping the whole + # table would look the same as over-matching within the class. + UNRELATED_DOCUMENT = "tea" + + # If a replayed WHERE clause omits the generated columns, the remaining + # predicate is only as selective as the column collation allows. Three + # things keep these tests from passing by accident: + # + # * The sibling rows differ on no non-generated column. A fixture with an + # id, a timestamp or a differing payload would pick out one row even + # against the broken code, and prove nothing. + # + # * Each test first asserts that the SOURCE changed exactly one row, so a + # multi-row source statement cannot make both sides equally wrong. + # + # * Nothing is asserted inside an on_status handler. Minitest::Assertion + # descends from Exception, and the callback server only rescues + # StandardError, so a failing assertion in a handler is swallowed. + + # DELETE is the silent case: over-matching removes rows from the target that + # still exist on the source, nothing errors, and Ghostferry reports success. + def test_binlog_delete_must_not_over_match_rows_equal_under_the_column_collation + seed_content_addressed_table + + ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY) + + docs_on_target_before_delete = nil + ghostferry.on_status(Ghostferry::Status::ROW_COPY_COMPLETED) do + docs_on_target_before_delete = content_addressed_docs(target_db) + + # Delete exactly ONE row, addressed by its primary key; this can only + # reach the target through the binlog. + source_db.query( + "DELETE FROM #{CONTENT_ADDRESSED_FULL_TABLE_NAME} " \ + "WHERE doc_hash = UNHEX(SHA2('#{CHOSEN_DOCUMENT}', 256))" + ) + end + + ghostferry.run + assert_nil ghostferry.error + + expected = (SIBLING_DOCUMENTS - [CHOSEN_DOCUMENT]).sort + + assert_equal SIBLING_DOCUMENTS.sort, docs_on_target_before_delete, + "test is broken: the row copy should have put every row on the target " \ + "before the DELETE was issued" + + assert_equal expected, content_addressed_docs(source_db), + "test is broken: the source DELETE should have removed exactly one row" + + target_docs = content_addressed_docs(target_db) + assert_equal expected, target_docs, + "binlog DELETE over-matched on the target: one row was deleted on the " \ + "source, but the target went from #{SIBLING_DOCUMENTS.length} rows to " \ + "#{target_docs.length}" + end + + # UPDATE over-matches for the same reason, but loudly: a replayed UPDATE + # assigns every non-generated column its after-image value, so over-matched + # rows are rewritten to the same content, their generated keys collide, and + # Ghostferry aborts on the duplicate-entry error. + def test_binlog_update_must_not_over_match_rows_equal_under_the_column_collation + seed_content_addressed_payload_table + + ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY) + + rows_on_target_before_update = nil + ghostferry.on_status(Ghostferry::Status::ROW_COPY_COMPLETED) do + rows_on_target_before_update = content_addressed_payload_rows(target_db) + + source_db.query( + "UPDATE #{CONTENT_ADDRESSED_PAYLOAD_FULL_TABLE_NAME} SET payload = 'updated' " \ + "WHERE doc_hash = UNHEX(SHA2('#{CHOSEN_DOCUMENT}', 256))" + ) + end + + begin + ghostferry.run + rescue Ghostferry::ExitError + flunk "ghostferry aborted while replaying the UPDATE, which means the " \ + "replayed statement matched more rows than the one the source " \ + "updated: #{ghostferry.error && ghostferry.error["ErrMessage"]}" + end + + assert_nil ghostferry.error + + expected = SIBLING_DOCUMENTS.to_h { |doc| [doc, doc == CHOSEN_DOCUMENT ? "updated" : "original"] } + + assert_equal SIBLING_DOCUMENTS.to_h { |doc| [doc, "original"] }, rows_on_target_before_update, + "test is broken: the row copy should have put every row on the target " \ + "before the UPDATE was issued" + + assert_equal expected, content_addressed_payload_rows(source_db), + "test is broken: the source UPDATE should have changed exactly one row" + + assert_equal expected, content_addressed_payload_rows(target_db), + "binlog UPDATE over-matched on the target" + end + + # Same failure in an ordinary-looking table: tenant, label, payload, and a + # content hash as primary key. The ordinary columns are in the WHERE clause + # and still do not save it, because once the generated column is removed + # nothing that remains is unique. + def test_binlog_delete_over_matches_even_when_ordinary_columns_are_present + seed_tenant_labels_table + + ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY) + + ghostferry.on_status(Ghostferry::Status::ROW_COPY_COMPLETED) do + source_db.query( + "DELETE FROM #{TENANT_LABELS_FULL_TABLE_NAME} " \ + "WHERE label_hash = UNHEX(SHA2('#{CHOSEN_DOCUMENT}', 256))" + ) + end + + ghostferry.run + assert_nil ghostferry.error + + expected = (SIBLING_DOCUMENTS - [CHOSEN_DOCUMENT] + [UNRELATED_DOCUMENT]).sort + + assert_equal expected, tenant_labels(source_db), + "test is broken: the source DELETE should have removed exactly one row" + + target_labels = tenant_labels(target_db) + + # Stated separately from the equality below so that a failure distinguishes + # "the replayed DELETE was too broad" from "it deleted everything". + assert_includes target_labels, UNRELATED_DOCUMENT, + "the replayed DELETE removed a row outside the collation equivalence class" + + assert_equal expected, target_labels, + "binlog DELETE over-matched on the target despite tenant and payload " \ + "appearing in the WHERE clause: the source lost 1 row, the target lost " \ + "#{SIBLING_DOCUMENTS.length + 1 - target_labels.length}" + end + + private + + # The collation is stated explicitly so the test cannot quietly stop testing + # anything if a server default changes. + COLLATED_TEXT = "TEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL".freeze + + def seed_content_addressed_table + create_on_both( + "CREATE TABLE IF NOT EXISTS #{CONTENT_ADDRESSED_FULL_TABLE_NAME} (" \ + "doc #{COLLATED_TEXT}, " \ + "doc_hash BINARY(32) AS (UNHEX(SHA2(doc, 256))) STORED, " \ + "PRIMARY KEY (doc_hash))" + ) + + statement = source_db.prepare("INSERT INTO #{CONTENT_ADDRESSED_FULL_TABLE_NAME} (doc) VALUES (?)") + SIBLING_DOCUMENTS.each { |doc| statement.execute(doc) } + + assert_equal SIBLING_DOCUMENTS.sort, content_addressed_docs(source_db), + "fixture is broken: the sibling documents must be stored as distinct rows" + end + + def seed_content_addressed_payload_table + create_on_both( + "CREATE TABLE IF NOT EXISTS #{CONTENT_ADDRESSED_PAYLOAD_FULL_TABLE_NAME} (" \ + "doc #{COLLATED_TEXT}, " \ + "payload VARCHAR(32) NOT NULL, " \ + "doc_hash BINARY(32) AS (UNHEX(SHA2(doc, 256))) STORED, " \ + "PRIMARY KEY (doc_hash))" + ) + + statement = source_db.prepare( + "INSERT INTO #{CONTENT_ADDRESSED_PAYLOAD_FULL_TABLE_NAME} (doc, payload) VALUES (?, 'original')" + ) + SIBLING_DOCUMENTS.each { |doc| statement.execute(doc) } + end + + def seed_tenant_labels_table + create_on_both( + "CREATE TABLE IF NOT EXISTS #{TENANT_LABELS_FULL_TABLE_NAME} (" \ + "tenant VARCHAR(32) NOT NULL, " \ + "label #{COLLATED_TEXT}, " \ + "payload VARCHAR(32) NOT NULL, " \ + "label_hash BINARY(32) AS (UNHEX(SHA2(label, 256))) STORED, " \ + "PRIMARY KEY (label_hash), " \ + "UNIQUE KEY tenant_label (tenant, label_hash))" + ) + + statement = source_db.prepare( + "INSERT INTO #{TENANT_LABELS_FULL_TABLE_NAME} (tenant, label, payload) VALUES ('acme', ?, 'same')" + ) + (SIBLING_DOCUMENTS + [UNRELATED_DOCUMENT]).each { |label| statement.execute(label) } + end + + def create_on_both(ddl) + [source_db, target_db].each do |db| + db.query("CREATE DATABASE IF NOT EXISTS #{DEFAULT_DB}") + db.query(ddl) + end + end + + # Sorted in Ruby, byte-wise: ordering in SQL by the text column would be + # ambiguous under a case-insensitive collation. + def content_addressed_docs(db) + db.query("SELECT doc FROM #{CONTENT_ADDRESSED_FULL_TABLE_NAME}").map { |row| row["doc"] }.sort + end + + def content_addressed_payload_rows(db) + db.query("SELECT doc, payload FROM #{CONTENT_ADDRESSED_PAYLOAD_FULL_TABLE_NAME}") + .to_h { |row| [row["doc"], row["payload"]] } + end + + def tenant_labels(db) + db.query("SELECT label FROM #{TENANT_LABELS_FULL_TABLE_NAME}").map { |row| row["label"] }.sort + end +end diff --git a/test/integration/inline_verifier_test.rb b/test/integration/inline_verifier_test.rb index 6a18bd61..1bad26ae 100644 --- a/test/integration/inline_verifier_test.rb +++ b/test/integration/inline_verifier_test.rb @@ -170,6 +170,15 @@ def test_catches_binlog_streamer_corruption ghostferry.run assert verification_ran + + # The exact fingerprints are pinned deliberately: they change if the set + # of columns the verifier fingerprints ever changes -- including if + # generated columns stop being included. A prefix match would not catch + # that. + expected_message = "cutover verification failed for: gftest.test_table_1 "\ + "[PKs: #{corrupting_id} (type: rows checksum difference, source: adcad6f2c7973d35980fcff146107b66, target: 383fc80f185d5ad58c9b828cbac2ea8d) ] " + + assert_equal expected_message, ghostferry.error_lines.last["msg"] end def test_target_corruption_is_ignored_if_skip_target_verification @@ -441,6 +450,10 @@ def test_positive_negative_zero assert verification_ran assert_equal ["#{DEFAULT_DB}.#{DEFAULT_TABLE}"], incorrect_tables + expected_message = "cutover verification failed for: #{DEFAULT_DB}.#{DEFAULT_TABLE} "\ + "[PKs: 1 (type: rows checksum difference, source: db0885c88b8ddb28777dceb902ee2d57, target: ef777e4d0c30b1fd01db82f5d39f209e) ] " + assert_equal expected_message, ghostferry.error_lines.last["msg"] + # Now we run the real test case. target_db.query("UPDATE #{DEFAULT_FULL_TABLE_NAME} SET data = -0.0 WHERE id = 1") @@ -519,6 +532,11 @@ def test_null_vs_null_string assert verification_ran assert_equal ["#{DEFAULT_DB}.#{DEFAULT_TABLE}"], incorrect_tables + + expected_message = "cutover verification failed for: gftest.test_table_1 " \ + "[PKs: 1 (type: rows checksum difference, source: 999119a8e3435fafe2de01fe01383b40, target: c4aa9a09fc8588badfaeeb1d7d42a9e6) ] " + + assert_equal expected_message, ghostferry.error_lines.last["msg"] end def test_null_in_different_order @@ -544,6 +562,11 @@ def test_null_in_different_order assert verification_ran assert_equal ["#{DEFAULT_DB}.#{DEFAULT_TABLE}"], incorrect_tables + + expected_message = "cutover verification failed for: gftest.test_table_1 "\ + "[PKs: 1 (type: rows checksum difference, source: 9b4ffa1cadf5b2fb5a0ca681f9f342e5, target: c18e2e5548da5b1becd1067e74d7fbc0) ] " + + assert_equal expected_message, ghostferry.error_lines.last["msg"] end ########################### diff --git a/test/integration/interrupt_resume_test.rb b/test/integration/interrupt_resume_test.rb index 0db3c6cb..6cee22b6 100644 --- a/test/integration/interrupt_resume_test.rb +++ b/test/integration/interrupt_resume_test.rb @@ -357,20 +357,27 @@ def test_interrupt_resume_between_consecutive_rows_events refute_nil dumped_state['LastStoredBinlogPositionForTargetVerifier']['Name'] refute_nil dumped_state['LastStoredBinlogPositionForTargetVerifier']['Pos'] - # Assert that the inline-verifier resumable position has advanced beyond - # the start. The InlineVerifier listener runs synchronously inside the - # BinlogStreamer event loop — before any blocking HTTP call — so its - # position is always updated before the interrupt signal is delivered. + # The inline verifier's resumable position is updated synchronously inside + # the BinlogStreamer event loop, so it has moved by the time + # AFTER_BINLOG_APPLY delivers the interrupt. Compare (file, position) as a + # pair: max_binlog_size is 4096 here, so the binlog rotates many times. # - # LastWrittenBinlogPosition is intentionally NOT checked here: it is - # advanced by the BinlogWriter goroutine asynchronously. When the - # interrupt fires immediately on the first AFTER_BINLOG_APPLY (before the - # writer has had time to flush), that position equals the initial value set - # by Start() and the comparison would be meaningless. The real correctness - # guarantee is the resume + assert_test_table_is_identical below. - if dumped_state['LastStoredBinlogPositionForInlineVerifier']['Name'] == start_binlog_status['File'] - refute_equal dumped_state['LastStoredBinlogPositionForInlineVerifier']['Pos'], start_binlog_status['Position'] - end + # LastWrittenBinlogPosition is deliberately NOT checked: the BinlogWriter + # goroutine advances it asynchronously, so asserting it had moved was a + # race. What we require of it is that resuming produces an identical + # target, asserted at the end of this test. + resumable_position = [ + dumped_state['LastStoredBinlogPositionForInlineVerifier']['Name'], + dumped_state['LastStoredBinlogPositionForInlineVerifier']['Pos'], + ] + position_before_run = [start_binlog_status['File'], start_binlog_status['Position']] + + assert_operator( + resumable_position <=> position_before_run, :>, 0, + "the inline verifier's resumable position #{resumable_position.inspect} did " \ + "not advance past #{position_before_run.inspect}, where the binlog was " \ + "before the run started", + ) ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY) # if we did not resume at a proper state, this invocation of ghostferry diff --git a/test/integration/iterative_verifier_test.rb b/test/integration/iterative_verifier_test.rb index 3046c671..a5c096e6 100644 --- a/test/integration/iterative_verifier_test.rb +++ b/test/integration/iterative_verifier_test.rb @@ -27,26 +27,33 @@ def test_iterative_verifier_succeeds_in_normal_run # that divergence in computed output between source and target is detected. # Base data (id, data) is identical on both sides; only the STORED generated # column expression differs on the target, which must trigger a failure. + # + # Two things about the shape of this test are load-bearing. The result is + # asserted after ghostferry.run returns, never inside the handler: a failing + # assertion there is swallowed, because Minitest::Assertion descends from + # Exception and the callback server only rescues StandardError. And there + # is deliberately no datawriter: with one running, the ordinary `data` + # column diverges too and the table is reported for that reason even if the + # verifier ignores generated columns entirely. def test_iterative_verifier_detects_stored_generated_column_divergence target_db.query( "ALTER TABLE #{DEFAULT_FULL_TABLE_NAME} " \ "MODIFY summary VARCHAR(32) AS (MD5(CONCAT(data, '_differs'))) STORED" ) - datawriter = new_source_datawriter ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY, config: { verifier_type: "Iterative" }) - start_datawriter_with_ghostferry(datawriter, ghostferry) - stop_datawriter_during_cutover(datawriter, ghostferry) - verification_ran = false - ghostferry.on_status(Ghostferry::Status::VERIFIED) do |*incorrect_tables| + incorrect_tables = nil + ghostferry.on_status(Ghostferry::Status::VERIFIED) do |*tables| verification_ran = true - assert_equal ["gftest.test_table_1"], incorrect_tables + incorrect_tables = tables end ghostferry.run + assert verification_ran + assert_equal ["gftest.test_table_1"], incorrect_tables end # Same but for a VIRTUAL generated column. @@ -56,20 +63,19 @@ def test_iterative_verifier_detects_virtual_generated_column_divergence "MODIFY length BIGINT(20) AS (LENGTH(data) + 1) VIRTUAL" ) - datawriter = new_source_datawriter ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY, config: { verifier_type: "Iterative" }) - start_datawriter_with_ghostferry(datawriter, ghostferry) - stop_datawriter_during_cutover(datawriter, ghostferry) - verification_ran = false - ghostferry.on_status(Ghostferry::Status::VERIFIED) do |*incorrect_tables| + incorrect_tables = nil + ghostferry.on_status(Ghostferry::Status::VERIFIED) do |*tables| verification_ran = true - assert_equal ["gftest.test_table_1"], incorrect_tables + incorrect_tables = tables end ghostferry.run + assert verification_ran + assert_equal ["gftest.test_table_1"], incorrect_tables end def test_iterative_verifier_fails_if_binlog_streamer_incorrectly_copies_data diff --git a/test/integration/types_test.rb b/test/integration/types_test.rb index d3e187ca..bc1c1226 100644 --- a/test/integration/types_test.rb +++ b/test/integration/types_test.rb @@ -409,101 +409,15 @@ def test_decimal end end - private - - def format_float_based_on_mysql_version(value) - # mysql 5.7 removes the trailing zeros when `cast...as json` is used - ENV["MYSQL_VERSION"] == "8.0" || ENV["MYSQL_VERSION"] == "8.4" ? value.to_s : value.to_i.to_s - end - - def insert_json_on_source - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_OBJ}')") - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_ARRAY}')") - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{EMPTY_JSON}')") - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES (NULL)") - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_NULL}')") - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_TRUE}')") - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_FALSE}')") - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_NUMBER}')") - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_FLOATING_POINT_WITH_NON_ZERO_FRACTIONAL_PART}')") - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_FLOATING_POINT_WITH_ZERO_FRACTIONAL_PART}')") - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_OBJ_WITH_TRAILING_ZERO}')") - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_OBJ_LAT_LONG}')") - end - - def execute_copy_data_in_fixed_size_binary_column(column_size:, inserted_data:, expected_inserted_data:, updated_data:) - # test for the BINARY columns needing 0-byte padding - # - # Also see: https://github.com/Shopify/ghostferry/pull/159#issuecomment-597769258 - [source_db, target_db].each do |db| - db.query("CREATE DATABASE IF NOT EXISTS #{DEFAULT_DB}") - db.query("CREATE TABLE IF NOT EXISTS #{DEFAULT_FULL_TABLE_NAME} (id bigint(20) not null auto_increment, data BINARY(#{column_size}), primary key(id))") - end - - source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (id, data) VALUES (1, _binary'#{inserted_data}')") - - ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY) - - row_copy_called = false - ghostferry.on_status(Ghostferry::Status::ROW_COPY_COMPLETED) do - # select row from the target and then make sure the data with 0 padding - # is present. We do this to make sure there are no races in the test - res = target_db.query("SELECT * FROM #{DEFAULT_FULL_TABLE_NAME}") - assert_equal 1, res.count - res.each do |row| - assert_equal 1, row["id"] - assert_equal expected_inserted_data, row["data"] - end - - # now that the target is guaranteed to be in the same state as the - # source, trigger an update that will cause the binlog to stream an - # entry that needs the 0-byte padding - # - # NOTE: If we use BINLOG_STREAMING_STARTED as hook instead, we race - # with getting the update into the batch-copy instead of into the - # streaming - source_db.query("UPDATE #{DEFAULT_FULL_TABLE_NAME} SET data = _binary'#{updated_data}' WHERE id = 1") - - # NOTE: We move this flag to the end of the callback to make sure that - # we don't confuse ourselves if the callback crashes before completing - row_copy_called = true - end - - ghostferry.run - - # make sure the test framework called the expected hooks above - otherwise - # the test doesn't make much sense - assert row_copy_called - assert_test_table_is_identical - - # just being paranoid here: make sure the test outcome is as expected. It - # should be, since we made sure the tables have the same checksums, but it - # helps understand what the test code does - res = target_db.query("SELECT * FROM #{DEFAULT_FULL_TABLE_NAME}") - assert_equal 1, res.count - res.each do |row| - assert_equal 1, row["id"] - assert_equal updated_data, row["data"] - end - - end - ########################### # Generated Columns # ########################### # - # Exercises the binlog DML path with VIRTUAL and STORED generated columns - # (seed_random_data creates `length VIRTUAL` and `summary STORED`). The - # initial seed is copied via the data iterator — already covered by the - # generated-column unit tests in test/go and by the divergence tests in - # inline_verifier_test.rb. The rows we INSERT/UPDATE/DELETE on the source - # *after* ROW_COPY_COMPLETED flow only through the binlog streamer, so this - # test is what validates dml_events.go end-to-end. - # - # On MySQL 8.0.23+ virtual columns are omitted from the binlog image; the - # length check in NewBinlogDMLEvents (dml_events.go) only passes if go-mysql - # pads omitted positions back to the full schema width. The 8.0 and 8.4 CI - # matrices exercise that path; 5.7 exercises the pre-omission path. + # Exercises the binlog DML path with the default table's `length` VIRTUAL + # and `summary` STORED columns: rows changed after ROW_COPY_COMPLETED reach + # the target only through the binlog streamer. These tables have an + # ordinary `id` primary key; the case where a generated column carries the + # row's identity is covered in generated_columns_test.rb. def test_binlog_insert_with_generated_columns seed_random_data(source_db, number_of_rows: 1) @@ -511,17 +425,11 @@ def test_binlog_insert_with_generated_columns ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY) - # Multi-row INSERT lands in MySQL as a single ROW event carrying all rows - # in one rowsEvent.Rows slice — this exercises the for-loop in - # NewBinlogDMLEvents (dml_events.go) and the flattenRowData / SQL-list - # construction across more than one row per event, which a single-row - # test would not catch. Generated columns must be filtered out per row, - # not just for the event as a whole. + # A multi-row INSERT arrives as a single ROW event, so generated columns + # must be filtered per row, not just per event. inserts = (1..3).map { |i| "(#{1000 + i}, 'binlog-insert-#{i}')" }.join(",") - ghostferry.on_status(Ghostferry::Status::BINLOG_STREAMING_STARTED) do - # Only base columns specified — `length` and `summary` are computed by - # MySQL on each side from the table's own expressions. + ghostferry.on_status(Ghostferry::Status::ROW_COPY_COMPLETED) do source_db.query( "INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (id, data) VALUES #{inserts}", ) @@ -538,8 +446,7 @@ def test_binlog_insert_with_generated_columns ).first refute_nil row, "binlog INSERT for id=#{1000 + i} did not propagate to target" assert_equal "binlog-insert-#{i}", row["data"] - # Generated column values must match the source's expression output, not - # whatever happened to be in the binlog image. + # Computed by the target from its own expressions. assert_equal "binlog-insert-#{i}".length, row["length"] assert_equal Digest::MD5.hexdigest("binlog-insert-#{i}"), row["summary"] end @@ -552,10 +459,9 @@ def test_binlog_update_with_generated_columns ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY) ghostferry.on_status(Ghostferry::Status::ROW_COPY_COMPLETED) do - # buildStringMapForSet must skip generated columns from the SET clause - # (MySQL rejects assignments to generated columns) and buildStringMapForWhere - # must skip them from the WHERE clause (the old image may contain stale - # or NULL values for virtual columns). + # The replayed UPDATE must leave generated columns out of SET (MySQL + # rejects the assignment) and KEEP them in WHERE. Do not "tidy up" the + # asymmetry: see buildStringMapForWhere in dml_events.go. source_db.query( "UPDATE #{DEFAULT_FULL_TABLE_NAME} SET data = 'binlog-update' WHERE id = 1", ) @@ -600,4 +506,84 @@ def test_binlog_delete_with_generated_columns ).first assert_equal 1, remaining["cnt"] end + + private + + def format_float_based_on_mysql_version(value) + # mysql 5.7 removes the trailing zeros when `cast...as json` is used + ENV["MYSQL_VERSION"] == "8.0" || ENV["MYSQL_VERSION"] == "8.4" ? value.to_s : value.to_i.to_s + end + + def insert_json_on_source + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_OBJ}')") + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_ARRAY}')") + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{EMPTY_JSON}')") + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES (NULL)") + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_NULL}')") + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_TRUE}')") + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_FALSE}')") + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_NUMBER}')") + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_FLOATING_POINT_WITH_NON_ZERO_FRACTIONAL_PART}')") + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_FLOATING_POINT_WITH_ZERO_FRACTIONAL_PART}')") + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_OBJ_WITH_TRAILING_ZERO}')") + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (data) VALUES ('#{JSON_OBJ_LAT_LONG}')") + end + + def execute_copy_data_in_fixed_size_binary_column(column_size:, inserted_data:, expected_inserted_data:, updated_data:) + # test for the BINARY columns needing 0-byte padding + # + # Also see: https://github.com/Shopify/ghostferry/pull/159#issuecomment-597769258 + [source_db, target_db].each do |db| + db.query("CREATE DATABASE IF NOT EXISTS #{DEFAULT_DB}") + db.query("CREATE TABLE IF NOT EXISTS #{DEFAULT_FULL_TABLE_NAME} (id bigint(20) not null auto_increment, data BINARY(#{column_size}), primary key(id))") + end + + source_db.query("INSERT INTO #{DEFAULT_FULL_TABLE_NAME} (id, data) VALUES (1, _binary'#{inserted_data}')") + + ghostferry = new_ghostferry(MINIMAL_GHOSTFERRY) + + row_copy_called = false + ghostferry.on_status(Ghostferry::Status::ROW_COPY_COMPLETED) do + # select row from the target and then make sure the data with 0 padding + # is present. We do this to make sure there are no races in the test + res = target_db.query("SELECT * FROM #{DEFAULT_FULL_TABLE_NAME}") + assert_equal 1, res.count + res.each do |row| + assert_equal 1, row["id"] + assert_equal expected_inserted_data, row["data"] + end + + # now that the target is guaranteed to be in the same state as the + # source, trigger an update that will cause the binlog to stream an + # entry that needs the 0-byte padding + # + # NOTE: If we use BINLOG_STREAMING_STARTED as hook instead, we race + # with getting the update into the batch-copy instead of into the + # streaming + source_db.query("UPDATE #{DEFAULT_FULL_TABLE_NAME} SET data = _binary'#{updated_data}' WHERE id = 1") + + # NOTE: We move this flag to the end of the callback to make sure that + # we don't confuse ourselves if the callback crashes before completing + row_copy_called = true + end + + ghostferry.run + + # make sure the test framework called the expected hooks above - otherwise + # the test doesn't make much sense + assert row_copy_called + assert_test_table_is_identical + + # just being paranoid here: make sure the test outcome is as expected. It + # should be, since we made sure the tables have the same checksums, but it + # helps understand what the test code does + res = target_db.query("SELECT * FROM #{DEFAULT_FULL_TABLE_NAME}") + assert_equal 1, res.count + res.each do |row| + assert_equal 1, row["id"] + assert_equal updated_data, row["data"] + end + + end + end