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