Feature: Incremental Append Scan

[smaheshwar-pltr]

Closes #2634.

Rationale for this change

Adds IncrementalAppendScan, which reads the data appended between two snapshots — the building block for incremental ingestion. Largely a revival of the work in #2235; see #2634 and the previous PRs for motivation.

Split out of #3364 at the reviewers' request, and builds on the now-merged BaseScan / ManifestGroupPlanner refactor (#3511), so this PR's diff is the append-scan feature alone.

The surface mirrors Iceberg-Java's engine-facing API (snapshot IDs, inclusive/exclusive start, optional start) rather than the narrower Spark read options, since PyIceberg is increasingly used by engines (e.g. Polars). See the API discussion on this PR.

References: https://github.com/apache/iceberg (Iceberg-Java and Spark) and apache/iceberg-cpp#590. Inline review-aid comments (prefixed [AI reviewer aid]) point at the relevant reference code.

API

Table.incremental_append_scan(...) returns an IncrementalAppendScan; StagedTable overrides it to raise, mirroring scan(). The scan reads the rows added by append snapshots in (from, to], projected onto the table's current schema; delete / overwrite / replace snapshots in the range (e.g. compaction) are ignored.

The range is set via the factory's Spark-style kwargs or the builder methods, each of which returns a refined copy (like select() / filter()):

table.incremental_append_scan(
    from_snapshot_id_exclusive=None,   # optional; defaults to the oldest ancestor of `to`
    to_snapshot_id_inclusive=None,     # optional; defaults to the current snapshot
    row_filter=..., selected_fields=..., case_sensitive=..., options=..., limit=...,
)

scan.from_snapshot_id_exclusive(id)    # or .from_snapshot_id_inclusive(id)
    .to_snapshot_id_inclusive(id)

The range is held as public attributes — from_snapshot_id + from_snapshot_inclusive + to_snapshot_id — a single start slot plus an inclusive flag, mirroring Java's TableScanContext and consistent with the other scans.

Changes

• Range resolution mirrors Java's BaseIncrementalScan: an unset start scans from the oldest ancestor of the end; an inclusive start resolves to its parent as the exclusive boundary; an exclusive start is validated with is_parent_ancestor_of, so an expired start cursor is accepted as long as the lineage still passes through it; the end defaults to the current snapshot; an empty table with no range set scans nothing.
• Planning walks the append-only ancestors in the range, dedups the data manifests whose added_snapshot_id is in range (set semantics via ManifestFile.__eq__ / __hash__), and filters manifest entries to ADDED-in-range via a new manifest_entry_filter on ManifestGroupPlanner.plan_files. Compacted (rewrite_data_files) output is therefore not picked up — no double counting.
• Projects onto the table's current schema (matching Java/C++), so rows written under an older schema in the range get NULL for newer columns.
• Adds snapshot helpers ancestors_between_ids, is_ancestor_of, and is_parent_ancestor_of.
• Arrow materialization (to_arrow / to_arrow_batch_reader) is shared with DataScan via small module-level helpers that take the projected schema explicitly, so BaseScan stays projection-free (per the Refactor: extract BaseScan and ManifestGroupPlanner #3511 review).

Out of scope (tracked follow-ups)

• Branch selection (use_branch) and per-endpoint ref/tag start & end (from_ref_* / to_ref_*) — the rest of the engine-facing surface Java exposes.
• count(), REST server-side planning, and user-facing doc examples (mkdocs).
• dictionary_columns on IncrementalAppendScan.to_arrow / to_arrow_batch_reader (added to DataScan in feat: add dictionary_columns to to_arrow() / to_arrow_batch_reader() for memory-efficient reads #3461; the shared helpers already thread it) — kept out to isolate this PR.

Are these changes tested?

Yes — unit tests (range resolution including unset / inclusive / exclusive and expired start, current-schema projection, builder and update() copies, empty table, staged-table guard) and integration tests (append-only, non-append snapshots ignored, compaction not double-counted, schema evolution within range, partition- and metrics-evaluator pruning, disconnected snapshots), plus the test_incremental_read provision fixture.

Are there any user-facing changes?

Yes — the new Table.incremental_append_scan(...) API and IncrementalAppendScan class. No changes to existing public surface.

[smaheshwar-pltr]

[AI reviewer aid] New convenience method mirroring Table.scan (naming thought). Args mirror scan minus snapshot_id plus the two snapshot-range args.

[smaheshwar-pltr]

[AI reviewer aid] Mirrors StagedTable.scan two lines up — staged tables have no committed metadata to scan against.

[smaheshwar-pltr]

[AI reviewer aid] Mirrors Java's IncrementalAppendScan interface and BaseIncrementalAppendScan implementation. Only the append variant of IncrementalScan — changelog scan is out of scope here.

[smaheshwar-pltr]

[AI reviewer aid] Always uses the table's current schema, unlike TableScan.projection() which uses the snapshot's schema when snapshot_id is set. Matches Java: BaseTable.newIncrementalAppendScan constructs the scan with schema(), which on BaseTable.schema() returns ops.current().schema() — the table's current schema, not snapshot-bound. C++ does the same: TableScanBuilder::ResolveSnapshotSchema falls through to metadata_->Schema() for incremental scans (no snapshot_id on the context). Older-schema rows in range get NULL for new columns — covered by test_incremental_append_scan_schema_evolution_within_range.

[smaheshwar-pltr]

[AI reviewer aid] Mirrors Java's BaseIncrementalAppendScan.doPlanFiles and appendFilesFromSnapshots — walk ancestors, filter to append snapshots, dedup manifests whose added_snapshot_id is in range, then filter manifest entries by (snapshot_id in range, status == ADDED). Set semantics on the manifest dedup match the Java snippet and rely on ManifestFile.__eq__/__hash__ being defined (which they are on main since #2233).

[smaheshwar-pltr]

[AI reviewer aid] This manifest filter is new. Introducing that for append scan logic where some manifests are skipped

[smaheshwar-pltr]

[AI reviewer aid] Mirrors Java's SnapshotUtil.ancestorsBetween. Differs from the existing ancestors_between (snapshot-based, inclusive-inclusive) above by taking IDs and being exclusive-inclusive, to match the incremental-scan validation pattern. Raises if to_snapshot_id_inclusive is missing from metadata, mirroring Java.

[smaheshwar-pltr]

[AI reviewer aid] Mirrors Java's SnapshotUtil.isParentAncestorOf, including the Cannot find snapshot raise on missing snapshot (Java throws one hop down, via ancestorsOf(long, lookup)).

[smaheshwar-pltr]

[AI reviewer aid] Filters on a non-partition column (number), so the manifest and partition evaluators degenerate to ALWAYS_TRUE and it's the per-file metrics evaluator (column min/max/null stats) that must do all the pruning. Covers a layer of ManifestGroupPlanner that the existing DataScan integration coverage doesn't exercise end-to-end through a real scan.

[kevinjqliu]

#3511 is now merged!

[smaheshwar-pltr]

Capturing @kevinjqliu's follow-ups: #3364 (comment).

I'm happy to follow-up with the docs PR for this right after - my thinking was to keep this PR isolated given it's large enough as is 🙌.

One more: dictionary_columns was added to DataScan's Arrow reads in #3461. I'll extend it to IncrementalAppendScan.to_arrow / to_arrow_batch_reader (with a dict-encoding test) in a follow-up too, to keep this PR as isolated as possible. (Edit: #3549)

[smaheshwar-pltr]

[AI reviewer aid] DataScan.to_arrow / to_arrow_batch_reader take a dictionary_columns argument (added in #3461), and the shared _to_arrow*_via_file_scan_tasks helpers already thread it. I've deliberately not exposed it on IncrementalAppendScan here, to keep this PR as isolated as possible. Follow-up will add dictionary_columns to the incremental scan's Arrow reads, with a dict-encoding test mirroring #3461's test_dictionary_columns_produces_dict_encoded_output.

[smaheshwar-pltr]

#3549 - the follow-up

[rambleraptor]

I want to make sure that we're supporting compaction properly. That seems to be the biggest missing piece we're missing

Can you write a test where we add some files, compact them, and then ensure that the scan doesn't pick up the newly compacted file (for a double count)?

[rambleraptor]

A couple nits and a question around adding a new test case. Thanks a lot!

[rambleraptor]

What do you think about changing this to update_from_snapshot_id_exclusive? (same for inclusive below)

The variables are public, so we have both a variable and a method and it can be unclear which to use.

[rambleraptor]

We should make this a private variable and then expose it out with a getter.

The methods below are trying to make it explicit how to change these variables, so we should make it impossible to change them directly.

[smaheshwar-pltr]

Thanks for bringing this up - I realised I need to think more about the API shape here now and in the future (#3364 (comment)).

Thinking out loud, a couple of examples:

scan = test_table.incremental_append_scan(
  selected_fields=("number",),
  # This is *required*, diverging from Java but mirroring Spark
  from_snapshot_id_exclusive=test_table.snapshots()[0].snapshot_id,
  to_snapshot_id_inclusive=test_table.snapshots()[2].snapshot_id,
)
# No setters on the `IncrementalAppendScan` - similar to `snapshot_id` with normal scans.
# Matches Spark surface, that only supports `from` and `to` snapshots being set, and nothing else
# https://iceberg.apache.org/docs/latest/spark-queries/#incremental-read

vs

scan = (
  test_table.incremental_append_scan(
    selected_fields=("number",)
    # No snapshot IDs here maybe? But that differs from normal `.scan()` surface where snapshot_id is present there but not on the `DataScan`
  )
  .from_snapshot_id_exclusive(...)  # *OR* inclusive *OR* omitted - diverging from Spark, doesn't matter because this is internal to the scan
  .to_snapshot_id_inclusive(...)
  .use_ref(...)  # Support more beyond just snapshot IDs, permitting bad combinations at API layer (they would throw at runtime instead)
)
# Matches Java surface that supports far more than Spark - but is engine facing (not user-facing)

I previously argued for matching Spark's semantics but I'm now thinking that we should keep the API surface here more open, similar to Java - which shouldn't be too hard to do. Let me have a proper think here (but happy if folks want to chime in with opinions).

[smaheshwar-pltr]

I've now moved to the following setup. Like the normal data scan, IncrementalAppendScan is a plain value object: its constructor parameters are exactly its public attributes — the invariant the inherited BaseScan.update() relies on to return refined copies, so the builders below are thin update() wrappers.

class IncrementalAppendScan(BaseScan):
    # range state (new) — a single `from` slot + an inclusive flag, mirroring java/cpp:
    from_snapshot_id: int | None
    from_snapshot_inclusive: bool          # False = exclusive (the default)
    to_snapshot_id: int | None
    branch: str | None
    # + inherited BaseScan config: row_filter, selected_fields, case_sensitive, options, limit

The factory's Spark-style from_snapshot_id_exclusive / to_snapshot_id_inclusive kwargs map
onto those attributes (from_snapshot_id + from_snapshot_inclusive):

# Factory (front door) — mirrors `table.scan(...)`; both range bounds optional.
table.incremental_append_scan(
    from_snapshot_id_exclusive: int | None = None,   # start, exclusive; default = oldest ancestor of `to`
    to_snapshot_id_inclusive: int | None = None,     # end, inclusive;   default = current snapshot
    row_filter: str | BooleanExpression = ALWAYS_TRUE,
    selected_fields: tuple[str, ...] = ("*",),
    case_sensitive: bool = True,
    options: Properties = EMPTY_DICT,
    limit: int | None = None,
) -> IncrementalAppendScan

# Builders (power surface) — each returns a refined copy, like select()/filter().
IncrementalAppendScan.from_snapshot_id_exclusive(id) / .from_snapshot_id_inclusive(id)
                     .to_snapshot_id_inclusive(id)
                     .use_branch(name)
                     # .from_ref_* / to_ref_* — start/end by ref, follow-up
                     # + inherited select() / filter() / with_case_sensitive() / update()

Choosing from_snapshot_id_exclusive / to_snapshot_id_inclusive instead of set/update_snapshot_id_exclusive matches the existing flow of filter() / select() naming for normal data scans and makes the chaining above read more smoothly, IMO.

(Note: I've left out branch / use_branch to be added as a follow-up together with appropriate tests, because that makes sense to me as a splitting point here)

[smaheshwar-pltr]

[AI reviewer aid] An unset start scans from the oldest ancestor of the end snapshot, matching Java's BaseIncrementalScan (a null exclusive boundary; ancestors_between_ids(None, to) then walks the whole lineage, cf. oldestAncestorOf). Earlier revisions required the start (Spark's start-snapshot-id); opened up to match Java's engine-facing surface.

[smaheshwar-pltr]

[AI reviewer aid] An inclusive start resolves to its parent as the exclusive boundary (None at the root) and must be present and an ancestor of the end. Mirrors Java's BaseIncrementalScan.fromSnapshotIdExclusive (isAncestorOf check, then parentId()).

[smaheshwar-pltr]

[AI reviewer aid] An exclusive start is validated with is_parent_ancestor_of (not is_ancestor_of), so an expired start cursor is accepted as long as the lineage still passes through it; equal from/to is rejected (a snapshot is never its own parent-ancestor). Mirrors Java's BaseIncrementalScan (see the expiry note there).

[smaheshwar-pltr]

[AI reviewer aid] Builders map to Java's overloaded fromSnapshotExclusive(long) / fromSnapshotInclusive(long). Python has no overloading, so the variants get distinct names; ref-based start (from_ref_*) is a follow-up. Each returns a refined copy via BaseScan.update(), like select() / filter().

[smaheshwar-pltr]

[AI reviewer aid] Mirrors Java's SnapshotUtil.isAncestorOf (a snapshot is its own ancestor). Validates an inclusive start; sibling of is_parent_ancestor_of below.

[smaheshwar-pltr]

[AI reviewer aid] Compaction (rewrite_data_files, snapshots[3]) commits a replace snapshot whose rewritten file is added by that snapshot, not by an append. The append-only filter drops the replace snapshot and the manifest added_snapshot_id / status == ADDED entry filter drops its file, so the appended rows are read exactly once — never double-counted with the rewritten file.

[smaheshwar-pltr]

Addressing #3512 (review)