Require from_snapshot_id_exclusive and drop range builders on IncrementalAppendScan

Address review feedback: remove the from_snapshot_exclusive/to_snapshot_inclusive
fluent builders (which clashed with the public from/to attributes and left them
mutable) in favour of setting the snapshot range at construction, mirroring
DataScan.snapshot_id and Spark's required start-snapshot-id option surface.

- from_snapshot_id_exclusive is now a required, keyword-only argument; the
  deferred plan-time 'start snapshot not set' error is gone (fail-fast).
- Range-first, keyword-only signatures on Table.incremental_append_scan and
  IncrementalAppendScan.__init__.
- StagedTable.incremental_append_scan keeps from optional so a no-arg call
  raises the staged-table ValueError, not a TypeError.
- Simplify user-facing docstrings to Spark's option surface (the expired-cursor
  behaviour stays covered by tests and internal comments).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: smaheshwar-pltr

pyiceberg/table/__init__.py

@@ -1264,11 +1264,12 @@ def scan(
 
     def incremental_append_scan(
         self,
+        *,
+        from_snapshot_id_exclusive: int,
+        to_snapshot_id_inclusive: int | None = None,
         row_filter: str | BooleanExpression = ALWAYS_TRUE,
         selected_fields: tuple[str, ...] = ("*",),
         case_sensitive: bool = True,
-        from_snapshot_id_exclusive: int | None = None,
-        to_snapshot_id_inclusive: int | None = None,
         options: Properties = EMPTY_DICT,
         limit: int | None = None,
     ) -> IncrementalAppendScan:
@@ -1278,6 +1279,11 @@ def incremental_append_scan(
         range that match the provided row_filter, projected onto the table's current schema.
 
         Args:
+            from_snapshot_id_exclusive:
+                ID of the snapshot to start the incremental scan from, exclusively.
+            to_snapshot_id_inclusive:
+                Optional ID of the snapshot to end the incremental scan at, inclusively. If not set, it defaults to
+                the table's current snapshot.
             row_filter:
                 A string or BooleanExpression that describes the
                 desired rows.
@@ -1286,12 +1292,6 @@ def incremental_append_scan(
                 to return in the output dataframe.
             case_sensitive:
                 If True column matching is case sensitive.
-            from_snapshot_id_exclusive:
-                Optional ID of the "from" snapshot, to start the incremental scan from, exclusively. This can be set
-                on the IncrementalAppendScan object returned, but ultimately must not be None.
-            to_snapshot_id_inclusive:
-                Optional ID of the "to" snapshot, to end the incremental scan at, inclusively. This can be set on the
-                IncrementalAppendScan object returned. If not set, it defaults to the table's current snapshot.
             options:
                 Additional Table properties as a dictionary of
                 string key value pairs to use for this scan.
@@ -1833,11 +1833,12 @@ def scan(
 
     def incremental_append_scan(
         self,
+        *,
+        from_snapshot_id_exclusive: int | None = None,
+        to_snapshot_id_inclusive: int | None = None,
         row_filter: str | BooleanExpression = ALWAYS_TRUE,
         selected_fields: tuple[str, ...] = ("*",),
         case_sensitive: bool = True,
-        from_snapshot_id_exclusive: int | None = None,
-        to_snapshot_id_inclusive: int | None = None,
         options: Properties = EMPTY_DICT,
         limit: int | None = None,
     ) -> IncrementalAppendScan:
@@ -2341,13 +2342,15 @@ def count(self) -> int:
         return res
 
 
-IAS = TypeVar("IAS", bound="IncrementalAppendScan", covariant=True)
-
-
 class IncrementalAppendScan(BaseScan):
     """An incremental scan of a table's data that accumulates appended data between two snapshots.
 
     Args:
+        from_snapshot_id_exclusive:
+            ID of the snapshot to start the incremental scan from, exclusively.
+        to_snapshot_id_inclusive:
+            Optional ID of the snapshot to end the incremental scan at, inclusively.
+            Omitting it will default to the table's current snapshot.
         row_filter:
             A string or BooleanExpression that describes the
             desired rows
@@ -2363,27 +2366,21 @@ class IncrementalAppendScan(BaseScan):
             An integer representing the number of rows to
             return in the scan result. If None, fetches all
             matching rows.
-        from_snapshot_id_exclusive:
-            Optional ID of the "from" snapshot, to start the incremental scan from, exclusively. When the scan is
-            ultimately planned, this must not be None. The snapshot does not need to be present in the table metadata
-            (it may have been expired), as long as it is the parent of some ancestor of the "to" snapshot.
-        to_snapshot_id_inclusive:
-            Optional ID of the "to" snapshot, to end the incremental scan at, inclusively.
-            Omitting it will default to the table's current snapshot.
     """
 
-    from_snapshot_id_exclusive: int | None
+    from_snapshot_id_exclusive: int
     to_snapshot_id_inclusive: int | None
 
     def __init__(
         self,
         table_metadata: TableMetadata,
         io: FileIO,
+        *,
+        from_snapshot_id_exclusive: int,
+        to_snapshot_id_inclusive: int | None = None,
         row_filter: str | BooleanExpression = ALWAYS_TRUE,
         selected_fields: tuple[str, ...] = ("*",),
         case_sensitive: bool = True,
-        from_snapshot_id_exclusive: int | None = None,
-        to_snapshot_id_inclusive: int | None = None,
         options: Properties = EMPTY_DICT,
         limit: int | None = None,
     ):
@@ -2399,28 +2396,6 @@ def __init__(
         self.from_snapshot_id_exclusive = from_snapshot_id_exclusive
         self.to_snapshot_id_inclusive = to_snapshot_id_inclusive
 
-    def from_snapshot_exclusive(self: IAS, from_snapshot_id_exclusive: int | None) -> IAS:
-        """Instructs this scan to look for changes starting from a particular snapshot (exclusive).
-
-        Args:
-            from_snapshot_id_exclusive: the start snapshot ID (exclusive)
-
-        Returns:
-            this for method chaining
-        """
-        return self.update(from_snapshot_id_exclusive=from_snapshot_id_exclusive)
-
-    def to_snapshot_inclusive(self: IAS, to_snapshot_id_inclusive: int | None) -> IAS:
-        """Instructs this scan to look for changes up to a particular snapshot (inclusive).
-
-        Args:
-            to_snapshot_id_inclusive: the end snapshot ID (inclusive)
-
-        Returns:
-            this for method chaining
-        """
-        return self.update(to_snapshot_id_inclusive=to_snapshot_id_inclusive)
-
     def projection(self) -> Schema:
         current_schema = self.table_metadata.schema()
         if "*" in self.selected_fields:
@@ -2490,9 +2465,6 @@ def to_arrow_batch_reader(self) -> pa.RecordBatchReader:
         return _to_arrow_batch_reader_via_file_scan_tasks(self, self.projection(), self.plan_files())
 
     def _validate_and_resolve_snapshots(self) -> tuple[int, int]:
-        if self.from_snapshot_id_exclusive is None:
-            raise ValueError("Start snapshot is not set, please set from_snapshot_id_exclusive")
-
         if self.to_snapshot_id_inclusive is None:
             current_snapshot = self.table_metadata.current_snapshot()
             if current_snapshot is None:

tests/integration/test_reads.py

@@ -1279,10 +1279,9 @@ def test_scan_source_field_missing_in_spec(catalog: Catalog, spark: SparkSession
 def test_incremental_append_scan_append_only(catalog: Catalog) -> None:
     test_table = catalog.load_table("default.test_incremental_read")
 
-    scan = (
-        test_table.incremental_append_scan()
-        .from_snapshot_exclusive(test_table.snapshots()[0].snapshot_id)
-        .to_snapshot_inclusive(test_table.snapshots()[2].snapshot_id)
+    scan = test_table.incremental_append_scan(
+        from_snapshot_id_exclusive=test_table.snapshots()[0].snapshot_id,
+        to_snapshot_id_inclusive=test_table.snapshots()[2].snapshot_id,
     )
 
     # snapshots[1] adds 1 file (letter=b); snapshots[2] adds 2 files (letter=b, letter=c).
@@ -1334,10 +1333,9 @@ def test_incremental_append_scan_schema_evolution_within_range(catalog: Catalog)
     # snapshots[1..2] are on the original schema (number, letter); snapshots[4] is on the evolved
     # schema (number, letter, extra) after ALTER TABLE ADD COLUMN. The scan must project the older
     # rows onto the current schema (extra -> null) and pick up the new value for the newer row.
-    scan = (
-        test_table.incremental_append_scan()
-        .from_snapshot_exclusive(test_table.snapshots()[0].snapshot_id)
-        .to_snapshot_inclusive(test_table.snapshots()[4].snapshot_id)
+    scan = test_table.incremental_append_scan(
+        from_snapshot_id_exclusive=test_table.snapshots()[0].snapshot_id,
+        to_snapshot_id_inclusive=test_table.snapshots()[4].snapshot_id,
     )
     assert len(list(scan.plan_files())) == 4
 
@@ -1361,10 +1359,10 @@ def test_incremental_append_scan_partition_pruning(catalog: Catalog) -> None:
     # `letter=c` only appears in snapshots[2]. The manifest evaluator rejects snapshots[1]'s
     # manifest (letter=b only); the partition evaluator rejects the letter=b entry in
     # snapshots[2]'s manifest. One file remains.
-    scan = (
-        test_table.incremental_append_scan(row_filter=EqualTo("letter", "c"))
-        .from_snapshot_exclusive(test_table.snapshots()[0].snapshot_id)
-        .to_snapshot_inclusive(test_table.snapshots()[2].snapshot_id)
+    scan = test_table.incremental_append_scan(
+        row_filter=EqualTo("letter", "c"),
+        from_snapshot_id_exclusive=test_table.snapshots()[0].snapshot_id,
+        to_snapshot_id_inclusive=test_table.snapshots()[2].snapshot_id,
     )
     assert len(list(scan.plan_files())) == 1
     assert scan.to_arrow()["number"].to_pylist() == [3]
@@ -1377,10 +1375,10 @@ def test_incremental_append_scan_metrics_pruning(catalog: Catalog) -> None:
 
     # Non-partition predicate: the manifest/partition evaluators degenerate, leaving the per-file
     # metrics evaluator to prune. `number=99` matches no file's [min, max] stats for `number`.
-    scan = (
-        test_table.incremental_append_scan(row_filter=EqualTo("number", 99))
-        .from_snapshot_exclusive(test_table.snapshots()[0].snapshot_id)
-        .to_snapshot_inclusive(test_table.snapshots()[2].snapshot_id)
+    scan = test_table.incremental_append_scan(
+        row_filter=EqualTo("number", 99),
+        from_snapshot_id_exclusive=test_table.snapshots()[0].snapshot_id,
+        to_snapshot_id_inclusive=test_table.snapshots()[2].snapshot_id,
     )
     assert len(list(scan.plan_files())) == 0
     assert len(scan.to_arrow()) == 0
@@ -1391,10 +1389,10 @@ def test_incremental_append_scan_metrics_pruning(catalog: Catalog) -> None:
 def test_incremental_append_scan_selected_fields(catalog: Catalog) -> None:
     test_table = catalog.load_table("default.test_incremental_read")
 
-    scan = (
-        test_table.incremental_append_scan(selected_fields=("number",))
-        .from_snapshot_exclusive(test_table.snapshots()[0].snapshot_id)
-        .to_snapshot_inclusive(test_table.snapshots()[2].snapshot_id)
+    scan = test_table.incremental_append_scan(
+        selected_fields=("number",),
+        from_snapshot_id_exclusive=test_table.snapshots()[0].snapshot_id,
+        to_snapshot_id_inclusive=test_table.snapshots()[2].snapshot_id,
     )
     result_table = scan.to_arrow()
     assert result_table.schema.equals(pa.schema([pa.field("number", pa.int32())]))
@@ -1406,10 +1404,10 @@ def test_incremental_append_scan_selected_fields(catalog: Catalog) -> None:
 def test_incremental_append_scan_limit(catalog: Catalog) -> None:
     test_table = catalog.load_table("default.test_incremental_read")
 
-    scan = (
-        test_table.incremental_append_scan(limit=2)
-        .from_snapshot_exclusive(test_table.snapshots()[0].snapshot_id)
-        .to_snapshot_inclusive(test_table.snapshots()[2].snapshot_id)
+    scan = test_table.incremental_append_scan(
+        limit=2,
+        from_snapshot_id_exclusive=test_table.snapshots()[0].snapshot_id,
+        to_snapshot_id_inclusive=test_table.snapshots()[2].snapshot_id,
     )
     assert len(scan.to_arrow()) == 2
 

tests/table/test_init.py

@@ -356,20 +356,23 @@ def test_data_scan_plan_files_no_current_snapshot(example_table_metadata_no_snap
 
 
 def test_incremental_append_scan_default(table_v2: Table) -> None:
-    scan = table_v2.incremental_append_scan()
+    scan = table_v2.incremental_append_scan(from_snapshot_id_exclusive=3051729675574597004)
     assert scan.row_filter == AlwaysTrue()
     assert scan.selected_fields == ("*",)
     assert scan.case_sensitive is True
-    assert scan.from_snapshot_id_exclusive is None
+    assert scan.from_snapshot_id_exclusive == 3051729675574597004
     assert scan.to_snapshot_id_inclusive is None
 
 
 def test_incremental_append_scan_chaining(table_v2: Table) -> None:
     older_snapshot_id, newer_snapshot_id = 3051729675574597004, 3055729675574597004
+    # The snapshot range is set at construction; the snapshot-independent helpers still chain
+    # and must round-trip the range through `update()`.
     scan = (
-        table_v2.incremental_append_scan()
-        .from_snapshot_exclusive(older_snapshot_id)
-        .to_snapshot_inclusive(newer_snapshot_id)
+        table_v2.incremental_append_scan(
+            from_snapshot_id_exclusive=older_snapshot_id,
+            to_snapshot_id_inclusive=newer_snapshot_id,
+        )
         .filter(EqualTo("x", 1))
         .select("x", "y")
         .with_case_sensitive(False)
@@ -382,18 +385,19 @@ def test_incremental_append_scan_chaining(table_v2: Table) -> None:
 
 
 def test_incremental_append_scan_projection_uses_current_schema(table_v2: Table) -> None:
-    scan = table_v2.incremental_append_scan(selected_fields=("x", "y"))
+    scan = table_v2.incremental_append_scan(from_snapshot_id_exclusive=3051729675574597004, selected_fields=("x", "y"))
     projection_schema = scan.projection()
     assert projection_schema.schema_id == 1
     assert {f.name for f in projection_schema.fields} == {"x", "y"}
 
 
 def test_incremental_append_scan_requires_from_snapshot(table_v2: Table) -> None:
-    scan = table_v2.incremental_append_scan(
-        to_snapshot_id_inclusive=table_v2.current_snapshot().snapshot_id,  # type: ignore[union-attr]
-    )
-    with pytest.raises(ValueError, match="Start snapshot is not set"):
-        list(scan.plan_files())
+    # `from_snapshot_id_exclusive` is a required argument, so omitting it fails fast at construction
+    # rather than deferring the error to plan time.
+    with pytest.raises(TypeError, match="from_snapshot_id_exclusive"):
+        table_v2.incremental_append_scan(
+            to_snapshot_id_inclusive=table_v2.current_snapshot().snapshot_id,  # type: ignore[union-attr,call-arg]
+        )
 
 
 def test_incremental_append_scan_invalid_to_snapshot(table_v2: Table) -> None:
@@ -470,19 +474,20 @@ def test_incremental_append_scan_admits_expired_from_snapshot(example_table_meta
 
 def test_incremental_append_scan_update_preserves_type(table_v2: Table) -> None:
     # `update()` lives on BaseScan and uses `type(self)(...)` to construct the copy;
-    # verify the concrete type is preserved through both `update` and the chaining helpers.
-    base_scan = table_v2.incremental_append_scan()
+    # verify the concrete type is preserved through both `update` and the fluent helpers.
+    base_scan = table_v2.incremental_append_scan(from_snapshot_id_exclusive=123)
     base_type = type(base_scan)
     assert base_type.__name__ == "IncrementalAppendScan"
-    assert type(base_scan.from_snapshot_exclusive(123)) is base_type
-    assert type(base_scan.to_snapshot_inclusive(456)) is base_type
+    assert type(base_scan.update(from_snapshot_id_exclusive=456)) is base_type
+    assert type(base_scan.update(to_snapshot_id_inclusive=789)) is base_type
     assert type(base_scan.filter(EqualTo("x", 1))) is base_type
     assert type(base_scan.select("x")) is base_type
     assert type(base_scan.with_case_sensitive(False)) is base_type
 
-    # None round-trips through the chaining helpers (update() reconstructs by keyword).
-    assert base_scan.from_snapshot_exclusive(123).from_snapshot_exclusive(None).from_snapshot_id_exclusive is None
-    assert base_scan.to_snapshot_inclusive(456).to_snapshot_inclusive(None).to_snapshot_id_inclusive is None
+    # The snapshot range round-trips through update() (which reconstructs by keyword).
+    assert base_scan.update(from_snapshot_id_exclusive=456).from_snapshot_id_exclusive == 456
+    assert base_scan.update(to_snapshot_id_inclusive=789).to_snapshot_id_inclusive == 789
+    assert base_scan.update(to_snapshot_id_inclusive=789).update(to_snapshot_id_inclusive=None).to_snapshot_id_inclusive is None
 
 
 def test_incremental_append_scan_staged_table_raises(table_v2: Table) -> None:
@@ -494,6 +499,8 @@ def test_incremental_append_scan_staged_table_raises(table_v2: Table) -> None:
         io=table_v2.io,
         catalog=table_v2.catalog,
     )
+    # No-arg call must raise the staged-table ValueError, not a TypeError for the missing
+    # required `from_snapshot_id_exclusive` (the override keeps it optional, like StagedTable.scan).
     with pytest.raises(ValueError, match="Cannot scan a staged table"):
         staged_table.incremental_append_scan()