Recognise all recoverable schema incompatibilities in vstream plan failures

lib/connect_client.go

@@ -940,19 +940,54 @@ 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 every variant with "failed to build table replication plan for
+// table <name>". The inner cause depends on the shape of the DDL:
+//
+//   - "column <c> not found in table <t>" — 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 <t>" — 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.
+//
+// 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: <n>, position: <i>" — a historical sync would hit the
+//     same unsupported column type.
+//   - "unknown table <t> 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
 	}
 
 	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, "failed to build ENUM and SET column integer to string mappings") ||
 		(strings.Contains(message, "column ") && strings.Contains(message, " not found in table "))
 }

lib/connect_client_test.go

@@ -1480,6 +1480,38 @@ 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,
+		},
+		{
+			// 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 {
@@ -1511,3 +1543,29 @@ 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"
+
+// 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"