Fix upsert after schema evolution

[kevinjqliu]

Closes #2467

Rationale for this change

Schema updates change the current table schema but do not create snapshots. The target branch can therefore still point to a snapshot with an older schema_id.

Before this change, upsert built its source and target rows with different schemas:

• The source is validated against self.table_metadata.schema(), the current transaction schema and the schema recorded on the new snapshot.
• The matched target rows came from a branch scan. use_ref() sets snapshot_id, which makes DataScan.projection() use the snapshot schema.

After adding created_at, this produced:

Source/current schema: cik_str, ticker, title, created_at
Target snapshot schema: cik_str, ticker, title

When get_rows_to_update casts the source to the target Arrow schema, PyArrow reports:

An error occurred during upsert: Target schema's field names are not matching the table's field names: ['cik_str', 'ticker', 'title', 'created_at'], ['cik_str', 'ticker', 'title']

The branch snapshot should determine which files to read, while the current transaction schema should define the rows being compared and written. This change plans files from the target branch and materializes them with the current transaction schema. Older files are projected into the current schema, so the source and matched target rows have the same columns. Snapshot-based projection remains unchanged for normal and time-travel scans.

Are these changes tested?

Yes. Added regression tests for main and a diverged branch.

Are there any user-facing changes?

Yes. Upsert works after schema evolution.

[kevinjqliu]

reviewer note:

the change here is only adding

branch: str | None = None,

[kevinjqliu]

simplifying the logic now that we can pass branch into _scan

[kevinjqliu]

this is now part of the _scan logic, by passing branch in directly

[kevinjqliu]

this is the main fix to ensure that we're using the same schema!

[Copilot]

Pull request overview

Copilot reviewed 2 out of 2 changed files in this pull request and generated no new comments.

[hedger9487]

Nice approach separating file planning from schema projection!

I was following #2467 and pulled this down to test it out locally, and noticed a small detail with _scan(branch=...): if a branch doesn't exist yet (or if there's a typo in the branch name), if branch in self.table_metadata.refs skips use_ref and silently falls back to returning the default scan on main.

In upsert(), this causes it to match rows against main (for example, running tbl.upsert(df, branch="new_branch") on a table with 1 row on main reports rows_updated=1). Curious if we should fail fast here (e.g. if branch is not None: return scan.use_ref(branch)) so it raises ValueError: Cannot scan unknown ref consistently?

Also, as a minor follow-up thought for the future: do you think it might eventually be worth exposing a projected schema override directly on DataScan (e.g. scan.with_schema(...) or project_as(...)), so operations don't need to manually orchestrate plan_files() with private helpers?