GeezerVison

bug (medium): distributed hosts can now disagree on the schema. Hosts agree on the snapshot id through startup_choice, but each host calls open_table_with_retries() on its own, so each host picks up whatever schema is current when it loads the table. Here is how that goes wrong: host 0 loads the table, then ALTER TABLE ADD COLUMN c DEFAULT 'x' commits, then host 1 loads it. Host 1's used_columns includes c and reads 'x'. Host 0 doesn't see c and emits NULL. The result is mixed rows from one immutable snapshot. A drop or rename between the two loads is worse: a NOT NULL SQL column fails on only some hosts. Before this PR, the snapshot's schema made the read deterministic. Could host 0 also publish the schema id in startup_choice, so every host pins to it with set_current_schema? Resume has the same issue: a restart re-reads the pinned snapshot with whatever schema is current then.

pitfall (low-medium): this rebuilds the table with Runtime::try_current() and default readonly, so it drops the source table's runtime (for example, a split io/cpu runtime set by the catalog) and its readonly flag. If the iceberg fork exposes them, carry the original table's runtime and readonly over. At the least, note in the comment that they are reset on purpose.

bug (high): follow mode freezes the startup schema across restarts. push_follow_boundary writes the schema pinned at startup into every follow checkpoint. On restart, this arm restores it, and prepare_snapshot_query then builds used_columns from it, and the follow loop projects changed files by those columns. So a column added while following never gets ingested again, even after a restart. Before this PR, the resume used the last followed snapshot's schema, so a restart did pick up additive changes, as the comment in follow_loop_inner promises. And if that old schema is ever removed from the metadata, the restart fails with "no longer has schema N". Reproduced with this test, which fails with Some(0) != Some(1):

*endpoint.last_resume_status.lock().unwrap() =
    Some(super::IcebergResumeInfo::follow(2, Some(0)));
endpoint.resolved_snapshot_id(&table).unwrap();
assert_eq!(endpoint.pinned_schema_id(), Some(table.metadata().current_schema_id()));

Suggestion: when following is set, record None (or resolve the current schema on resume), and keep the pin only for snapshot-phase checkpoints.

bug (high): build() re-binds the default sort order and partition spec to the schema set here. If either references a column added after schema_id (e.g. Spark ALTER TABLE … WRITE ORDERED BY added, or partition evolution on a new column), build() fails with Cannot find source column for sort field. This breaks every path that pins an older schema: explicit snapshot_id/datetime, resuming from a pre-PR checkpoint, and resuming a follow checkpoint. Before this PR, scan().snapshot_id() read these fine. I reproduced it with this test, which fails on this branch:

#[test]
fn pin_snapshot_survives_sort_order_on_added_column() {
    use iceberg::spec::{NullOrder, SortDirection, SortField, Transform};
    let table = table_with_column_added_after_last_write();
    let _guard = TOKIO.enter();
    let order = SortOrder::builder().with_order_id(1)
        .with_sort_field(SortField { source_id: 2, transform: Transform::Identity,
            direction: SortDirection::Ascending, null_order: NullOrder::First })
        .build_unbound().unwrap();
    let metadata = TableMetadataBuilder::new_from_metadata(table.metadata().clone(), None)
        .add_sort_order(order).unwrap().set_default_sort_order(-1).unwrap()
        .build().unwrap().metadata;
    let table = Table::builder().runtime(Runtime::try_current().unwrap()).metadata(metadata)
        .identifier(table.identifier().clone()).file_io(table.file_io().clone())
        .build().unwrap();
    pin_snapshot(&table, 1, Some(0)).unwrap(); // Err: Cannot find source column for sort field
}

Reads don't need the write order or default spec, so you could reset them to unsorted/unpartitioned before calling build(). Please add a partition-spec variant of the test too.

bug (medium): An older host 0 scanned with scan().snapshot_id(id), so it used the snapshot's own schema. This path leaves pinned_schema_id as None, and pin_snapshot(.., None) then reads with the current schema. In a mixed-version rollout, the hosts would split one snapshot across two schemas, which is the problem this PR is meant to prevent. The fallback should use snapshot_by_id(id).schema_id(), as the old-checkpoint path at resolved_snapshot_id already does. The test comment at line 3723 ("the current schema is what it read") is also wrong.

pitfall (medium): Follow checkpoints now carry the startup schema id forward with each later snapshot id. On restart, prepare_snapshot_query pins (N, startup_schema), so used_columns never includes a column added while following. Before this PR, a restart picked up snapshot N's schema. If that's intended, please document it. Otherwise, follow boundaries should record the schema current at N. There's no test for restarting a follow-mode pipeline after a schema change.

bug (high): Sort order 0 isn't guaranteed to exist. Java and pyiceberg give a table created with a sort order id 1 and store only that order, and so does TableMetadataBuilder::new. On such a table this returns Cannot set default sort order to unknown order with id: '0'. Latest now always pins current_schema_id(), so every snapshot-mode start on a sorted-at-creation table fails, and that worked before this PR. I reproduced it locally with the test below, which fails on this head. Suggested fix: skip the whole block when schema_id == metadata.current_schema_id(). When the schema really changes, add SortOrder::unsorted_order() with add_sort_order before switching to it, instead of assuming id 0.

#[test]
fn qa_pin_snapshot_table_created_sorted() {
    // TableMetadataBuilder::new(schema, unpartitioned, SortOrder{order_id:1, field on id 1}, ...)
    // + add_snapshot(1); assert!(sort_order_by_id(0).is_none());
    pin_snapshot(&table, 1, Some(table.metadata().current_schema_id())).unwrap();
    // => DataInvalid: Cannot set default sort order to unknown order with id: '0'
}

bug (medium): The default partition spec has the same problem as the sort order. build() binds the default spec to the new current schema, so it fails after partition evolution onto a column added after the snapshot (ADD COLUMN region; ADD PARTITION FIELD region, no write since). That breaks snapshot_id/datetime time travel, resuming from a pre-PR checkpoint and the bare-choice fallback with Cannot find partition source field with id 2 in schema. Reads never use the default spec, so it can be swapped for an unpartitioned spec the same way. Repro: qa_pin_snapshot_partition_on_added_column (add the spec with add_default_partition_spec, then pin_snapshot(&table, 1, Some(0))).

nit: This comment contradicts the code. resolved_snapshot_id gives a bare choice the snapshot's schema (see a_bare_choice_from_an_older_host_reads_the_snapshot_schema). Here pinned_schema_id is None only because chosen_snapshot_id leaves it for the caller to fill in. Likewise in share.rs:656, "only pinning an older schema fails" no longer matches: both calls succeed.

bug (high): distributed snapshot mode now fails when the pin is current. provider.schema() is now the current schema, and it is handed to IcebergShareTableProvider below. IcebergShareScan::read still calls table.scan().snapshot_id(id), and with the bumped iceberg-rust that resolves column names against the snapshot's schema. So any renamed or added column in used_columns fails at build(). I confirmed this with a unit test in share.rs: the snapshot schema has id, the current schema renames field 1 to renamed, the current snapshot is pinned, and select(["renamed"]) returns:

DataInvalid => Column renamed not found in table

Before this PR, the same table gave a NULL column. Now the multihost pipeline fails. The share scan needs the same pin_is_current decision. One option is to pass a "read current" flag, or skip .snapshot_id() when it equals the host's current_snapshot_id(). Take care if another host's copy of the table has moved on. A distributed variant of rest_snapshot_pin_reads_current_schema would cover this.

pitfall (medium): snapshot_id comes from resolved_snapshot_id, which prefers the checkpointed id on resume. If the table gets any new commit before a restart, pin != current and the connector falls back to the old snapshot schema. The same pipeline then reads s before the restart and NULL after it (or fails to parse if s is NOT NULL). The same applies to snapshot_and_follow after it switches to following: used_columns is computed from the old schema here, and follow_loop_inner then maps those names against the current schema in project_field_ids, so the renamed column is silently dropped from every followed snapshot. Could used_columns always come from table.metadata().current_schema(), mapped by field id, with only the DataFusion provider pinned? If not, please add a resume test (pin → rename → append → restart) so the behaviour is at least pinned down.

bug (low-medium, docs): "A single-host connector reads the table's current schema" is only true when the pinned snapshot is the current one. An explicit older snapshot_id, a datetime that resolves to an older snapshot, or a resume after the table has advanced all still read the snapshot's schema and lose renamed or added columns. The changelog entry claims the same. Please state the condition.

bug (medium, docs): "reads NULL" only holds for nullable SQL columns. For a NOT NULL column the connector fails to parse instead. Your own rest_resume_past_the_pin_reads_the_snapshot_schema asserts missing field \s``, and the changelog entry says so too. Please say the same here and at line 334.

pitfall (medium): The current schema can also drop or retype a column, not just add or rename one. Example: a column is dropped and no snapshot is committed afterwards. Before this change, the pinned-current read still returned that column. Now used_columns leaves it out, so it reads NULL, or fails to parse if it is NOT NULL. That is probably the right semantics for "read the table as it is", but it changes behaviour in a way the changelog does not mention, and no test covers it. Please mention dropped columns in the changelog. Ideally also add a create-dropped fixture op next to create-renamed.

bug (medium): whether this is true depends on whether the table has moved since the pipeline started, so one logical read can switch schemas partway through. Example: snapshot mode with a nullable renamed column. The pipeline suspends in the middle of the initial snapshot, so the checkpoint is snapshot_progress(X, ts). Someone appends to the table, and the pipeline resumes. The first half of the snapshot used the current schema and got values. The rest falls back to X's schema, so the column reads NULL. For a dropped column it goes the other way: the stale values come back after resume. Before this PR the read was consistently wrong; now the output is silently inconsistent, which is harder to spot. One option is to record in the resume info that the read uses the current schema (or record the schema id) and reuse that on resume, rather than re-checking it against current_snapshot_id(). Whichever way you go, a test for resuming mid-snapshot with a nullable renamed column would cover it. rest_resume_past_the_pin_reads_the_snapshot_schema only covers the NOT NULL / post-EOI case, where both runs fail loudly.