Propagate new columns behind an opt-in, gated on the DDL event

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",

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)
+}

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

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,14 +158,15 @@ 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,
 		serializeTinyIntAsBool: serializeTinyIntAsBool,
 		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

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)
+			}
+		})
+	}
+}

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())