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

[orware]

Important

Stacked on #97 — please merge that one first.

The base branch here is feat/schema-incompat-cursor-reset (#97), not main, so the diff
below shows only this change rather than #97's as well. Once #97 merges, GitHub retargets
this PR to main automatically and drops the already-merged commits — no action needed.

The two are independent in substance: #97 recovers from a hard stall, this one stops new
columns from being silently dropped. They're stacked only because they both add a config
flag and both touch Read(), so reviewing them in order avoids four keep-both conflicts.
lib/connect_client.go itself merges cleanly between them.

What this fixes

Fivetran tells us, per table, whether new columns should be synced — TableSelection.include_new_columns — and the SDK's source-connector guide makes acting on it our job:

Make sure to handle new schemas/tables/columns per the information and user choices in UpdateRequest#selection.

We never read that field. includedColumns() looks only at TableSelection.columns, so a column added after a connection was set up is never named in the VStream projection and its values never arrive. Nothing errors, syncs report success, and only a historical re-sync recovers. This is the silent half of the 9r0duiqkwh1l investigation.

Why it waits for the DDL

We can't just widen the projection up front. Naming a column that didn't exist at the replay position is exactly what produces column X not found in table Y against a pre-DDL TABLE_MAP. So the stream starts on Fivetran's selection as-is and rebuilds the projection only once it observes the VEventType_DDL event for the table (which handleVStreamEvent previously discarded — there was no DDL case).

Resuming there is safe because the cursor is already past the schema change:

1. vstreamer.go emits a GTID event carrying EncodePosition(vs.pos) immediately before the DDL event.
2. vs.pos = AppendGTID(vs.pos, gtid) already ran when the preceding GTID event was processed, so the position includes the DDL's own GTID.
3. vstream_manager.go sendEventsLocked converts that GTID to the VGTID we consume in place, preserving order.

Only post-DDL row events are replayed, so their TABLE_MAP carries the new column and the plan resolves. This needs no --track-schema-versions.

mustSendDDL already scopes DDL events to tables matching the stream's filter, and our filter names one table, so no statement parsing is needed here.

Opt-in

Two gates, both required:

• propagate_new_columns — new source setting, defaults to false, labelled [Experimental] in the setup form.
• Fivetran's per-table include_new_columns.

A column Fivetran explicitly deselected is present-and-false in the map and stays excluded; only columns it never named at all are adopted. That matches include_new_columns semantics, and it's how Convex's connector models the same choice (other_columns: InclusionDefault).

The serializer needed the same change — columnSelection and columnWriters are both built from TableSelection.columns, so a widened projection would have been discarded on the way out, and a missing writer is a hard error rather than a silent skip.

Scope — verified against a live branch with tracking off, not inferred

Reproduced on a dedicated test database: seed row → capture cursor → unconsumed pre-DDL row → ALTER TABLE repro ADD COLUMN c_new int → post-DDL row with c_new=424242. Same stale cursor for every case.

Requested columns	setting / per-table	Result
base	off / on	no error, rows flow, c_new silently absent
base	on / off	no error, rows flow, c_new absent — gate holds
base	on / on	no error, c_new=424242 delivered
base + c_new	off / off	column c_new not found in table repro
base + c_new	on / on	identical to the row above — no change

Connector log for the fixed case:

Filtering with columns id,created_at,name,anchor_col,tail_col,status
Schema change observed at position [...a678ef2c...:1-109], stopping to rebuild the projection
Schema change observed; rebuilding projection (added: [c_new], removed: [])
ROW: id=2 ... name=poison ... status=active
ROW: id=3 ... name=post   ... status=archived c_new=424242

The pre-DDL row arrives on the narrow projection — correct, the column didn't exist yet — and the post-DDL row carries the real value.

What this does not change:

• Hard stalls. When Fivetran's selection already names the new column, filterExistingColumns keeps it (it exists live), so the first window fails before reaching the DDL and the gate never fires. The error is identical with the setting on or off and is still matched by IsVStreamSchemaIncompatibilityError, so recovery stays with Recognise all recoverable schema incompatibilities in vstream plan failures #96/Add opt-in auto re-sync on schema incompatibility #97.
• Dropped columns on a lagging cursor. The width check (cannot determine table columns) fires on the pre-DDL row before the DDL is reached, so that case still depends on the historian. rebuildProjection drops vanished columns for coherence, which only matters for a drop landing mid-Read.

Please don't "fix" the stall by loosening the projection

Falling back to SELECT * silently corrupts data across a mid-table ADD. analyzeExprs copies Table.Fields positionally for a StarExpr and never calls findColumn, so nothing catches the skew from the st.Fields[:len(tm.Types)] truncation. On a lagging cursor with ADD COLUMN newcol varchar(64) AFTER a over (id, a, b, c):

	id	a	newcol	b	c
ground truth	2	a2-poison	NULL	b2-poison	c2-poison
delivered by SELECT *	2	a2-poison	b2-poison	c2-poison	absent

No error; the sync reported success. The AreTypesCompatible guard only catches this when the shifted types happen to differ, so a same-typed tail sails through. The explicit projection from #92 is a guardrail, not the defect — it converted silent corruption into a loud FAILED_PRECONDITION.

Relatedly: Filter.FieldEventMode's zero value is ERR_ON_MISMATCH, so we get strict errors without opting in. Setting BEST_EFFORT would swap them for synthetic @1..@N field names that the serializer drops wholesale — silent total data loss. Don't.

Tests

Build, vet, gofmt and staticcheck clean; full suite passes. New coverage:

• handleVStreamEvent reports DDL as a schema change and doesn't move the cursor; other event types don't.
• sync stops at the DDL with the cursor at the DDL position, delivers the pre-DDL row, checkpoints before handing back — and ignores DDL entirely when not adopting, so existing behaviour is unchanged.
• Read widens the projection after a DDL (asserting the actual filter strings), narrows it after a drop, and does not widen unless both opt-ins are set.
• Read disarms adoption if the same position yields a DDL twice, so a resume that re-delivered its own DDL can't spin.
• rebuildProjection add/drop/both/no-op cases, and keeps the projection intact when the column lookup fails.
• Serializer adopts an unnamed column only when both opt-ins are set, and never a deselected one.
• propagate_new_columns parsing: defaults off, rejects garbage.
• The setup-form field is present, flagged experimental, and not required.

I verified the key test isn't vacuous by disabling the widening and confirming it fails.

🤖 Generated with Claude Code