Return read errors with VStream schema guidance

Makefile

@@ -45,11 +45,11 @@ test-ci: proto
 
 .PHONY: test-integration
 test-integration: proto
-	@go test -tags=integration -run TestIntegration -count=1 -v ./cmd/internal/server/handlers
+	@go test -tags=integration -run TestIntegration -count=1 -timeout 20m -v ./cmd/internal/server/handlers
 
 .PHONY: test-integration-stress
 test-integration-stress: proto
-	@INTEGRATION_STRESS=1 go test -tags=integration -run TestIntegration -count=1 -v ./cmd/internal/server/handlers
+	@INTEGRATION_STRESS=1 go test -tags=integration -run TestIntegration -count=1 -timeout 20m -v ./cmd/internal/server/handlers
 
 .PHONY: build
 build: proto/connector_sdk.proto

cmd/internal/server/handlers/integration_test.go

@@ -18,6 +18,8 @@ import (
 	psdbconnect "github.com/planetscale/airbyte-source/proto/psdbconnect/v1alpha1"
 	fivetransdk "github.com/planetscale/fivetran-source/fivetran_sdk.v2"
 	"github.com/planetscale/fivetran-source/lib"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
 )
 
@@ -299,6 +301,63 @@ func TestIntegrationRepeatedSyncBursts(t *testing.T) {
 	}
 }
 
+func TestIntegrationMidTableDDLRequiresHistoricalResync(t *testing.T) {
+	psc := loadIntegrationSource(t)
+	ctx := context.Background()
+	tableName := integrationTableName(t)
+
+	db := openIntegrationSQL(t, psc)
+	t.Cleanup(func() {
+		if _, err := db.ExecContext(context.Background(), "drop table if exists "+quoteIdent(tableName)); err != nil {
+			t.Logf("drop integration table %s: %v", tableName, err)
+		}
+		_ = db.Close()
+	})
+
+	mustExec(t, ctx, db, fmt.Sprintf(`
+		create table %s (
+			id bigint primary key,
+			before_col varchar(64) not null,
+			after_col varchar(64) not null
+		)`, quoteIdent(tableName)))
+
+	columns := []string{"id", "before_col", "after_col"}
+	_, state := runIntegrationSync(t, psc, tableName, columns, nil)
+
+	mustExec(t, ctx, db, fmt.Sprintf(
+		"insert into %s (id, before_col, after_col) values (1, 'before-old', 'after-old')",
+		quoteIdent(tableName),
+	))
+	mustExec(t, ctx, db, fmt.Sprintf(
+		"alter table %s add column middle_col varchar(64) null after before_col",
+		quoteIdent(tableName),
+	))
+	mustExec(t, ctx, db, fmt.Sprintf(
+		"insert into %s (id, before_col, middle_col, after_col) values (2, 'before-new', 'middle-new', 'after-new')",
+		quoteIdent(tableName),
+	))
+
+	sender, failedState, err := runIntegrationSyncWithError(t, psc, tableName, columns, state)
+	if err == nil {
+		t.Fatalf("expected mid-table DDL to require a historical re-sync")
+	}
+	if got := status.Code(err); got != codes.FailedPrecondition {
+		t.Fatalf("expected FailedPrecondition, got %s: %v", got, err)
+	}
+	for _, want := range []string{"historical re-sync", "failed to build table replication plan"} {
+		if !strings.Contains(err.Error(), want) {
+			t.Fatalf("expected error to contain %q, got %v", want, err)
+		}
+	}
+	assertIntegrationRecordCount(t, sender.recordsForTable(tableName), 0)
+	if _, ok := sender.latestState(t); ok {
+		t.Fatalf("failed sync should not checkpoint state")
+	}
+	if !reflect.DeepEqual(state, failedState) {
+		t.Fatalf("failed sync changed state")
+	}
+}
+
 func TestIntegrationStressBurstLoad(t *testing.T) {
 	if !integrationStressEnabled() {
 		t.Skip("set INTEGRATION_STRESS=1 to run stress integration tests")
@@ -463,6 +522,16 @@ func openIntegrationSQL(t *testing.T, psc lib.PlanetScaleSource) *sql.DB {
 func runIntegrationSync(t *testing.T, psc lib.PlanetScaleSource, tableName string, columns []string, state *lib.SyncState) (*integrationSender, *lib.SyncState) {
 	t.Helper()
 
+	sender, state, err := runIntegrationSyncWithError(t, psc, tableName, columns, state)
+	if err != nil {
+		t.Fatalf("run sync: %v", err)
+	}
+	return sender, state
+}
+
+func runIntegrationSyncWithError(t *testing.T, psc lib.PlanetScaleSource, tableName string, columns []string, state *lib.SyncState) (*integrationSender, *lib.SyncState, error) {
+	t.Helper()
+
 	mysqlClient, err := lib.NewMySQL(&psc)
 	if err != nil {
 		t.Fatalf("create mysql client: %v", err)
@@ -504,12 +573,12 @@ func runIntegrationSync(t *testing.T, psc lib.PlanetScaleSource, tableName strin
 	logger := NewSchemaAwareSerializer(sender, "integration", psc.TreatTinyIntAsBoolean, sourceSchema.SchemaList, sourceSchema.EnumsAndSets)
 	syncer := &Sync{}
 	if err := syncer.Handle(&psc, &connectClient, logger, state, selection); err != nil {
-		t.Fatalf("run sync: %v", err)
+		return sender, state, err
 	}
 	if checkpoint, ok := sender.latestState(t); ok {
 		state = checkpoint
 	}
-	return sender, state
+	return sender, state, nil
 }
 
 func integrationSelection(schemaName, tableName string, columns []string) *fivetransdk.Selection_WithSchema {

cmd/internal/server/handlers/sync.go

@@ -80,6 +80,9 @@ func (s *Sync) Handle(psc *lib.PlanetScaleSource, db *lib.ConnectClient, logger
 				columns := includedColumns(table)
 				sc, err := (*db).Read(ctx, logger, *psc, table.TableName, columns, tc, onRow, onCursor, onUpdate)
 				if err != nil {
+					if lib.IsVStreamSchemaIncompatibilityError(err) {
+						return status.Error(codes.FailedPrecondition, err.Error())
+					}
 					return status.Error(codes.Internal, fmt.Sprintf("failed to download rows for table : %s , error : %s", table.TableName, err.Error()))
 				}
 				if sc != nil {

cmd/internal/server/handlers/sync_test.go

@@ -9,6 +9,8 @@ import (
 	psdbconnect "github.com/planetscale/airbyte-source/proto/psdbconnect/v1alpha1"
 	fivetransdk "github.com/planetscale/fivetran-source/fivetran_sdk.v2"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 func TestCallsReadWithSelectedSchema(t *testing.T) {
@@ -172,3 +174,48 @@ func TestCallsReadWithStartingGtids(t *testing.T) {
 	}, schemaSelection)
 	assert.NoError(t, err)
 }
+
+func TestVStreamSchemaIncompatibilityReturnsFailedPrecondition(t *testing.T) {
+	psc := &lib.PlanetScaleSource{}
+	tl := &testLogger{}
+	schema := fivetransdk.SchemaSelection{
+		SchemaName: "SalesDB",
+		Included:   true,
+		Tables: []*fivetransdk.TableSelection{
+			{
+				Included:  true,
+				TableName: "customers",
+			},
+		},
+	}
+	schemaSelection := &fivetransdk.Selection_WithSchema{
+		WithSchema: &fivetransdk.TablesWithSchema{
+			Schemas: []*fivetransdk.SchemaSelection{
+				&schema,
+			},
+		},
+	}
+
+	readFn := func(ctx context.Context, logger lib.DatabaseLogger, ps lib.PlanetScaleSource, tableName string, columns []string,
+		tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate,
+	) (*lib.SerializedCursor, error) {
+		return nil, status.Error(codes.Unknown, "Code: FAILED_PRECONDITION\n"+
+			"column after_col not found in table customers\n\n"+
+			"failed to build table replication plan for table customers")
+	}
+
+	state, err := psc.GetInitialState("SalesDB", []string{"-"})
+	assert.NoError(t, err)
+	db := lib.NewTestConnectClient(readFn)
+	err = (&Sync{}).Handle(psc, &db, tl, &lib.SyncState{
+		Keyspaces: map[string]lib.KeyspaceState{
+			"SalesDB": {
+				Streams: map[string]lib.ShardStates{
+					"SalesDB:customers": state,
+				},
+			},
+		},
+	}, schemaSelection)
+	assert.Error(t, err)
+	assert.Equal(t, codes.FailedPrecondition, status.Code(err))
+}

lib/connect_client.go

@@ -18,6 +18,7 @@ import (
 	clientoptions "github.com/planetscale/psdb/core/pool/options"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 
 	"vitess.io/vitess/go/sqltypes"
 
@@ -154,7 +155,7 @@ func (p connectClient) Read(ctx context.Context, logger DatabaseLogger, ps Plane
 			if s, ok := status.FromError(err); ok {
 				// if the error is anything other than server timeout, keep going
 				if s.Code() != codes.DeadlineExceeded {
-					logger.Warning(fmt.Sprintf("%vGot error [%v] with message [%q], Returning with cursor :[%v] after non-timeout error", preamble, s.Code(), err, currentPosition))
+					logger.Warning(fmt.Sprintf("%vGot error [%v] with message [%q], returning error with cursor :[%v] after non-timeout error", preamble, s.Code(), err, currentPosition))
 
 					// Check for binlog expiration error and reset cursor for historical sync
 					if IsBinlogsExpirationError(err) {
@@ -175,7 +176,11 @@ func (p connectClient) Read(ctx context.Context, logger DatabaseLogger, ps Plane
 						continue
 					}
 
-					return currentSerializedCursor, nil
+					if IsVStreamSchemaIncompatibilityError(err) {
+						return currentSerializedCursor, errors.Wrapf(err, "PlanetScale VStream cannot continue incremental replication for table %s after a schema change; run a Fivetran historical re-sync for this connection to recover", tableName)
+					}
+
+					return currentSerializedCursor, err
 				} else {
 					consecutiveTimeouts++
 					logger.Info(fmt.Sprintf("%sTimeout occurred (%d/%d consecutive timeouts)", preamble, consecutiveTimeouts, maxConsecutiveTimeouts))
@@ -271,6 +276,7 @@ func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableNam
 	if tc.LastKnownPk != nil {
 		tc.Position = ""
 	}
+	syncStartCursor := cloneTableCursor(tc)
 
 	logger.Info(fmt.Sprintf("%sSyncing with cursor position : [%v], using last known PK : %v, stop cursor is : [%v]", preamble, tc.Position, tc.LastKnownPk != nil, stopPosition))
 
@@ -308,7 +314,7 @@ func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableNam
 					}
 					sqlResult.Rows = append(sqlResult.Rows, row)
 					if err := onResult(sqlResult, OpType_Insert); err != nil {
-						return tc, status.Error(codes.Internal, "unable to serialize row")
+						return syncStartCursor, status.Error(codes.Internal, "unable to serialize row")
 					}
 				}
 			}
@@ -321,7 +327,7 @@ func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableNam
 					}
 					sqlResult.Rows = append(sqlResult.Rows, row)
 					if err := onResult(sqlResult, OpType_Delete); err != nil {
-						return nil, status.Error(codes.Internal, "unable to serialize row")
+						return syncStartCursor, status.Error(codes.Internal, "unable to serialize row")
 					}
 				}
 			}
@@ -334,7 +340,7 @@ func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableNam
 					After:  serializeQueryResult(update.After),
 				}
 				if err := onUpdate(updatedRow); err != nil {
-					return nil, status.Error(codes.Internal, "unable to serialize update")
+					return syncStartCursor, status.Error(codes.Internal, "unable to serialize update")
 				}
 			}
 		}
@@ -359,6 +365,13 @@ func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableNam
 	}
 }
 
+func cloneTableCursor(tc *psdbconnect.TableCursor) *psdbconnect.TableCursor {
+	if tc == nil {
+		return nil
+	}
+	return proto.Clone(tc).(*psdbconnect.TableCursor)
+}
+
 func (p connectClient) filterExistingColumns(ctx context.Context, ps PlanetScaleSource, tableName string, columns []string) ([]string, error) {
 	existingColumns := []string{}
 	results, err := (*p.Mysql).GetKeyspaceTableColumns(ctx, ps.Database, tableName)
@@ -457,3 +470,20 @@ func IsBinlogsExpirationError(err error) bool {
 	}
 	return strings.Contains(err.Error(), "Cannot replicate because the source purged required binary logs")
 }
+
+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, "column ") && strings.Contains(message, " not found in table "))
+}