From 88d9e2eae5e00c50c1740f591f24671bc3d30639 Mon Sep 17 00:00:00 2001 From: Omar Ramos Date: Tue, 4 Aug 2026 20:04:09 -0700 Subject: [PATCH] Propagate new columns behind an opt-in, gated on the DDL event Fivetran tells us per table whether new columns should be synced (TableSelection.include_new_columns), and the SDK guide makes acting on it the connector's job. We never read the field: includedColumns() looks only at TableSelection.columns, so a column added after a connection was set up is never named in the VStream projection and its values never arrive. No error is raised, syncs report success, and only a historical re-sync recovers. Widening the projection is not something we can simply do up front, though. Naming a column that did not exist at the replay position is exactly what produces "column X not found in table Y" against a pre-DDL TABLE_MAP. So the stream starts on Fivetran's selection as-is and rebuilds the projection only once it observes the DDL event for the table. Resuming there is safe because the cursor is already past the schema change: vstreamer emits a GTID event carrying EncodePosition(vs.pos) immediately before the DDL event, vs.pos already includes the DDL's own GTID, and vtgate converts that GTID to the VGTID we consume in place. Only post-DDL row events are replayed, so their TABLE_MAP carries the new column and the plan resolves without help from vttablet's schema historian -- this needs no --track-schema-versions. The serializer needed the same treatment. columnSelection and columnWriters are both built from TableSelection.columns, so a widened projection would have been discarded on the way out, and a missing writer is a hard error rather than a silent skip. Both opt-ins must be set: the new propagate_new_columns source setting and Fivetran's per-table include_new_columns. A column Fivetran explicitly deselected is present-and-false in the map and stays excluded; only columns it never named at all are adopted. The setting defaults to false and is labelled experimental. Scope, verified against a live branch with tracking off rather than inferred: - Fixes new columns never arriving. The pre-DDL row is delivered on the narrow projection and the post-DDL row carries the new column's value. - Does not change hard stalls. When Fivetran's selection already names the new column, filterExistingColumns keeps it (it exists live), so the first window fails before reaching the DDL and the gate never fires. The error is identical with the setting on or off and is still matched by IsVStreamSchemaIncompatibilityError, so recovery stays with #96/#97. - Does not change dropped columns on a lagging cursor. The width check fires on the pre-DDL row before the DDL is reached, so that case still depends on the historian. rebuildProjection drops vanished columns for coherence, which matters only for a drop landing mid-Read. Note for anyone tempted to loosen the projection instead: falling back to SELECT * silently corrupts data across a mid-table ADD. analyzeExprs copies Table.Fields positionally for a StarExpr and never calls findColumn, so nothing catches the skew from the st.Fields[:len(tm.Types)] truncation. On a lagging cursor with ADD COLUMN newcol AFTER a over (id, a, b, c), * delivered b's value as newcol and c's value as b with no error at all. The explicit projection is a guardrail, not the defect. Co-Authored-By: Claude Opus 5 (1M context) --- .../server/handlers/configuration_form.go | 17 + .../handlers/configuration_form_test.go | 36 ++ .../server/handlers/integration_test.go | 2 +- .../handlers/schema_aware_serializer.go | 32 +- .../handlers/schema_aware_serializer_test.go | 105 +++- cmd/internal/server/handlers/sync.go | 2 +- cmd/internal/server/handlers/sync_test.go | 14 +- cmd/internal/server/server.go | 2 +- cmd/internal/server/server_test.go | 49 +- cmd/internal/server/types.go | 8 + lib/connect_client.go | 170 +++++- lib/connect_client_test.go | 494 +++++++++++++++++- lib/planetscale_source.go | 16 + lib/test_types.go | 6 +- 14 files changed, 884 insertions(+), 69 deletions(-) diff --git a/cmd/internal/server/handlers/configuration_form.go b/cmd/internal/server/handlers/configuration_form.go index 9bedaf0..6bf8b54 100644 --- a/cmd/internal/server/handlers/configuration_form.go +++ b/cmd/internal/server/handlers/configuration_form.go @@ -18,6 +18,11 @@ func (ConfigurationForm) Handle(ctx context.Context, _ *fivetransdk.Configuratio tinyIntDesc := "Enable this setting to serialize tinyint(1) as boolean values" useReplicaDesc := "Only set to true if your PlanetScale branch has a replica. PlanetScale Development branches do not have replicas." autoResyncDesc := "When a schema change leaves the saved position unreadable, automatically reset the cursor and run a historical sync for the affected table instead of failing the sync and waiting for a manual re-sync. Defaults to false; enabling it trades additional monthly active rows for unattended recovery." + propagateNewColumnsDesc := "EXPERIMENTAL - leave disabled unless you have been asked to turn it on. " + + "When enabled, a column added to a table after this connection was set up starts syncing on its own, " + + "instead of requiring a historical re-sync; a dropped column stops being requested. " + + "This only applies to tables where you have allowed new columns in the connection's schema settings. " + + "Defaults to false. If a sync behaves unexpectedly after enabling this, turn it off and re-sync." required := true resp := &fivetransdk.ConfigurationFormResponse{ Fields: []*fivetransdk.FormField{ @@ -100,6 +105,18 @@ func (ConfigurationForm) Handle(ctx context.Context, _ *fivetransdk.Configuratio }, }, }, + { + Name: "propagate_new_columns", + Label: "[Experimental] Adapt to added and dropped columns automatically?", + Description: &propagateNewColumnsDesc, + Type: &fivetransdk.FormField_DropdownField{ + DropdownField: &fivetransdk.DropdownField{ + DropdownField: []string{ + "true", "false", + }, + }, + }, + }, { Name: "starting_gtids", Label: "JSON containing keyspace, shard, and starting GTIDs", diff --git a/cmd/internal/server/handlers/configuration_form_test.go b/cmd/internal/server/handlers/configuration_form_test.go index 5ac8282..1d606ab 100644 --- a/cmd/internal/server/handlers/configuration_form_test.go +++ b/cmd/internal/server/handlers/configuration_form_test.go @@ -1 +1,37 @@ package handlers + +import ( + "context" + "strings" + "testing" + + fivetransdk "github.com/planetscale/fivetran-source/fivetran_sdk.v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// propagate_new_columns is experimental, so the form has to say so plainly and +// must not present itself as a safe default. +func TestConfigurationForm_PropagateNewColumnsIsOfferedAsExperimental(t *testing.T) { + resp, err := ConfigurationForm{}.Handle(context.Background(), &fivetransdk.ConfigurationFormRequest{}) + require.NoError(t, err) + + var field *fivetransdk.FormField + for _, f := range resp.Fields { + if f.Name == "propagate_new_columns" { + field = f + } + } + + require.NotNil(t, field, "propagate_new_columns must be offered on the setup form") + assert.Contains(t, strings.ToLower(field.Label), "experimental", "the label must flag the field as experimental") + require.NotNil(t, field.Description) + assert.Contains(t, strings.ToLower(*field.Description), "experimental") + + dropdown, ok := field.Type.(*fivetransdk.FormField_DropdownField) + require.Truef(t, ok, "expected a dropdown, got %T", field.Type) + assert.ElementsMatch(t, []string{"true", "false"}, dropdown.DropdownField.DropdownField) + + // Nothing marks it required -- an untouched form must leave it off. + assert.True(t, field.Required == nil || !*field.Required) +} diff --git a/cmd/internal/server/handlers/integration_test.go b/cmd/internal/server/handlers/integration_test.go index 8a6b596..e2bb34d 100644 --- a/cmd/internal/server/handlers/integration_test.go +++ b/cmd/internal/server/handlers/integration_test.go @@ -802,7 +802,7 @@ func runIntegrationSyncWithSender(t *testing.T, ctx context.Context, psc lib.Pla } selection := integrationSelection(psc.Database, tableName, columns) - logger := NewSchemaAwareSerializer(sender, "integration", psc.TreatTinyIntAsBoolean, sourceSchema.SchemaList, sourceSchema.EnumsAndSets) + logger := NewSchemaAwareSerializer(sender, "integration", psc.TreatTinyIntAsBoolean, sourceSchema.SchemaList, sourceSchema.EnumsAndSets, psc.PropagateNewColumns) syncer := &Sync{} if err := syncer.Handle(ctx, &psc, &connectClient, logger, state, selection); err != nil { return state, err diff --git a/cmd/internal/server/handlers/schema_aware_serializer.go b/cmd/internal/server/handlers/schema_aware_serializer.go index d6cbc57..79269ed 100644 --- a/cmd/internal/server/handlers/schema_aware_serializer.go +++ b/cmd/internal/server/handlers/schema_aware_serializer.go @@ -47,6 +47,12 @@ type schemaAwareSerializer struct { serializers map[string]*recordSerializer schemaList *fivetransdk.SchemaList enumsAndSets SchemaEnumsAndSets + // propagateNewColumns mirrors PlanetScaleSource.PropagateNewColumns. When + // set, columns that exist in the database but were never named in + // Fivetran's selection are serialized too, for tables that allow new + // columns. Without this the widened VStream projection would be discarded + // here. + propagateNewColumns bool } type recordSerializer interface { @@ -152,7 +158,7 @@ func convertRowToMap(row *sqltypes.Row, columns []string) (map[string]sqltypes.V return record, nil } -func NewSchemaAwareSerializer(sender LogSender, prefix string, serializeTinyIntAsBool bool, schemaList *fivetransdk.SchemaList, enumsAndSets SchemaEnumsAndSets) Serializer { +func NewSchemaAwareSerializer(sender LogSender, prefix string, serializeTinyIntAsBool bool, schemaList *fivetransdk.SchemaList, enumsAndSets SchemaEnumsAndSets, propagateNewColumns bool) Serializer { return &schemaAwareSerializer{ prefix: prefix, sender: sender, @@ -160,6 +166,7 @@ func NewSchemaAwareSerializer(sender LogSender, prefix string, serializeTinyIntA schemaList: schemaList, enumsAndSets: enumsAndSets, serializers: map[string]*recordSerializer{}, + propagateNewColumns: propagateNewColumns, } } @@ -300,6 +307,14 @@ func (l *schemaAwareSerializer) generateRecordSerializer(table *fivetransdk.Tabl serializers := map[string]func(value sqltypes.Value) (*fivetransdk.ValueType, error){} var err error pks := map[string]bool{} + + // Start from Fivetran's selection verbatim so behaviour is unchanged when + // the feature is off, then add any adopted columns as selected. + effectiveSelection := make(map[string]bool, len(table.Columns)) + for colName, included := range table.Columns { + effectiveSelection[colName] = included + } + for _, schema := range l.schemaList.Schemas { if schema.Name != selectedSchemaName { continue @@ -326,7 +341,18 @@ func (l *schemaAwareSerializer) generateRecordSerializer(table *fivetransdk.Tabl tableSchemaEnumAndSetValues = map[string]ValueMap{} } - for colName, included := range table.Columns { + // A column absent from table.Columns is one Fivetran has never seen; + // adopt it when the table allows new columns. An explicitly deselected + // column is present with false and stays excluded. + if l.propagateNewColumns && table.IncludeNewColumns { + for _, columnWithSchema := range tableSchema.Columns { + if _, named := table.Columns[columnWithSchema.Name]; !named { + effectiveSelection[columnWithSchema.Name] = true + } + } + } + + for colName, included := range effectiveSelection { if !included { continue } @@ -366,7 +392,7 @@ func (l *schemaAwareSerializer) generateRecordSerializer(table *fivetransdk.Tabl } return &schemaAwareRecordSerializer{ - columnSelection: table.Columns, + columnSelection: effectiveSelection, primaryKeys: pks, columnWriters: serializers, }, nil diff --git a/cmd/internal/server/handlers/schema_aware_serializer_test.go b/cmd/internal/server/handlers/schema_aware_serializer_test.go index 9e7dc29..b02afd0 100644 --- a/cmd/internal/server/handlers/schema_aware_serializer_test.go +++ b/cmd/internal/server/handlers/schema_aware_serializer_test.go @@ -22,7 +22,7 @@ func TestCanSerializeInsert(t *testing.T) { row, s, err := generateTestRecord("PhaniRaj") require.NoError(t, err) tl := &testLogSender{} - l := NewSchemaAwareSerializer(tl, "", true, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}) + l := NewSchemaAwareSerializer(tl, "", true, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}, false) schema := &fivetransdk.SchemaSelection{ Included: true, @@ -80,7 +80,7 @@ func TestRecordReturnsSenderError(t *testing.T) { row, s, err := generateTestRecord("PhaniRaj") require.NoError(t, err) tl := &testLogSender{sendError: assert.AnError} - l := NewSchemaAwareSerializer(tl, "", true, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}) + l := NewSchemaAwareSerializer(tl, "", true, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}, false) schema := &fivetransdk.SchemaSelection{ Included: true, @@ -182,7 +182,7 @@ func TestCanSerializeMappedEnumsAndSets(t *testing.T) { "customer_type": {columnType: "enum", values: []string{"employee", "customer"}}, }, }, - }) + }, false) schema := &fivetransdk.SchemaSelection{ Included: true, @@ -271,7 +271,7 @@ func TestCanSerializeIndexedEnumsAndSets(t *testing.T) { "customer_type": {columnType: "enum", values: []string{"employee", "customer"}}, }, }, - }) + }, false) schema := &fivetransdk.SchemaSelection{ Included: true, @@ -336,7 +336,7 @@ func TestCanSerializeNulLValues(t *testing.T) { } tl := &testLogSender{} - l := NewSchemaAwareSerializer(tl, "", false, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}) + l := NewSchemaAwareSerializer(tl, "", false, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}, false) schema := &fivetransdk.SchemaSelection{ Included: true, SchemaName: s.Name, @@ -376,7 +376,7 @@ func TestCanSerializeDelete(t *testing.T) { row, s, err := generateTestRecord("PhaniRaj") require.NoError(t, err) tl := &testLogSender{} - l := NewSchemaAwareSerializer(tl, "", false, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}) + l := NewSchemaAwareSerializer(tl, "", false, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}, false) schema := &fivetransdk.SchemaSelection{ Included: true, @@ -416,7 +416,7 @@ func TestCanSerializeUpdate(t *testing.T) { require.NoError(t, err) tl := &testLogSender{} - l := NewSchemaAwareSerializer(tl, "", false, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}) + l := NewSchemaAwareSerializer(tl, "", false, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}, false) schema := &fivetransdk.SchemaSelection{ Included: true, @@ -462,7 +462,7 @@ func TestCanSerializeTruncate(t *testing.T) { _, s, err := generateTestRecord("PhaniRaj") assert.NoError(t, err) tl := &testLogSender{} - l := NewSchemaAwareSerializer(tl, "", false, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}) + l := NewSchemaAwareSerializer(tl, "", false, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}, false) schema := &fivetransdk.SchemaSelection{ Included: true, @@ -797,7 +797,7 @@ func TestCanSkipColumns(t *testing.T) { }, }, }}, - map[string]map[string]map[string]ValueMap{}) + map[string]map[string]map[string]ValueMap{}, false) schema := &fivetransdk.SchemaSelection{ Included: true, @@ -841,7 +841,7 @@ func BenchmarkRecordSerialization_Serializer(b *testing.B) { s, }, }, - map[string]map[string]map[string]ValueMap{}) + map[string]map[string]map[string]ValueMap{}, false) schema := &fivetransdk.SchemaSelection{ SchemaName: "SalesDB", @@ -857,3 +857,88 @@ func BenchmarkRecordSerialization_Serializer(b *testing.B) { } } } + +// The widened VStream projection delivers values for columns Fivetran never +// selected; the serializer decides whether they survive into the Record. +func TestSerializer_AdoptsColumnsAbsentFromSelection(t *testing.T) { + cases := []struct { + name string + propagateNewColumns bool + includeNewColumns bool + // deselect explicitly records the column as excluded rather than + // leaving it unnamed. + deselectExplicitly bool + expectSerialized bool + }{ + { + name: "adopted when both opt-ins are set", + propagateNewColumns: true, + includeNewColumns: true, + expectSerialized: true, + }, + { + name: "dropped by default", + propagateNewColumns: false, + includeNewColumns: true, + expectSerialized: false, + }, + { + name: "dropped when the table disallows new columns", + propagateNewColumns: true, + includeNewColumns: false, + expectSerialized: false, + }, + { + name: "an explicitly deselected column is never adopted", + propagateNewColumns: true, + includeNewColumns: true, + deselectExplicitly: true, + expectSerialized: false, + }, + } + + const newColumn = "notes" + + for _, tt := range cases { + t.Run(tt.name, func(t *testing.T) { + row, s, err := generateTestRecord("PhaniRaj") + require.NoError(t, err) + tl := &testLogSender{} + l := NewSchemaAwareSerializer(tl, "", true, &fivetransdk.SchemaList{Schemas: []*fivetransdk.Schema{s}}, map[string]map[string]map[string]ValueMap{}, tt.propagateNewColumns) + + schema := &fivetransdk.SchemaSelection{Included: true, SchemaName: s.Name} + table := &fivetransdk.TableSelection{ + TableName: "Customers", + Included: true, + Columns: map[string]bool{}, + IncludeNewColumns: tt.includeNewColumns, + } + for _, f := range row.Fields { + if f.Name == newColumn { + if tt.deselectExplicitly { + table.Columns[f.Name] = false + } + continue + } + table.Columns[f.Name] = true + } + + require.NoError(t, l.Record(row, schema, table, lib.OpType_Insert)) + require.NotNil(t, tl.lastResponse) + + operationRecord, ok := tl.lastResponse.Operation.(*fivetransdk.UpdateResponse_Record) + require.Truef(t, ok, "recordResponse Operation is not of type %s", "UpdateResponse_Record") + data := operationRecord.Record.Data + + // A selected column always comes through, so a missing new column is + // a real exclusion rather than a broken record. + assert.Equal(t, int32(123), data["customer_id"].GetInt()) + + if tt.expectSerialized { + assert.Equal(t, "string:\"Something great comes this way\"", data[newColumn].String()) + } else { + assert.NotContains(t, data, newColumn) + } + }) + } +} diff --git a/cmd/internal/server/handlers/sync.go b/cmd/internal/server/handlers/sync.go index 737a903..049e045 100644 --- a/cmd/internal/server/handlers/sync.go +++ b/cmd/internal/server/handlers/sync.go @@ -79,7 +79,7 @@ func (s *Sync) Handle(ctx context.Context, psc *lib.PlanetScaleSource, db *lib.C } columns := includedColumns(table) - sc, err := (*db).Read(ctx, logger, *psc, table.TableName, columns, tc, onRow, onCursor, onUpdate) + sc, err := (*db).Read(ctx, logger, *psc, table.TableName, columns, table.IncludeNewColumns, tc, onRow, onCursor, onUpdate) if err != nil { if lib.IsVStreamSchemaIncompatibilityError(err) { return status.Error(codes.FailedPrecondition, err.Error()) diff --git a/cmd/internal/server/handlers/sync_test.go b/cmd/internal/server/handlers/sync_test.go index 6d49cd6..6480984 100644 --- a/cmd/internal/server/handlers/sync_test.go +++ b/cmd/internal/server/handlers/sync_test.go @@ -44,7 +44,7 @@ func TestCallsReadWithSelectedSchema(t *testing.T) { } 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, + includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate, ) (*lib.SerializedCursor, error) { assert.Equal(t, "sync-request", ctx.Value(contextKey{})) assert.Equal(t, "customers", tableName) @@ -89,7 +89,7 @@ func TestCallsTruncateOnInitialSync(t *testing.T) { } 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, + includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate, ) (*lib.SerializedCursor, error) { assert.Equal(t, "customers", tableName) return nil, nil @@ -135,7 +135,7 @@ func TestInitialSyncReturnsErrorWhenTruncateFails(t *testing.T) { readCalled := false 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, + includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate, ) (*lib.SerializedCursor, error) { readCalled = true return nil, nil @@ -191,7 +191,7 @@ func TestCallsReadWithStartingGtids(t *testing.T) { } 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, + includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate, ) (*lib.SerializedCursor, error) { assert.Equal(t, "customers", tableName) return nil, nil @@ -251,7 +251,7 @@ func TestVStreamSchemaIncompatibilityReturnsFailedPrecondition(t *testing.T) { } 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, + includeNewColumns bool, 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"+ @@ -300,7 +300,7 @@ func TestCheckpointsHistoricalCopyCursorFromRead(t *testing.T) { LastKnownPk: testLastKnownPK("42"), } 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, + includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate, ) (*lib.SerializedCursor, error) { return nil, onCursor(copyCursor) } @@ -351,7 +351,7 @@ func TestDoesNotTruncateWhenStateHasHistoricalCopyProgress(t *testing.T) { readCalled := false 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, + includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate, ) (*lib.SerializedCursor, error) { readCalled = true assert.Empty(t, tc.Position) diff --git a/cmd/internal/server/server.go b/cmd/internal/server/server.go index 0d4d19c..f8d150a 100644 --- a/cmd/internal/server/server.go +++ b/cmd/internal/server/server.go @@ -174,7 +174,7 @@ func (c *connectorServer) Update(request *fivetran_sdk_v2.UpdateRequest, server return status.Errorf(codes.InvalidArgument, "unable get source schema for this database : %q", err) } - logger := handlers.NewSchemaAwareSerializer(server, requestId, psc.TreatTinyIntAsBoolean, sourceSchema.SchemaList, sourceSchema.EnumsAndSets) + logger := handlers.NewSchemaAwareSerializer(server, requestId, psc.TreatTinyIntAsBoolean, sourceSchema.SchemaList, sourceSchema.EnumsAndSets, psc.PropagateNewColumns) shards, err := db.ListShards(ctx, *psc) if err != nil { diff --git a/cmd/internal/server/server_test.go b/cmd/internal/server/server_test.go index ee165a5..7109940 100644 --- a/cmd/internal/server/server_test.go +++ b/cmd/internal/server/server_test.go @@ -291,7 +291,7 @@ func geometryTypeTest(t *testing.T, geometry []byte, geojson string) { return nil }, 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, + includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate, ) (*lib.SerializedCursor, error) { assert.Equal(t, "customers", tableName) assert.NotNil(t, columns) @@ -389,7 +389,7 @@ func TestUpdateReturnsInserts(t *testing.T) { return nil }, 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, + includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate, ) (*lib.SerializedCursor, error) { assert.Equal(t, "customers", tableName) assert.NotNil(t, columns) @@ -508,7 +508,7 @@ func TestUpdateReturnsErrors(t *testing.T) { return nil }, 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, + includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate, ) (*lib.SerializedCursor, error) { assert.Equal(t, "customers", tableName) assert.NotNil(t, columns) @@ -592,7 +592,7 @@ func TestUpdateReturnsDeletes(t *testing.T) { return nil }, 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, + includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate, ) (*lib.SerializedCursor, error) { assert.Equal(t, "customers", tableName) assert.NotNil(t, columns) @@ -709,7 +709,7 @@ func TestUpdateReturnsUpdates(t *testing.T) { CanConnectFn: func(ctx context.Context, ps lib.PlanetScaleSource) error { return nil }, - 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) { + ReadFn: func(ctx context.Context, logger lib.DatabaseLogger, ps lib.PlanetScaleSource, tableName string, columns []string, includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate) (*lib.SerializedCursor, error) { assert.Equal(t, "customers", tableName) assert.NotNil(t, columns) onUpdate(&lib.UpdatedRow{ @@ -960,7 +960,7 @@ func TestUpdateReturnsState(t *testing.T) { }, 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, + includeNewColumns bool, tc *psdbconnect.TableCursor, onResult lib.OnResult, onCursor lib.OnCursor, onUpdate lib.OnUpdate, ) (*lib.SerializedCursor, error) { onCursor(&psdbconnect.TableCursor{ Position: "THIS_IS_A_VALID_GTID", @@ -1153,3 +1153,40 @@ func TestAutoResyncFlagRoundTripsFromConfigurationForm(t *testing.T) { _, err = SourceFromRequest(withConnDetails(map[string]string{fieldName: "yes-please"})) assert.Error(t, err, "non-boolean value should be rejected rather than silently ignored") } + +// Same round trip as the flag above, for the same reason: propagate_new_columns +// gates an experimental behaviour, so a drift between the form field name and +// the key the parser reads would leave the operator setting a toggle that does +// nothing. An absent or malformed value must never silently enable it. +func TestPropagateNewColumnsFlagRoundTripsFromConfigurationForm(t *testing.T) { + ctx := context.Background() + form, err := (handlers.ConfigurationForm{}).Handle(ctx, &fivetransdk.ConfigurationFormRequest{}) + assert.NoError(t, err) + + var fieldName string + for _, f := range form.Fields { + if f.Name == "propagate_new_columns" { + fieldName = f.Name + assert.NotNil(t, f.GetDropdownField(), "expected a dropdown, matching the other boolean fields") + assert.Equal(t, []string{"true", "false"}, f.GetDropdownField().DropdownField) + } + } + assert.NotEmpty(t, fieldName, "propagate_new_columns missing from the configuration form") + + on, err := SourceFromRequest(withConnDetails(map[string]string{fieldName: "true"})) + assert.NoError(t, err) + assert.True(t, on.PropagateNewColumns, "form field name does not match the key SourceFromRequest reads") + + off, err := SourceFromRequest(withConnDetails(map[string]string{fieldName: "false"})) + assert.NoError(t, err) + assert.False(t, off.PropagateNewColumns) + + // Absent key must default to false: the feature is experimental and must + // never switch itself on. + def, err := SourceFromRequest(withConnDetails(nil)) + assert.NoError(t, err) + assert.False(t, def.PropagateNewColumns) + + _, err = SourceFromRequest(withConnDetails(map[string]string{fieldName: "yes-please"})) + assert.ErrorContains(t, err, "propagate_new_columns is not a boolean") +} diff --git a/cmd/internal/server/types.go b/cmd/internal/server/types.go index 78c8ffc..7bbc446 100644 --- a/cmd/internal/server/types.go +++ b/cmd/internal/server/types.go @@ -109,6 +109,14 @@ func SourceFromRequest(request ConfiguredRequest) (*lib.PlanetScaleSource, error psc.AutoResyncOnSchemaChange = b } + if val, ok := configuration["propagate_new_columns"]; ok { + b, err := strconv.ParseBool(val) + if err != nil { + return nil, errors.New("propagate_new_columns is not a boolean") + } + psc.PropagateNewColumns = b + } + if val, ok := configuration["starting_gtids"]; ok { psc.StartingGtids = val } diff --git a/lib/connect_client.go b/lib/connect_client.go index 655acf8..188bb25 100644 --- a/lib/connect_client.go +++ b/lib/connect_client.go @@ -48,11 +48,20 @@ var ( maxConsecutiveSyncTimeouts = 5 ) +// errAdoptNewColumns is an internal signal from sync to Read: the stream +// stopped cleanly at a schema change so the projection can be widened before +// resuming. It never reaches Fivetran. +var errAdoptNewColumns = errors.New("schema change observed; rebuild the column projection") + // ConnectClient is a general purpose interface // that defines all the data access methods needed for the PlanetScale Fivetran source to function. type ConnectClient interface { CanConnect(ctx context.Context, ps PlanetScaleSource) error - Read(ctx context.Context, logger DatabaseLogger, ps PlanetScaleSource, tableName string, columns []string, lastKnownPosition *psdbconnect.TableCursor, onResult OnResult, onCursor OnCursor, onUpdate OnUpdate) (*SerializedCursor, error) + // Read streams a table from lastKnownPosition. columns is Fivetran's + // selection for the table; includeNewColumns is Fivetran's per-table + // include_new_columns choice, honoured only when the source opts in via + // PropagateNewColumns. + Read(ctx context.Context, logger DatabaseLogger, ps PlanetScaleSource, tableName string, columns []string, includeNewColumns bool, lastKnownPosition *psdbconnect.TableCursor, onResult OnResult, onCursor OnCursor, onUpdate OnUpdate) (*SerializedCursor, error) ListShards(ctx context.Context, ps PlanetScaleSource) ([]string, error) } @@ -113,7 +122,7 @@ func (p connectClient) checkEdgePassword(ctx context.Context, psc PlanetScaleSou // 3. Ask vstream to stream from the last known vgtid // 4. When we reach the stopping point, read all rows available at this vgtid // 5. End the stream when (a) a vgtid newer than latest vgtid is encountered or (b) the timeout kicks in. -func (p connectClient) Read(ctx context.Context, logger DatabaseLogger, ps PlanetScaleSource, tableName string, columns []string, lastKnownPosition *psdbconnect.TableCursor, onResult OnResult, onCursor OnCursor, onUpdate OnUpdate) (*SerializedCursor, error) { +func (p connectClient) Read(ctx context.Context, logger DatabaseLogger, ps PlanetScaleSource, tableName string, columns []string, includeNewColumns bool, lastKnownPosition *psdbconnect.TableCursor, onResult OnResult, onCursor OnCursor, onUpdate OnUpdate) (*SerializedCursor, error) { var ( err error sErr error @@ -140,6 +149,23 @@ func (p connectClient) Read(ctx context.Context, logger DatabaseLogger, ps Plane logger.Info(fmt.Sprintf("%sCouldn't fetch existing columns, falling back to requested columns: %s", preamble, err.Error())) } + // Adopting a column that Fivetran has not selected is only safe once the + // replay position is at or past the DDL that added it -- naming it earlier + // is exactly what makes vstreamer fail to build a plan against a pre-DDL + // TABLE_MAP. So the stream starts on the selection as-is and only widens + // after it observes the DDL event for this table. + adoptNewColumns := ps.PropagateNewColumns && includeNewColumns + if adoptNewColumns { + logger.Info(fmt.Sprintf("%sNew columns will be adopted into the projection when a schema change is observed for this table", preamble)) + } + // Resuming at a DDL must not re-deliver that same DDL, or the stream would + // stop and rebuild forever without making progress. That should be + // impossible -- the cursor is past the DDL by the time we act on it -- but + // a spin here would burn a sync window silently, so make it structurally + // unreachable rather than relying on the position semantics. + lastRebuildPosition := "" + rebuiltAtLeastOnce := false + logger.Info(fmt.Sprintf("%sFiltering with columns %s", preamble, strings.Join(existingColumns, ","))) logger.Info(fmt.Sprintf("%sUsing read timeout: %v", preamble, readDuration)) @@ -161,7 +187,7 @@ func (p connectClient) Read(ctx context.Context, logger DatabaseLogger, ps Plane logger.Info(fmt.Sprintf(preamble+"syncing rows with cursor [%v]", currentPosition)) previousPosition := cloneTableCursor(currentPosition) - currentPosition, err = p.sync(ctx, logger, tableName, existingColumns, currentPosition, latestCursorPosition, ps, tabletType, readDuration, onResult, onCursor, onUpdate) + currentPosition, err = p.sync(ctx, logger, tableName, existingColumns, adoptNewColumns, currentPosition, latestCursorPosition, ps, tabletType, readDuration, onResult, onCursor, onUpdate) madeProgress := tableCursorMadeProgress(previousPosition, currentPosition) if tableCursorHasProgress(currentPosition) { currentSerializedCursor, sErr = TableCursorToSerializedCursor(currentPosition) @@ -171,6 +197,37 @@ func (p connectClient) Read(ctx context.Context, logger DatabaseLogger, ps Plane } } if err != nil { + // The stream stopped at a schema change so the projection can be + // rebuilt against the new schema. The cursor is already at the DDL + // (vstreamer emits a GTID immediately before the DDL event, which + // vtgate converts to the VGTID we consumed), so resuming from here + // replays no pre-DDL row events and the widened projection + // resolves. + if errors.Is(err, errAdoptNewColumns) { + if rebuiltAtLeastOnce && currentPosition.Position == lastRebuildPosition { + logger.Warning(fmt.Sprintf("%sSchema change observed again at position [%v] with no progress; leaving the projection alone for the rest of this read", preamble, currentPosition.Position)) + adoptNewColumns = false + continue + } + lastRebuildPosition = currentPosition.Position + rebuiltAtLeastOnce = true + + rebuild, rebuildErr := p.rebuildProjection(ctx, ps, tableName, existingColumns) + if rebuildErr != nil { + // Losing the rebuild is survivable -- keep streaming on the + // current projection rather than failing the table. + logger.Warning(fmt.Sprintf("%sCouldn't read columns after schema change, continuing with the current projection: %s", preamble, rebuildErr.Error())) + continue + } + if rebuild.changed() { + logger.Info(fmt.Sprintf("%sSchema change observed; rebuilding projection (added: [%s], removed: [%s])", preamble, strings.Join(rebuild.Added, ","), strings.Join(rebuild.Removed, ","))) + existingColumns = rebuild.Columns + } else { + logger.Info(fmt.Sprintf("%sSchema change observed with no column changes; continuing", preamble)) + } + continue + } + if s, ok := status.FromError(err); ok { // if the error is anything other than server timeout, keep going if s.Code() != codes.DeadlineExceeded { @@ -292,7 +349,7 @@ func (p connectClient) Read(ctx context.Context, logger DatabaseLogger, ps Plane } } -func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableName string, columns []string, tc *psdbconnect.TableCursor, stopPosition string, ps PlanetScaleSource, tabletType psdbconnect.TabletType, readDuration time.Duration, onResult OnResult, onCursor OnCursor, onUpdate OnUpdate) (*psdbconnect.TableCursor, error) { +func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableName string, columns []string, stopOnDDL bool, tc *psdbconnect.TableCursor, stopPosition string, ps PlanetScaleSource, tabletType psdbconnect.TabletType, readDuration time.Duration, onResult OnResult, onCursor OnCursor, onUpdate OnUpdate) (*psdbconnect.TableCursor, error) { ctx, cancel := context.WithTimeout(ctx, readDuration) defer cancel() @@ -331,17 +388,39 @@ func (p connectClient) sync(ctx context.Context, logger DatabaseLogger, tableNam } copyCompleted := false + schemaChanged := false for _, event := range res.Events { var ( eventRecords int eventCopyCompleted bool + eventSchemaChanged bool ) - tc, eventRecords, eventCopyCompleted, err = handleVStreamEvent(tableName, tc, event, fieldsByTable, onResult, onUpdate) + tc, eventRecords, eventCopyCompleted, eventSchemaChanged, err = handleVStreamEvent(tableName, tc, event, fieldsByTable, onResult, onUpdate) if err != nil { return syncStartCursor, err } recordsSinceCheckpoint += eventRecords copyCompleted = copyCompleted || eventCopyCompleted + if eventSchemaChanged && stopOnDDL { + schemaChanged = true + // Stop on the first DDL rather than draining the response: any + // later event in this batch could be a row event that the + // widened projection should have decoded. + break + } + } + + // Hand the cursor back at the DDL so Read can widen the projection and + // resume from here. Checkpoint first so the rows already streamed in + // this session are not replayed. + if schemaChanged { + logger.Info(fmt.Sprintf("%sSchema change observed at position [%v], stopping to rebuild the projection", preamble, tc.Position)) + if onCursor != nil { + if err := onCursor(tc); err != nil { + return tc, status.Error(codes.Internal, "unable to serialize cursor") + } + } + return tc, errAdoptNewColumns } // A single VGTID can appear in multiple ordered responses. Once we reach @@ -706,7 +785,10 @@ func quoteVStreamIdentifier(identifier string) string { return "`" + strings.ReplaceAll(identifier, "`", "``") + "`" } -func handleVStreamEvent(tableName string, cursor *psdbconnect.TableCursor, event *binlogdatapb.VEvent, fieldsByTable map[string][]*query.Field, onResult OnResult, onUpdate OnUpdate) (*psdbconnect.TableCursor, int, bool, error) { +// handleVStreamEvent applies a single VStream event, returning the advanced +// cursor, the number of records emitted, whether the historical copy finished, +// and whether the event was a schema change for this table. +func handleVStreamEvent(tableName string, cursor *psdbconnect.TableCursor, event *binlogdatapb.VEvent, fieldsByTable map[string][]*query.Field, onResult OnResult, onUpdate OnUpdate) (*psdbconnect.TableCursor, int, bool, bool, error) { switch event.Type { case binlogdatapb.VEventType_FIELD: if event.FieldEvent != nil { @@ -715,19 +797,24 @@ func handleVStreamEvent(tableName string, cursor *psdbconnect.TableCursor, event case binlogdatapb.VEventType_ROW: count, err := handleVStreamRows(tableName, event.RowEvent, fieldsByTable, onResult, onUpdate) if err != nil { - return cursor, 0, false, err + return cursor, 0, false, false, err } - return cursor, count, false, nil + return cursor, count, false, false, nil case binlogdatapb.VEventType_VGTID: next, err := tableCursorFromVGtid(cursor, event.Vgtid, tableName) - return next, 0, false, err + return next, 0, false, false, err case binlogdatapb.VEventType_LASTPK: next, err := tableCursorFromLastPK(cursor, event.LastPKEvent, tableName) - return next, 0, false, err + return next, 0, false, false, err case binlogdatapb.VEventType_COPY_COMPLETED: - return cursor, 0, true, nil - } - return cursor, 0, false, nil + return cursor, 0, true, false, nil + case binlogdatapb.VEventType_DDL: + // vstreamer only sends DDL events whose statement matches a table in + // the stream's filter (mustSendDDL -> tableMatches), and this stream + // filters on a single table, so no statement parsing is needed here. + return cursor, 0, false, true, nil + } + return cursor, 0, false, false, nil } func handleVStreamRows(tableName string, rowEvent *binlogdatapb.RowEvent, fieldsByTable map[string][]*query.Field, onResult OnResult, onUpdate OnUpdate) (int, error) { @@ -905,6 +992,63 @@ func (p connectClient) filterExistingColumns(ctx context.Context, ps PlanetScale return existingColumns, err } +// projectionRebuild is the result of recomputing the VStream projection against +// a table's current columns. +type projectionRebuild struct { + Columns []string + Added []string + Removed []string +} + +func (r projectionRebuild) changed() bool { + return len(r.Added) > 0 || len(r.Removed) > 0 +} + +// rebuildProjection recomputes the projection against the table's current +// columns: the previously projected columns that still exist, plus columns the +// database has that Fivetran never named at all. Columns Fivetran explicitly +// deselected are absent from projected and are not recovered here -- only +// never-named columns are adopted, which is what include_new_columns means. +// +// Dropping columns that no longer exist matters as much as adding new ones: +// naming a column that has gone away fails the post-DDL replay the same way +// naming one that did not yet exist fails a pre-DDL replay. +func (p connectClient) rebuildProjection(ctx context.Context, ps PlanetScaleSource, tableName string, projected []string) (projectionRebuild, error) { + results, err := (*p.Mysql).GetKeyspaceTableColumns(ctx, ps.Database, tableName) + if err != nil { + return projectionRebuild{Columns: projected}, err + } + + live := make(map[string]bool, len(results)) + for _, result := range results { + live[result.Name] = true + } + projectedSet := make(map[string]bool, len(projected)) + for _, c := range projected { + projectedSet[c] = true + } + + rebuild := projectionRebuild{ + Columns: make([]string, 0, len(results)), + Added: []string{}, + Removed: []string{}, + } + for _, c := range projected { + if live[c] { + rebuild.Columns = append(rebuild.Columns, c) + } else { + rebuild.Removed = append(rebuild.Removed, c) + } + } + for _, result := range results { + if !projectedSet[result.Name] { + rebuild.Columns = append(rebuild.Columns, result.Name) + rebuild.Added = append(rebuild.Added, result.Name) + } + } + return rebuild, nil +} + func (p connectClient) getLatestCursorPosition(ctx context.Context, shard, keyspace string, tableName string, ps PlanetScaleSource, tabletType psdbconnect.TabletType) (string, error) { timeout := 45 * time.Second ctx, cancel := context.WithTimeout(ctx, timeout) diff --git a/lib/connect_client_test.go b/lib/connect_client_test.go index e30c6a0..bb04633 100644 --- a/lib/connect_client_test.go +++ b/lib/connect_client_test.go @@ -66,7 +66,7 @@ func TestRead_CanPeekBeforeRead(t *testing.T) { onCursor := func(*psdbconnect.TableCursor) error { return nil } - sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, tc, onRow, onCursor, nil) + sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, false, tc, onRow, onCursor, nil) assert.NoError(t, err) esc, err := TableCursorToSerializedCursor(tc) assert.NoError(t, err) @@ -110,7 +110,7 @@ func TestRead_CanEarlyExitIfNoNewVGtidInPeek(t *testing.T) { onCursor := func(*psdbconnect.TableCursor) error { return nil } - sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, tc, onRow, onCursor, nil) + sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, false, tc, onRow, onCursor, nil) assert.NoError(t, err) esc, err := TableCursorToSerializedCursor(tc) assert.NoError(t, err) @@ -143,7 +143,7 @@ func TestRead_ReturnsLatestCursorSyncError(t *testing.T) { return &cc, nil } - sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{}, "customers", nil, tc, nil, nil, nil) + sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{}, "customers", nil, false, tc, nil, nil, nil) assert.Nil(t, sc) assert.ErrorContains(t, err, "Unable to get latest cursor position") assert.ErrorContains(t, err, "sync unavailable") @@ -175,7 +175,7 @@ func TestRead_ReturnsLatestCursorRecvError(t *testing.T) { return &cc, nil } - sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{}, "customers", nil, tc, nil, nil, nil) + sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{}, "customers", nil, false, tc, nil, nil, nil) assert.Nil(t, sc) assert.ErrorContains(t, err, "Unable to get latest cursor position") assert.ErrorContains(t, err, "EOF") @@ -221,7 +221,7 @@ func TestRead_CanPickPrimaryForShardedKeyspaces(t *testing.T) { onCursor := func(*psdbconnect.TableCursor) error { return nil } - sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, tc, onRow, onCursor, nil) + sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, false, tc, onRow, onCursor, nil) assert.NoError(t, err) esc, err := TableCursorToSerializedCursor(tc) assert.NoError(t, err) @@ -269,7 +269,7 @@ func TestRead_CanPickReplicaForShardedKeyspaces(t *testing.T) { onCursor := func(*psdbconnect.TableCursor) error { return nil } - sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, tc, onRow, onCursor, nil) + sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, false, tc, onRow, onCursor, nil) assert.NoError(t, err) esc, err := TableCursorToSerializedCursor(tc) assert.NoError(t, err) @@ -323,7 +323,7 @@ func TestRead_CanReturnNewCursorIfNewFound(t *testing.T) { mysqlClient := NewTestMysqlClient(getKeyspaceTableColumnsFunc) ped.Mysql = &mysqlClient - sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, tc, onRow, onCursor, nil) + sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, false, tc, onRow, onCursor, nil) assert.NoError(t, err) esc, err := TableCursorToSerializedCursor(newTC) assert.NoError(t, err) @@ -376,7 +376,7 @@ func TestRead_SchemaIncompatibilityResetsCursor(t *testing.T) { ped.Mysql = &mysqlClient source := PlanetScaleSource{Database: "connect-test", AutoResyncOnSchemaChange: true} - sc, err := ped.Read(context.Background(), dbl, source, "customers", []string{"id", "before_col", "after_col"}, tc, nil, nil, nil) + sc, err := ped.Read(context.Background(), dbl, source, "customers", []string{"id", "before_col", "after_col"}, false, tc, nil, nil, nil) assert.ErrorContains(t, err, "peek failed after reset") if assert.NotNil(t, sc) { cursor, cErr := sc.SerializedCursorToTableCursor() @@ -436,7 +436,7 @@ func TestRead_SchemaIncompatibilityWithoutOptInReturnsError(t *testing.T) { mysqlClient := NewTestMysqlClient(getKeyspaceTableColumnsFunc) ped.Mysql = &mysqlClient - sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id", "before_col", "after_col"}, tc, nil, nil, nil) + sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id", "before_col", "after_col"}, false, tc, nil, nil, nil) assert.Error(t, err) assert.Contains(t, err.Error(), "historical re-sync") assert.Contains(t, err.Error(), "auto_resync_on_schema_change") @@ -484,7 +484,7 @@ func TestRead_ReturnsGenericNonTimeoutErrors(t *testing.T) { mysqlClient := NewTestMysqlClient(getKeyspaceTableColumnsFunc) ped.Mysql = &mysqlClient - sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, tc, nil, nil, nil) + sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, false, tc, nil, nil, nil) assert.Error(t, err) assert.Contains(t, err.Error(), "tablet unavailable") assert.NotContains(t, err.Error(), "historical re-sync") @@ -596,7 +596,7 @@ func TestRead_CallbackErrorsReturnStartCursor(t *testing.T) { mysqlClient := NewTestMysqlClient(getKeyspaceTableColumnsFunc) ped.Mysql = &mysqlClient - sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", nil, tc, tt.onRow, nil, tt.onUpdate) + sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", nil, false, tc, tt.onRow, nil, tt.onUpdate) assert.Error(t, err) assert.Contains(t, err.Error(), tt.errorMessage) esc, err := TableCursorToSerializedCursor(tc) @@ -713,7 +713,7 @@ func TestRead_CanStopAtWellKnownCursor(t *testing.T) { mysqlClient := NewTestMysqlClient(getKeyspaceTableColumnsFunc) ped.Mysql = &mysqlClient - sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, responses[0].Cursor, onRow, onCursor, nil) + sc, err := ped.Read(context.Background(), dbl, ps, "customers", nil, false, responses[0].Cursor, onRow, onCursor, nil) assert.NoError(t, err) // sync should start at the first vgtid @@ -817,7 +817,7 @@ func TestRead_FiltersNonExistentColumns(t *testing.T) { onCursor := func(*psdbconnect.TableCursor) error { return nil } - sc, err := ped.Read(ctx, dbl, ps, "customers", tt.requestedColumns, tc, onRow, onCursor, nil) + sc, err := ped.Read(ctx, dbl, ps, "customers", tt.requestedColumns, false, tc, onRow, onCursor, nil) assert.NoError(t, err) esc, err := TableCursorToSerializedCursor(newTC) assert.NoError(t, err) @@ -876,7 +876,7 @@ func TestRead_ReturnsLastKnownPKCursorAfterMaxNoProgressTimeout(t *testing.T) { return &cc, nil } - sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, copyCursor, nil, nil, nil) + sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, false, copyCursor, nil, nil, nil) assert.NoError(t, err) if assert.NotNil(t, sc) { cursor, err := sc.SerializedCursorToTableCursor() @@ -965,7 +965,7 @@ func TestRead_DoesNotStopAfterProgressingTimeoutWindows(t *testing.T) { } rows := 0 - sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, initialCursor, func(*sqltypes.Result, Operation) error { + sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, false, initialCursor, func(*sqltypes.Result, Operation) error { rows++ return nil }, nil, nil) @@ -1034,7 +1034,7 @@ func TestRead_CancelDuringTimeoutBackoffReturnsImmediately(t *testing.T) { } start := time.Now() - _, err := ped.Read(ctx, dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, startCursor, nil, nil, nil) + _, err := ped.Read(ctx, dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, false, startCursor, nil, nil, nil) assert.ErrorIs(t, err, context.Canceled) assert.Less(t, time.Since(start), time.Second) } @@ -1089,7 +1089,7 @@ func TestRead_ResumesHistoricalCopyEvenWhenPeekMatchesCopyCursorPosition(t *test return &cc, nil } - sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, copyCursor, nil, nil, nil) + sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, false, copyCursor, nil, nil, nil) assert.NoError(t, err) if assert.NotNil(t, sc) { cursor, err := sc.SerializedCursorToTableCursor() @@ -1155,7 +1155,7 @@ func TestRead_BinlogExpirationReturnsResetCursor(t *testing.T) { return &cc, nil } - sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, initialCursor, nil, nil, nil) + sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id"}, false, initialCursor, nil, nil, nil) assert.ErrorContains(t, err, "peek failed after reset") if assert.NotNil(t, sc) { cursor, err := sc.SerializedCursorToTableCursor() @@ -1254,7 +1254,7 @@ func TestSync_DirectVStreamHandlesRawEventsAndCopyCompleted(t *testing.T) { records := 0 checkpoints := []*psdbconnect.TableCursor{} - returnedCursor, err := ped.sync(context.Background(), dbl, "customers", []string{"id", "name"}, startCursor, "STOP_GTID", PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, func(*sqltypes.Result, Operation) error { + returnedCursor, err := ped.sync(context.Background(), dbl, "customers", []string{"id", "name"}, false, startCursor, "STOP_GTID", PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, func(*sqltypes.Result, Operation) error { records++ return nil }, func(cursor *psdbconnect.TableCursor) error { @@ -1306,7 +1306,7 @@ func TestSync_DirectVStreamRejectsFieldlessLastKnownPK(t *testing.T) { return vc, nil } - returnedCursor, err := ped.sync(context.Background(), dbl, "customers", []string{"id", "name"}, startCursor, "STOP_GTID", PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, nil, nil, nil) + returnedCursor, err := ped.sync(context.Background(), dbl, "customers", []string{"id", "name"}, false, startCursor, "STOP_GTID", PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, nil, nil, nil) assert.Error(t, err) assert.Equal(t, codes.Internal, status.Code(err)) @@ -1381,7 +1381,7 @@ func TestSync_CheckpointsHistoricalCopyProgress(t *testing.T) { return nil } - returnedCursor, err := ped.sync(context.Background(), dbl, "customers", []string{"id", "name"}, startCursor, stopCursor.Position, PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, onResult, onCursor, nil) + returnedCursor, err := ped.sync(context.Background(), dbl, "customers", []string{"id", "name"}, false, startCursor, stopCursor.Position, PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, onResult, onCursor, nil) assert.True(t, errors.Is(err, io.EOF)) assert.True(t, proto.Equal(afterStopCursor, returnedCursor)) assert.Equal(t, 2, rows) @@ -1452,7 +1452,7 @@ func TestSync_DoesNotPeriodicallyCheckpointVGTIDProgress(t *testing.T) { return nil } - returnedCursor, err := ped.sync(context.Background(), dbl, "customers", []string{"id", "name"}, startCursor, stopCursor.Position, PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, func(*sqltypes.Result, Operation) error { + returnedCursor, err := ped.sync(context.Background(), dbl, "customers", []string{"id", "name"}, false, startCursor, stopCursor.Position, PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, func(*sqltypes.Result, Operation) error { return nil }, onCursor, nil) assert.True(t, errors.Is(err, io.EOF)) @@ -1484,7 +1484,7 @@ func TestSync_ResumesHistoricalCopyWithLastKnownPKOnly(t *testing.T) { return &cc, nil } - _, err := ped.sync(context.Background(), dbl, "customers", []string{"id"}, resumeCursor, "STOP_GTID", PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, nil, nil, nil) + _, err := ped.sync(context.Background(), dbl, "customers", []string{"id"}, false, resumeCursor, "STOP_GTID", PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, nil, nil, nil) assert.True(t, errors.Is(err, io.EOF)) assert.True(t, requestChecked) } @@ -1512,7 +1512,7 @@ func TestSync_ResumesHistoricalCopyWithPositionAndLastKnownPK(t *testing.T) { return &cc, nil } - _, err := ped.sync(context.Background(), dbl, "customers", []string{"id"}, resumeCursor, "STOP_GTID", PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, nil, nil, nil) + _, err := ped.sync(context.Background(), dbl, "customers", []string{"id"}, false, resumeCursor, "STOP_GTID", PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, nil, nil, nil) assert.True(t, errors.Is(err, io.EOF)) assert.True(t, requestChecked) } @@ -1576,3 +1576,449 @@ 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" + +func TestHandleVStreamEvent_ReportsDDLAsSchemaChange(t *testing.T) { + cursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "DDL_GTID"} + + returned, records, copyCompleted, schemaChanged, err := handleVStreamEvent( + "customers", + cursor, + &binlogdatapb.VEvent{Type: binlogdatapb.VEventType_DDL, Statement: "alter table customers add column new_col int"}, + map[string][]*query.Field{}, + nil, + nil, + ) + + assert.NoError(t, err) + assert.True(t, schemaChanged) + assert.False(t, copyCompleted) + assert.Equal(t, 0, records) + assert.True(t, proto.Equal(cursor, returned), "a DDL event must not move the cursor") +} + +func TestHandleVStreamEvent_OtherEventsAreNotSchemaChanges(t *testing.T) { + cursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test"} + + for _, eventType := range []binlogdatapb.VEventType{ + binlogdatapb.VEventType_BEGIN, + binlogdatapb.VEventType_COMMIT, + binlogdatapb.VEventType_OTHER, + binlogdatapb.VEventType_HEARTBEAT, + } { + _, _, _, schemaChanged, err := handleVStreamEvent("customers", cursor, &binlogdatapb.VEvent{Type: eventType}, map[string][]*query.Field{}, nil, nil) + assert.NoError(t, err) + assert.False(t, schemaChanged, "event %v must not be reported as a schema change", eventType) + } +} + +// A DDL is only a stopping point when Read asked for one. Without stopOnDDL the +// stream must behave exactly as it did before this feature existed. +func TestSync_IgnoresDDLWhenNotAdoptingNewColumns(t *testing.T) { + dbl := &dbLogger{} + ped := connectClient{} + + startCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test"} + ddlCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "DDL_GTID"} + stopCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "STOP_GTID"} + afterStopCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "AFTER_STOP_GTID"} + + rawStream := &vstreamClientMock{ + responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{ + vstreamVGtidEventFromCursor("connect-test.customers", ddlCursor), + {Type: binlogdatapb.VEventType_DDL, Statement: "alter table customers add column new_col int"}, + }}, + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", stopCursor)}}, + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", afterStopCursor)}}, + }, + } + vc := &vstreamConnectionMock{ + vstreamFn: func(ctx context.Context, in *vtgatepb.VStreamRequest, opts ...grpc.CallOption) (vtgateservicepb.Vitess_VStreamClient, error) { + return rawStream, nil + }, + } + ped.vstreamClientFn = func(ctx context.Context, ps PlanetScaleSource) (vstreamClient, error) { return vc, nil } + + returnedCursor, err := ped.sync(context.Background(), dbl, "customers", []string{"id", "name"}, false, startCursor, "STOP_GTID", PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, nil, nil, nil) + + assert.True(t, errors.Is(err, io.EOF), "expected the stream to run to its stop position, got %v", err) + assert.False(t, errors.Is(err, errAdoptNewColumns)) + assert.True(t, proto.Equal(afterStopCursor, returnedCursor)) +} + +func TestSync_StopsAtDDLWhenAdoptingNewColumns(t *testing.T) { + dbl := &dbLogger{} + ped := connectClient{} + + startCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test"} + ddlCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "DDL_GTID"} + testFields := sqltypes.MakeTestFields("id|name", "int64|varbinary") + rows := sqltypes.ResultToProto3(sqltypes.MakeTestResult(testFields, "1|keep-me")).Rows + + rawStream := &vstreamClientMock{ + responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{ + {Type: binlogdatapb.VEventType_FIELD, FieldEvent: &binlogdatapb.FieldEvent{ + TableName: "connect-test.customers", + Fields: testFields, + }}, + // A row written before the DDL: it must still be delivered on the + // narrow projection rather than being dropped by the early stop. + {Type: binlogdatapb.VEventType_ROW, RowEvent: &binlogdatapb.RowEvent{ + TableName: "connect-test.customers", + RowChanges: []*binlogdatapb.RowChange{{After: rows[0]}}, + }}, + vstreamVGtidEventFromCursor("connect-test.customers", ddlCursor), + {Type: binlogdatapb.VEventType_DDL, Statement: "alter table customers add column new_col int"}, + // Anything after the DDL belongs to the widened projection. + {Type: binlogdatapb.VEventType_ROW, RowEvent: &binlogdatapb.RowEvent{ + TableName: "connect-test.customers", + RowChanges: []*binlogdatapb.RowChange{{After: rows[0]}}, + }}, + }}, + }, + } + vc := &vstreamConnectionMock{ + vstreamFn: func(ctx context.Context, in *vtgatepb.VStreamRequest, opts ...grpc.CallOption) (vtgateservicepb.Vitess_VStreamClient, error) { + return rawStream, nil + }, + } + ped.vstreamClientFn = func(ctx context.Context, ps PlanetScaleSource) (vstreamClient, error) { return vc, nil } + + records := 0 + checkpoints := []*psdbconnect.TableCursor{} + returnedCursor, err := ped.sync(context.Background(), dbl, "customers", []string{"id", "name"}, true, startCursor, "STOP_GTID", PlanetScaleSource{Database: "connect-test"}, psdbconnect.TabletType_primary, time.Second, func(*sqltypes.Result, Operation) error { + records++ + return nil + }, func(cursor *psdbconnect.TableCursor) error { + checkpoints = append(checkpoints, cloneTableCursor(cursor)) + return nil + }, nil) + + assert.True(t, errors.Is(err, errAdoptNewColumns), "expected the adopt signal, got %v", err) + // The cursor must sit at the DDL so the widened projection resolves when the + // stream resumes -- this is the whole safety property of the DDL gate. + assert.True(t, proto.Equal(ddlCursor, returnedCursor)) + assert.Equal(t, 1, records, "the pre-DDL row is delivered, the post-DDL row is not") + if assert.Len(t, checkpoints, 1, "progress must be checkpointed before handing back") { + assert.True(t, proto.Equal(ddlCursor, checkpoints[0])) + } +} + +func TestRead_WidensProjectionAfterDDL(t *testing.T) { + dbl := &dbLogger{} + ped := connectClient{} + + startCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test"} + ddlCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "DDL_GTID"} + stopCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "STOP_GTID"} + afterStopCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "AFTER_STOP_GTID"} + + // The database has a column Fivetran never selected. + mysqlClient := NewTestMysqlClient(func(ctx context.Context, keyspaceName string, tableName string) ([]MysqlColumn, error) { + return []MysqlColumn{{Name: "id"}, {Name: "name"}, {Name: "new_col"}}, nil + }) + ped.Mysql = &mysqlClient + + filters := []string{} + vc := &vstreamConnectionMock{ + vstreamFn: func(ctx context.Context, in *vtgatepb.VStreamRequest, opts ...grpc.CallOption) (vtgateservicepb.Vitess_VStreamClient, error) { + // The peek that Read does before each sync window. + if in.Vgtid.ShardGtids[0].Gtid == "current" { + return &vstreamClientMock{responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", stopCursor)}}, + }}, nil + } + + filters = append(filters, in.Filter.Rules[0].Filter) + if len(filters) == 1 { + // First window: stop at the DDL. + return &vstreamClientMock{responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{ + vstreamVGtidEventFromCursor("connect-test.customers", ddlCursor), + {Type: binlogdatapb.VEventType_DDL, Statement: "alter table customers add column new_col int"}, + }}, + }}, nil + } + // Second window: run to the stop position. + return &vstreamClientMock{responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", stopCursor)}}, + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", afterStopCursor)}}, + }}, nil + }, + } + ped.vstreamClientFn = func(ctx context.Context, ps PlanetScaleSource) (vstreamClient, error) { return vc, nil } + + ps := PlanetScaleSource{Database: "connect-test", PropagateNewColumns: true} + sc, err := ped.Read(context.Background(), dbl, ps, "customers", []string{"id", "name"}, true, startCursor, nil, nil, nil) + + assert.NoError(t, err) + if assert.Len(t, filters, 2, "expected the stream to restart once after the DDL") { + assert.Equal(t, "SELECT `id`,`name` FROM `customers`", filters[0], "the first window must not name the unselected column") + assert.Equal(t, "SELECT `id`,`name`,`new_col` FROM `customers`", filters[1], "the second window must adopt it") + } + expected, err := TableCursorToSerializedCursor(afterStopCursor) + assert.NoError(t, err) + assert.Equal(t, expected, sc) +} + +// The opt-in flag and Fivetran's per-table choice must both be set; either one +// alone leaves the projection alone. +func TestRead_DoesNotWidenProjectionUnlessBothOptInsAreSet(t *testing.T) { + cases := []struct { + name string + propagateNewColumns bool + includeNewColumns bool + }{ + {name: "both off", propagateNewColumns: false, includeNewColumns: false}, + {name: "connector opt-in only", propagateNewColumns: true, includeNewColumns: false}, + {name: "fivetran selection only", propagateNewColumns: false, includeNewColumns: true}, + } + + for _, tt := range cases { + t.Run(tt.name, func(t *testing.T) { + dbl := &dbLogger{} + ped := connectClient{} + + startCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test"} + ddlCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "DDL_GTID"} + stopCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "STOP_GTID"} + afterStopCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "AFTER_STOP_GTID"} + + mysqlClient := NewTestMysqlClient(func(ctx context.Context, keyspaceName string, tableName string) ([]MysqlColumn, error) { + return []MysqlColumn{{Name: "id"}, {Name: "name"}, {Name: "new_col"}}, nil + }) + ped.Mysql = &mysqlClient + + filters := []string{} + vc := &vstreamConnectionMock{ + vstreamFn: func(ctx context.Context, in *vtgatepb.VStreamRequest, opts ...grpc.CallOption) (vtgateservicepb.Vitess_VStreamClient, error) { + if in.Vgtid.ShardGtids[0].Gtid == "current" { + return &vstreamClientMock{responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", stopCursor)}}, + }}, nil + } + filters = append(filters, in.Filter.Rules[0].Filter) + // The DDL is present either way; it just must not be acted on. + return &vstreamClientMock{responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{ + vstreamVGtidEventFromCursor("connect-test.customers", ddlCursor), + {Type: binlogdatapb.VEventType_DDL, Statement: "alter table customers add column new_col int"}, + vstreamVGtidEventFromCursor("connect-test.customers", stopCursor), + }}, + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", afterStopCursor)}}, + }}, nil + }, + } + ped.vstreamClientFn = func(ctx context.Context, ps PlanetScaleSource) (vstreamClient, error) { return vc, nil } + + ps := PlanetScaleSource{Database: "connect-test", PropagateNewColumns: tt.propagateNewColumns} + _, err := ped.Read(context.Background(), dbl, ps, "customers", []string{"id", "name"}, tt.includeNewColumns, startCursor, nil, nil, nil) + + assert.NoError(t, err) + if assert.Len(t, filters, 1, "the stream must not restart") { + assert.Equal(t, "SELECT `id`,`name` FROM `customers`", filters[0]) + } + }) + } +} + +func TestRebuildProjection(t *testing.T) { + tests := []struct { + name string + liveColumns []string + projected []string + expectedColumns []string + expectedAdded []string + expectedRemoved []string + expectedChanged bool + }{ + { + name: "adds columns the database has and Fivetran never named", + liveColumns: []string{"id", "name", "added_a", "added_b"}, + projected: []string{"id", "name"}, + expectedColumns: []string{"id", "name", "added_a", "added_b"}, + expectedAdded: []string{"added_a", "added_b"}, + expectedRemoved: []string{}, + expectedChanged: true, + }, + { + // Naming a dropped column fails the post-DDL replay exactly the way + // naming a not-yet-existing one fails a pre-DDL replay. + name: "removes columns that no longer exist", + liveColumns: []string{"id", "name"}, + projected: []string{"id", "name", "gone"}, + expectedColumns: []string{"id", "name"}, + expectedAdded: []string{}, + expectedRemoved: []string{"gone"}, + expectedChanged: true, + }, + { + name: "handles an add and a drop in one schema change", + liveColumns: []string{"id", "name", "added"}, + projected: []string{"id", "name", "gone"}, + expectedColumns: []string{"id", "name", "added"}, + expectedAdded: []string{"added"}, + expectedRemoved: []string{"gone"}, + expectedChanged: true, + }, + { + // An index-only DDL still emits a DDL event; it must be a no-op. + name: "reports no change when the layout is unchanged", + liveColumns: []string{"id", "name"}, + projected: []string{"id", "name"}, + expectedColumns: []string{"id", "name"}, + expectedAdded: []string{}, + expectedRemoved: []string{}, + expectedChanged: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ped := connectClient{} + mysqlClient := NewTestMysqlClient(func(ctx context.Context, keyspaceName string, tableName string) ([]MysqlColumn, error) { + columns := make([]MysqlColumn, 0, len(tt.liveColumns)) + for _, c := range tt.liveColumns { + columns = append(columns, MysqlColumn{Name: c}) + } + return columns, nil + }) + ped.Mysql = &mysqlClient + + rebuild, err := ped.rebuildProjection(context.Background(), PlanetScaleSource{Database: "connect-test"}, "customers", tt.projected) + assert.NoError(t, err) + assert.Equal(t, tt.expectedColumns, rebuild.Columns) + assert.Equal(t, tt.expectedAdded, rebuild.Added) + assert.Equal(t, tt.expectedRemoved, rebuild.Removed) + assert.Equal(t, tt.expectedChanged, rebuild.changed()) + }) + } +} + +func TestRebuildProjection_KeepsProjectionOnLookupFailure(t *testing.T) { + ped := connectClient{} + mysqlClient := NewTestMysqlClient(func(ctx context.Context, keyspaceName string, tableName string) ([]MysqlColumn, error) { + return nil, errors.New("information_schema unavailable") + }) + ped.Mysql = &mysqlClient + + rebuild, err := ped.rebuildProjection(context.Background(), PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id", "name"}) + assert.Error(t, err) + assert.Equal(t, []string{"id", "name"}, rebuild.Columns, "a failed lookup must not narrow or widen the projection") + assert.False(t, rebuild.changed()) +} + +// The same DDL gate that makes an added column safe also makes a dropped one +// safe: resuming at the DDL means no pre-DDL row events are replayed, so the +// rebuilt projection just stops naming the column that went away. +func TestRead_NarrowsProjectionAfterDroppedColumn(t *testing.T) { + dbl := &dbLogger{} + ped := connectClient{} + + startCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test"} + ddlCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "DDL_GTID"} + stopCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "STOP_GTID"} + afterStopCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "AFTER_STOP_GTID"} + + // "doomed" is selected and exists at the start; the DDL drops it. + liveColumns := []MysqlColumn{{Name: "id"}, {Name: "name"}, {Name: "doomed"}} + mysqlClient := NewTestMysqlClient(func(ctx context.Context, keyspaceName string, tableName string) ([]MysqlColumn, error) { + return liveColumns, nil + }) + ped.Mysql = &mysqlClient + + filters := []string{} + vc := &vstreamConnectionMock{ + vstreamFn: func(ctx context.Context, in *vtgatepb.VStreamRequest, opts ...grpc.CallOption) (vtgateservicepb.Vitess_VStreamClient, error) { + if in.Vgtid.ShardGtids[0].Gtid == "current" { + return &vstreamClientMock{responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", stopCursor)}}, + }}, nil + } + + filters = append(filters, in.Filter.Rules[0].Filter) + if len(filters) == 1 { + // The DDL drops the column, so the rebuild sees it gone. + liveColumns = []MysqlColumn{{Name: "id"}, {Name: "name"}} + return &vstreamClientMock{responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{ + vstreamVGtidEventFromCursor("connect-test.customers", ddlCursor), + {Type: binlogdatapb.VEventType_DDL, Statement: "alter table customers drop column doomed"}, + }}, + }}, nil + } + return &vstreamClientMock{responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", stopCursor)}}, + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", afterStopCursor)}}, + }}, nil + }, + } + ped.vstreamClientFn = func(ctx context.Context, ps PlanetScaleSource) (vstreamClient, error) { return vc, nil } + + ps := PlanetScaleSource{Database: "connect-test", PropagateNewColumns: true} + _, err := ped.Read(context.Background(), dbl, ps, "customers", []string{"id", "name", "doomed"}, true, startCursor, nil, nil, nil) + + assert.NoError(t, err) + if assert.Len(t, filters, 2, "expected the stream to restart once after the DDL") { + assert.Equal(t, "SELECT `id`,`name`,`doomed` FROM `customers`", filters[0]) + assert.Equal(t, "SELECT `id`,`name` FROM `customers`", filters[1], "the dropped column must leave the projection") + } +} + +// If a resume ever re-delivered the DDL it just stopped at, the naive loop +// would rebuild forever. The second stop at the same position must disarm +// adoption instead of spinning. +func TestRead_DoesNotSpinWhenDDLRepeatsAtTheSamePosition(t *testing.T) { + dbl := &dbLogger{} + ped := connectClient{} + + startCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test"} + ddlCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "DDL_GTID"} + stopCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "STOP_GTID"} + afterStopCursor := &psdbconnect.TableCursor{Shard: "-", Keyspace: "connect-test", Position: "AFTER_STOP_GTID"} + + mysqlClient := NewTestMysqlClient(func(ctx context.Context, keyspaceName string, tableName string) ([]MysqlColumn, error) { + return []MysqlColumn{{Name: "id"}, {Name: "name"}, {Name: "new_col"}}, nil + }) + ped.Mysql = &mysqlClient + + syncWindows := 0 + vc := &vstreamConnectionMock{ + vstreamFn: func(ctx context.Context, in *vtgatepb.VStreamRequest, opts ...grpc.CallOption) (vtgateservicepb.Vitess_VStreamClient, error) { + if in.Vgtid.ShardGtids[0].Gtid == "current" { + return &vstreamClientMock{responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", stopCursor)}}, + }}, nil + } + + syncWindows++ + if syncWindows > 10 { + t.Fatal("Read is spinning on the same DDL") + } + + // Always re-deliver the same DDL at the same position: the + // pathological case the guard exists for. + if syncWindows <= 2 { + return &vstreamClientMock{responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{ + vstreamVGtidEventFromCursor("connect-test.customers", ddlCursor), + {Type: binlogdatapb.VEventType_DDL, Statement: "alter table customers add column new_col int"}, + }}, + }}, nil + } + return &vstreamClientMock{responses: []*vtgatepb.VStreamResponse{ + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", stopCursor)}}, + {Events: []*binlogdatapb.VEvent{vstreamVGtidEventFromCursor("connect-test.customers", afterStopCursor)}}, + }}, nil + }, + } + ped.vstreamClientFn = func(ctx context.Context, ps PlanetScaleSource) (vstreamClient, error) { return vc, nil } + + ps := PlanetScaleSource{Database: "connect-test", PropagateNewColumns: true} + _, err := ped.Read(context.Background(), dbl, ps, "customers", []string{"id", "name"}, true, startCursor, nil, nil, nil) + + assert.NoError(t, err) + // One rebuild, one disarm, then the stream runs to its stop position. + assert.Equal(t, 3, syncWindows) +} diff --git a/lib/planetscale_source.go b/lib/planetscale_source.go index 7dc7fa0..6a6db13 100644 --- a/lib/planetscale_source.go +++ b/lib/planetscale_source.go @@ -26,6 +26,22 @@ type PlanetScaleSource struct { // undecodable. Defaults to false, which keeps the existing behaviour of // surfacing the error and waiting for an operator-triggered re-sync. AutoResyncOnSchemaChange bool `json:"auto_resync_on_schema_change"` + // PropagateNewColumns opts in to honouring Fivetran's per-table + // include_new_columns choice: when the stream sees a schema change, the + // projection is rebuilt against the table's current columns, so a column + // that exists in the database but was never named in the selection starts + // streaming, and one that has been dropped stops being requested. + // + // Rebuilding is deferred until the stream observes the DDL event for the + // table, because naming a column that did not exist at the replay position + // is exactly what makes a lagging cursor fail ("column X not found in + // table Y"). Resuming at the DDL means no pre-DDL row events are replayed, + // which is why this needs no help from vttablet's schema historian + // (--track-schema-versions). + // + // Defaults to false: the behaviour is not yet well exercised in + // production. + PropagateNewColumns bool `json:"propagate_new_columns"` } // DSN returns a DataSource that mysql libraries can use to connect to a PlanetScale database. diff --git a/lib/test_types.go b/lib/test_types.go index 06b5355..3a696f3 100644 --- a/lib/test_types.go +++ b/lib/test_types.go @@ -150,7 +150,7 @@ func NewTestMysqlClient(gktc GetKeyspaceTableColumnsFunc) MysqlClient { } type ( - ReadFunc func(ctx context.Context, logger DatabaseLogger, ps PlanetScaleSource, tableName string, columns []string, tc *psdbconnect.TableCursor, onResult OnResult, onCursor OnCursor, onUpdate OnUpdate) (*SerializedCursor, error) + ReadFunc func(ctx context.Context, logger DatabaseLogger, ps PlanetScaleSource, tableName string, columns []string, includeNewColumns bool, tc *psdbconnect.TableCursor, onResult OnResult, onCursor OnCursor, onUpdate OnUpdate) (*SerializedCursor, error) CanConnectFunc func(ctx context.Context, ps PlanetScaleSource) error ListShardsFunc func(ctx context.Context, ps PlanetScaleSource) ([]string, error) @@ -176,9 +176,9 @@ func (tcc *TestConnectClient) CanConnect(ctx context.Context, ps PlanetScaleSour return errors.New("CanConnect is Unimplemented") } -func (tcc *TestConnectClient) Read(ctx context.Context, logger DatabaseLogger, ps PlanetScaleSource, tableName string, columns []string, lastKnownPosition *psdbconnect.TableCursor, onResult OnResult, onCursor OnCursor, onUpdate OnUpdate) (*SerializedCursor, error) { +func (tcc *TestConnectClient) Read(ctx context.Context, logger DatabaseLogger, ps PlanetScaleSource, tableName string, columns []string, includeNewColumns bool, lastKnownPosition *psdbconnect.TableCursor, onResult OnResult, onCursor OnCursor, onUpdate OnUpdate) (*SerializedCursor, error) { if tcc.ReadFn != nil { - return tcc.ReadFn(ctx, logger, ps, tableName, columns, lastKnownPosition, onResult, onCursor, onUpdate) + return tcc.ReadFn(ctx, logger, ps, tableName, columns, includeNewColumns, lastKnownPosition, onResult, onCursor, onUpdate) } return nil, errors.New("Read is Unimplemented")