From c9f7b1601ca7c7e35c88ab0e7ddf3b03e59fecd0 Mon Sep 17 00:00:00 2001 From: Omar Ramos Date: Mon, 3 Aug 2026 13:10:50 -0700 Subject: [PATCH 1/2] Match dropped-column schema incompatibility in vstream errors IsVStreamSchemaIncompatibilityError required "Code: FAILED_PRECONDITION" in the error text. Two of the three ways vstreamer reports a schema mismatch carry that code, but the DROP COLUMN variant does not: it is raised as a plain fmt.Errorf inside buildTableColumns and reaches the connector with no gRPC code attached. The result was that a deploy dropping a column from a replicated table fell through to the generic error return, so the operator got an opaque Internal error rather than the FailedPrecondition and the explicit "run a historical re-sync" guidance added in #89. Gate on the "failed to build table replication plan" wrapper, which all three variants carry, and add "cannot determine table columns" to the set of recognised causes. The FAILED_PRECONDITION check is dropped rather than relaxed because it excluded a real failure shape while adding no selectivity the wrapper does not already provide. Co-Authored-By: Claude Opus 5 (1M context) --- lib/connect_client.go | 25 ++++++++++++++++++++++--- lib/connect_client_test.go | 23 +++++++++++++++++++++++ 2 files changed, 45 insertions(+), 3 deletions(-) diff --git a/lib/connect_client.go b/lib/connect_client.go index 6190e12..c82bb35 100644 --- a/lib/connect_client.go +++ b/lib/connect_client.go @@ -940,19 +940,38 @@ func IsBinlogsExpirationError(err error) bool { return strings.Contains(err.Error(), "Cannot replicate because the source purged required binary logs") } +// IsVStreamSchemaIncompatibilityError reports whether err indicates that the +// source tablet could not build a replication plan for a table because the +// schema recorded in the binlog event no longer lines up with the columns this +// stream asked for. Recovery is the same in every case: drop the cursor and run +// a historical sync. +// +// vstreamer wraps all variants with "failed to build table replication plan for +// table ". The inner cause depends on the shape of the DDL: +// +// - "column not found in table " — a column was appended and the stream +// now requests it, but the event being replayed predates the ALTER. +// - "cannot use column names in vstream filter ..." — columns were inserted +// mid-table (ADD COLUMN ... AFTER) or reordered, so the truncated schema no +// longer type-matches the event and only positional names are available. +// - "cannot determine table columns for " — a column was dropped, so the +// current schema is narrower than the event. +// +// The first two are raised through vterrors as FAILED_PRECONDITION. The +// drop-column variant is a plain fmt.Errorf inside vstreamer and reaches us with +// no gRPC code attached, so the code is deliberately not part of the match: +// requiring it silently excluded every DROP COLUMN deploy. func IsVStreamSchemaIncompatibilityError(err error) bool { if err == nil { return false } message := err.Error() - if !strings.Contains(message, "Code: FAILED_PRECONDITION") { - return false - } if !strings.Contains(message, "failed to build table replication plan") { return false } return strings.Contains(message, "cannot use column names in vstream filter") || + strings.Contains(message, "cannot determine table columns") || (strings.Contains(message, "column ") && strings.Contains(message, " not found in table ")) } diff --git a/lib/connect_client_test.go b/lib/connect_client_test.go index e6a5252..9d4cba9 100644 --- a/lib/connect_client_test.go +++ b/lib/connect_client_test.go @@ -1480,6 +1480,20 @@ func TestIsVStreamSchemaIncompatibilityError(t *testing.T) { err: status.Error(codes.Unknown, "Code: FAILED_PRECONDITION\nanother replication error"), want: false, }, + { + // DROP COLUMN leaves the live schema narrower than the replayed + // event. vstreamer raises this with a plain fmt.Errorf, so unlike + // the other variants it carries no FAILED_PRECONDITION code. + name: "dropped column without a failed precondition code", + err: status.Error(codes.Unknown, vstreamDroppedColumnErrorMessage), + want: true, + }, + { + name: "plan failure with an unrelated cause is not schema incompatibility", + err: status.Error(codes.Unknown, "stream (at source tablet) error: unsupported type: 245, position: 3\n\n"+ + "failed to build table replication plan for table customers"), + want: false, + }, } for _, tt := range tests { @@ -1511,3 +1525,12 @@ const vstreamColumnNotFoundErrorMessage = "error starting stream from shard GTID "column after_col not found in table customers\n\n" + "failed to build table replication plan for table customers\n" + "failed to parse transaction payload's internal event" + +// Emitted after a DROP COLUMN, when the live schema is narrower than the event +// being replayed. Note the absence of a "Code: FAILED_PRECONDITION" annotation: +// vstreamer raises this one with a plain fmt.Errorf rather than through +// vterrors, so no gRPC code is attached. +const vstreamDroppedColumnErrorMessage = "error starting stream from shard GTID keyspace:\"fivetran\" shard:\"-\": persistent error in vstream: " + + "stream (at source tablet) error @ (including the GTID we failed to process): \n" + + "cannot determine table columns for customers: event has [8 18 18 252 5 3], schema has [id before_col after_col]\n\n" + + "failed to build table replication plan for table customers" From 408e6f4e7ef17847d1d3d1125765d361f4c4bce6 Mon Sep 17 00:00:00 2001 From: Omar Ramos Date: Mon, 3 Aug 2026 13:53:29 -0700 Subject: [PATCH 2/2] Also match dropped ENUM column "failed to build ENUM and SET column integer to string mappings" shares the "failed to build table replication plan" wrapper and is recovered the same way, but was falling through to the generic error return. It is raised when an ENUM or SET column is dropped and its value list can no longer be recovered to decode the integers in the replayed event. Like the drop-column variant it carries no gRPC code. Two other errors share the wrapper and are deliberately excluded, each with a negative test: - "unsupported type: , position: ", since a historical sync would hit the same unsupported column type. - "unknown table in schema", which is not a stale-cursor condition. A historian miss returns (nil, nil) and the schema engine falls back to the live schema rather than erroring, so this indicates an undecodable GTID or a table genuinely absent from the tablet schema. The latter can be transient during an online DDL rename swap, where resetting the cursor would force an unnecessary historical sync. Co-Authored-By: Claude Opus 5 (1M context) --- lib/connect_client.go | 26 +++++++++++++++++++++----- lib/connect_client_test.go | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 5 deletions(-) diff --git a/lib/connect_client.go b/lib/connect_client.go index c82bb35..0b666a8 100644 --- a/lib/connect_client.go +++ b/lib/connect_client.go @@ -946,7 +946,7 @@ func IsBinlogsExpirationError(err error) bool { // stream asked for. Recovery is the same in every case: drop the cursor and run // a historical sync. // -// vstreamer wraps all variants with "failed to build table replication plan for +// vstreamer wraps every variant with "failed to build table replication plan for // table ". The inner cause depends on the shape of the DDL: // // - "column not found in table " — a column was appended and the stream @@ -956,11 +956,26 @@ func IsBinlogsExpirationError(err error) bool { // longer type-matches the event and only positional names are available. // - "cannot determine table columns for " — a column was dropped, so the // current schema is narrower than the event. +// - "failed to build ENUM and SET column integer to string mappings" — an ENUM +// or SET column was dropped, so its value list can no longer be recovered to +// decode the integers in the event. // -// The first two are raised through vterrors as FAILED_PRECONDITION. The -// drop-column variant is a plain fmt.Errorf inside vstreamer and reaches us with -// no gRPC code attached, so the code is deliberately not part of the match: -// requiring it silently excluded every DROP COLUMN deploy. +// Only the first two are raised through vterrors as FAILED_PRECONDITION; the +// others are plain errors and reach us with no gRPC code attached, so the code is +// deliberately not part of the match. Requiring it silently excluded every DROP +// COLUMN deploy. +// +// The wrapper alone is not sufficient. Two other errors share it and are not +// recoverable by re-syncing: +// +// - "unsupported type: , position: " — a historical sync would hit the +// same unsupported column type. +// - "unknown table in schema" — the tablet could not resolve the table at +// all. This is not a stale-cursor condition: a historian miss falls back to +// the live schema rather than erroring, so this indicates an undecodable GTID +// or a table genuinely absent from the tablet's schema, which can be +// transient during an online DDL rename swap. Resetting the cursor for a +// transient condition would force an unnecessary historical sync. func IsVStreamSchemaIncompatibilityError(err error) bool { if err == nil { return false @@ -973,5 +988,6 @@ func IsVStreamSchemaIncompatibilityError(err error) bool { return strings.Contains(message, "cannot use column names in vstream filter") || strings.Contains(message, "cannot determine table columns") || + strings.Contains(message, "failed to build ENUM and SET column integer to string mappings") || (strings.Contains(message, "column ") && strings.Contains(message, " not found in table ")) } diff --git a/lib/connect_client_test.go b/lib/connect_client_test.go index 9d4cba9..c27b797 100644 --- a/lib/connect_client_test.go +++ b/lib/connect_client_test.go @@ -1489,11 +1489,29 @@ func TestIsVStreamSchemaIncompatibilityError(t *testing.T) { want: true, }, { + // Dropping an ENUM or SET column loses the value list needed to + // decode the integers in the replayed event. + name: "dropped enum column loses its string mappings", + err: status.Error(codes.Unknown, vstreamEnumMappingErrorMessage), + want: true, + }, + { + // Shares the wrapper but a historical sync would hit the same + // unsupported column type, so re-syncing cannot recover it. name: "plan failure with an unrelated cause is not schema incompatibility", err: status.Error(codes.Unknown, "stream (at source tablet) error: unsupported type: 245, position: 3\n\n"+ "failed to build table replication plan for table customers"), want: false, }, + { + // Also shares the wrapper, but a historian miss falls back to the + // live schema rather than erroring, so this is not a stale-cursor + // condition. It can be transient during an online DDL rename swap, + // where resetting the cursor would force a needless historical sync. + name: "unresolvable table is not schema incompatibility", + err: status.Error(codes.Unknown, vstreamUnknownTableErrorMessage), + want: false, + }, } for _, tt := range tests { @@ -1534,3 +1552,20 @@ const vstreamDroppedColumnErrorMessage = "error starting stream from shard GTID "stream (at source tablet) error @ (including the GTID we failed to process): \n" + "cannot determine table columns for customers: event has [8 18 18 252 5 3], schema has [id before_col after_col]\n\n" + "failed to build table replication plan for table customers" + +// Emitted when the tablet cannot resolve the table at all. Deliberately NOT +// treated as a schema incompatibility: a historian miss falls back to the live +// schema rather than erroring, so this does not indicate a stale cursor, and it +// can be transient during an online DDL rename swap. +const vstreamUnknownTableErrorMessage = "error starting stream from shard GTID keyspace:\"fivetran\" shard:\"-\": persistent error in vstream: " + + "stream (at source tablet) error @ (including the GTID we failed to process): \n" + + "unknown table customers in schema\n\n" + + "failed to build table replication plan for table customers" + +// Emitted after dropping an ENUM or SET column: the value list needed to decode +// the integers in the replayed event is no longer recoverable. +const vstreamEnumMappingErrorMessage = "error starting stream from shard GTID keyspace:\"fivetran\" shard:\"-\": persistent error in vstream: " + + "stream (at source tablet) error @ (including the GTID we failed to process): \n" + + "enum or set column status does not have valid string values: \n\n" + + "failed to build ENUM and SET column integer to string mappings\n\n" + + "failed to build table replication plan for table customers"