kevinjqliu commented on code in PR #3512:
URL: https://github.com/apache/iceberg-python/pull/3512#discussion_r3488385648
##########
dev/provision.py:
##########
Review Comment:
if there are integration tests from java repo, would be good to follow up
and replicate the different scenarios to see if the results line up when
queried through pyiceberg's api vs spark's
##########
pyiceberg/table/__init__.py:
##########
@@ -2263,6 +2351,194 @@ 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:
+ ID of the snapshot to start the incremental scan from. If None,
the scan starts from
+ the oldest ancestor of the "to" snapshot (inclusive).
+ from_snapshot_inclusive:
+ Whether from_snapshot_id is included in the scan. If False, the
start snapshot is
+ exclusive.
+ to_snapshot_id:
+ 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
+ selected_fields:
+ A tuple of strings representing the column names
+ to return in the output dataframe.
+ case_sensitive:
+ If True column matching is case sensitive
+ options:
+ Additional Table properties as a dictionary of
+ string key value pairs to use for this scan.
+ limit:
+ An integer representing the number of rows to
+ return in the scan result. If None, fetches all
+ matching rows.
+ """
+
+ from_snapshot_id: int | None
+ from_snapshot_inclusive: bool
+ to_snapshot_id: int | None
+
+ def __init__(
+ self,
+ table_metadata: TableMetadata,
+ io: FileIO,
+ row_filter: str | BooleanExpression = ALWAYS_TRUE,
+ selected_fields: tuple[str, ...] = ("*",),
+ case_sensitive: bool = True,
+ from_snapshot_id: int | None = None,
+ from_snapshot_inclusive: bool = False,
+ to_snapshot_id: int | None = None,
+ options: Properties = EMPTY_DICT,
+ limit: int | None = None,
+ ):
+ super().__init__(
+ table_metadata=table_metadata,
+ io=io,
+ row_filter=row_filter,
+ selected_fields=selected_fields,
+ case_sensitive=case_sensitive,
+ options=options,
+ limit=limit,
+ )
+ self.from_snapshot_id = from_snapshot_id
+ self.from_snapshot_inclusive = from_snapshot_inclusive
+ self.to_snapshot_id = to_snapshot_id
+
+ def from_snapshot_id_exclusive(self: IAS, from_snapshot_id: int) -> IAS:
+ """Return a copy of this scan that starts (exclusively) from the given
snapshot ID."""
+ return self.update(from_snapshot_id=from_snapshot_id,
from_snapshot_inclusive=False)
+
+ def from_snapshot_id_inclusive(self: IAS, from_snapshot_id: int) -> IAS:
+ """Return a copy of this scan that starts (inclusively) from the given
snapshot ID."""
+ return self.update(from_snapshot_id=from_snapshot_id,
from_snapshot_inclusive=True)
+
+ def to_snapshot_id_inclusive(self: IAS, to_snapshot_id: int) -> IAS:
+ """Return a copy of this scan that ends (inclusively) at the given
snapshot ID."""
+ return self.update(to_snapshot_id=to_snapshot_id)
+
+ def projection(self) -> Schema:
+ current_schema = self.table_metadata.schema()
+ if "*" in self.selected_fields:
+ return current_schema
+ return current_schema.select(*self.selected_fields,
case_sensitive=self.case_sensitive)
+
+ def plan_files(self) -> Iterable[FileScanTask]:
+ """Plans the relevant files added between the specified snapshots."""
+ # With neither bound set, an empty table (no current snapshot) has
nothing to scan.
+ if self.from_snapshot_id is None and self.to_snapshot_id is None and
self.table_metadata.current_snapshot() is None:
+ return []
+
+ from_snapshot_id_exclusive, to_snapshot_id =
self._validate_and_resolve_snapshots()
+
+ append_snapshots = [
+ snapshot
+ for snapshot in ancestors_between_ids(
+ from_snapshot_id_exclusive=from_snapshot_id_exclusive,
+ to_snapshot_id_inclusive=to_snapshot_id,
+ table_metadata=self.table_metadata,
+ )
+ if snapshot.summary is not None and snapshot.summary.operation ==
Operation.APPEND
+ ]
+ if len(append_snapshots) == 0:
+ return []
+
+ append_snapshot_ids = {snapshot.snapshot_id for snapshot in
append_snapshots}
+
+ manifests = list(
+ {
+ manifest_file
+ for snapshot in append_snapshots
+ for manifest_file in snapshot.manifests(self.io)
+ if manifest_file.content == ManifestContent.DATA and
manifest_file.added_snapshot_id in append_snapshot_ids
+ }
+ )
+
+ return ManifestGroupPlanner(
+ table_metadata=self.table_metadata,
+ io=self.io,
+ row_filter=self.row_filter,
+ case_sensitive=self.case_sensitive,
+ options=self.options,
+ ).plan_files(
+ manifests=manifests,
+ manifest_entry_filter=lambda manifest_entry:
manifest_entry.snapshot_id in append_snapshot_ids
+ and manifest_entry.status == ManifestEntryStatus.ADDED,
+ )
+
+ def to_arrow(self) -> pa.Table:
+ """Read an Arrow table eagerly from this IncrementalAppendScan.
+
+ All rows will be loaded into memory at once.
Review Comment:
we should call out that the data is only whats appended, and may not
represent the full table.
something like:
```
"""Read the rows added by append snapshots in this scan's range into an
Arrow table.
All matching rows are loaded into memory at once. The scan reads only
files added by append snapshots in the resolved range; it does not read
the full table and does not apply later delete, overwrite, or replace
snapshots as net changes.
```
we should do the same for `to_arrow_batch_reader`
##########
pyiceberg/table/__init__.py:
##########
@@ -2263,6 +2351,194 @@ 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:
+ ID of the snapshot to start the incremental scan from. If None,
the scan starts from
+ the oldest ancestor of the "to" snapshot (inclusive).
+ from_snapshot_inclusive:
+ Whether from_snapshot_id is included in the scan. If False, the
start snapshot is
+ exclusive.
+ to_snapshot_id:
+ 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
+ selected_fields:
+ A tuple of strings representing the column names
+ to return in the output dataframe.
+ case_sensitive:
+ If True column matching is case sensitive
+ options:
+ Additional Table properties as a dictionary of
+ string key value pairs to use for this scan.
+ limit:
+ An integer representing the number of rows to
+ return in the scan result. If None, fetches all
+ matching rows.
+ """
+
+ from_snapshot_id: int | None
+ from_snapshot_inclusive: bool
+ to_snapshot_id: int | None
+
+ def __init__(
+ self,
+ table_metadata: TableMetadata,
+ io: FileIO,
+ row_filter: str | BooleanExpression = ALWAYS_TRUE,
+ selected_fields: tuple[str, ...] = ("*",),
+ case_sensitive: bool = True,
+ from_snapshot_id: int | None = None,
+ from_snapshot_inclusive: bool = False,
Review Comment:
bikeshedding: should this be `from_snapshot_inclusive=False` or
`from_snapshot_exclusive=True`?
we should just follow what java is doing
##########
pyiceberg/table/snapshots.py:
##########
@@ -431,6 +431,55 @@ def ancestors_between(from_snapshot: Snapshot | None,
to_snapshot: Snapshot, tab
yield from ancestors_of(to_snapshot, table_metadata)
+def ancestors_between_ids(
Review Comment:
nit: could the docstring mention that non-null `from_snapshot_id_exclusive`
must already be validated as part of `to`’s lineage? Otherwise, if it is never
reached, this yields all ancestors of `to`.
`IncrementalAppendScan` already validates before calling this, but just in
case someone else calls this function later
##########
pyiceberg/table/__init__.py:
##########
@@ -2263,6 +2351,194 @@ 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:
+ ID of the snapshot to start the incremental scan from. If None,
the scan starts from
+ the oldest ancestor of the "to" snapshot (inclusive).
+ from_snapshot_inclusive:
+ Whether from_snapshot_id is included in the scan. If False, the
start snapshot is
+ exclusive.
+ to_snapshot_id:
+ 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
+ selected_fields:
+ A tuple of strings representing the column names
+ to return in the output dataframe.
+ case_sensitive:
+ If True column matching is case sensitive
+ options:
+ Additional Table properties as a dictionary of
+ string key value pairs to use for this scan.
+ limit:
+ An integer representing the number of rows to
+ return in the scan result. If None, fetches all
+ matching rows.
+ """
+
+ from_snapshot_id: int | None
+ from_snapshot_inclusive: bool
+ to_snapshot_id: int | None
+
+ def __init__(
+ self,
+ table_metadata: TableMetadata,
+ io: FileIO,
+ row_filter: str | BooleanExpression = ALWAYS_TRUE,
+ selected_fields: tuple[str, ...] = ("*",),
+ case_sensitive: bool = True,
+ from_snapshot_id: int | None = None,
+ from_snapshot_inclusive: bool = False,
+ to_snapshot_id: int | None = None,
+ options: Properties = EMPTY_DICT,
+ limit: int | None = None,
+ ):
+ super().__init__(
+ table_metadata=table_metadata,
+ io=io,
+ row_filter=row_filter,
+ selected_fields=selected_fields,
+ case_sensitive=case_sensitive,
+ options=options,
+ limit=limit,
+ )
+ self.from_snapshot_id = from_snapshot_id
+ self.from_snapshot_inclusive = from_snapshot_inclusive
+ self.to_snapshot_id = to_snapshot_id
+
+ def from_snapshot_id_exclusive(self: IAS, from_snapshot_id: int) -> IAS:
+ """Return a copy of this scan that starts (exclusively) from the given
snapshot ID."""
+ return self.update(from_snapshot_id=from_snapshot_id,
from_snapshot_inclusive=False)
+
+ def from_snapshot_id_inclusive(self: IAS, from_snapshot_id: int) -> IAS:
+ """Return a copy of this scan that starts (inclusively) from the given
snapshot ID."""
+ return self.update(from_snapshot_id=from_snapshot_id,
from_snapshot_inclusive=True)
+
+ def to_snapshot_id_inclusive(self: IAS, to_snapshot_id: int) -> IAS:
+ """Return a copy of this scan that ends (inclusively) at the given
snapshot ID."""
+ return self.update(to_snapshot_id=to_snapshot_id)
+
+ def projection(self) -> Schema:
+ current_schema = self.table_metadata.schema()
+ if "*" in self.selected_fields:
+ return current_schema
+ return current_schema.select(*self.selected_fields,
case_sensitive=self.case_sensitive)
+
+ def plan_files(self) -> Iterable[FileScanTask]:
+ """Plans the relevant files added between the specified snapshots."""
+ # With neither bound set, an empty table (no current snapshot) has
nothing to scan.
+ if self.from_snapshot_id is None and self.to_snapshot_id is None and
self.table_metadata.current_snapshot() is None:
+ return []
+
+ from_snapshot_id_exclusive, to_snapshot_id =
self._validate_and_resolve_snapshots()
+
+ append_snapshots = [
+ snapshot
+ for snapshot in ancestors_between_ids(
+ from_snapshot_id_exclusive=from_snapshot_id_exclusive,
+ to_snapshot_id_inclusive=to_snapshot_id,
+ table_metadata=self.table_metadata,
+ )
+ if snapshot.summary is not None and snapshot.summary.operation ==
Operation.APPEND
+ ]
+ if len(append_snapshots) == 0:
+ return []
+
+ append_snapshot_ids = {snapshot.snapshot_id for snapshot in
append_snapshots}
+
+ manifests = list(
+ {
+ manifest_file
+ for snapshot in append_snapshots
+ for manifest_file in snapshot.manifests(self.io)
+ if manifest_file.content == ManifestContent.DATA and
manifest_file.added_snapshot_id in append_snapshot_ids
+ }
+ )
+
+ return ManifestGroupPlanner(
+ table_metadata=self.table_metadata,
+ io=self.io,
+ row_filter=self.row_filter,
+ case_sensitive=self.case_sensitive,
+ options=self.options,
+ ).plan_files(
+ manifests=manifests,
+ manifest_entry_filter=lambda manifest_entry:
manifest_entry.snapshot_id in append_snapshot_ids
+ and manifest_entry.status == ManifestEntryStatus.ADDED,
+ )
+
+ def to_arrow(self) -> pa.Table:
+ """Read an Arrow table eagerly from this IncrementalAppendScan.
+
+ All rows will be loaded into memory at once.
+
+ Returns:
+ pa.Table: Materialized Arrow Table from the Iceberg table's
IncrementalAppendScan
+ """
+ return _to_arrow_via_file_scan_tasks(self, self.projection(),
self.plan_files())
+
+ def to_arrow_batch_reader(self) -> pa.RecordBatchReader:
+ """Return an Arrow RecordBatchReader from this IncrementalAppendScan.
+
+ For large results, using a RecordBatchReader requires less memory than
+ loading an Arrow Table for the same IncrementalAppendScan, because a
+ RecordBatch is read one at a time.
+
+ Returns:
+ pa.RecordBatchReader: Arrow RecordBatchReader from the Iceberg
table's IncrementalAppendScan
+ which can be used to read a stream of record batches one by
one.
+ """
+ return _to_arrow_batch_reader_via_file_scan_tasks(self,
self.projection(), self.plan_files())
+
+ def _validate_and_resolve_snapshots(self) -> tuple[int | None, int]:
+ """Resolve the configured range to ``(from_snapshot_id_exclusive,
to_snapshot_id_inclusive)``.
+
+ A ``None`` "from" means the scan starts from the oldest ancestor of
the end snapshot.
Review Comment:
nit: maybe call out that `from_snapshot_inclusive` is resolved in this helper
##########
pyiceberg/table/__init__.py:
##########
@@ -1262,6 +1269,61 @@ def scan(
table_identifier=self._identifier,
)
+ 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,
+ options: Properties = EMPTY_DICT,
+ limit: int | None = None,
+ ) -> IncrementalAppendScan:
+ """Fetch an IncrementalAppendScan based on the table's current
metadata.
+
+ The incremental append scan returns the rows added by append snapshots
in a snapshot
+ range that match the provided row_filter, projected onto the table's
current schema.
+
+ Args:
+ from_snapshot_id_exclusive:
+ Optional ID of the snapshot to start the incremental scan
from, exclusively. If not set, the scan
+ starts from the oldest ancestor of the end snapshot
(inclusive).
+ 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.
+ selected_fields:
+ A tuple of strings representing the column names
+ to return in the output dataframe.
+ case_sensitive:
+ If True column matching is case sensitive.
+ options:
+ Additional Table properties as a dictionary of
+ string key value pairs to use for this scan.
+ limit:
+ An integer representing the number of rows to
+ return in the scan result. If None, fetches all
+ matching rows.
+
+ Returns:
+ An IncrementalAppendScan based on the table's current metadata and
provided parameters.
+ """
+ return IncrementalAppendScan(
+ table_metadata=self.metadata,
+ io=self.io,
+ row_filter=row_filter,
+ selected_fields=selected_fields,
+ case_sensitive=case_sensitive,
+ from_snapshot_id=from_snapshot_id_exclusive,
+ from_snapshot_inclusive=False,
Review Comment:
nit: add an inline comment to call this out. by default `from_snapshot_id`
is exclusive
##########
pyiceberg/table/__init__.py:
##########
@@ -2263,6 +2351,194 @@ 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.
Review Comment:
think it'll help if we add a disclaimer like "Non-append snapshots
in the range are ignored, so this is not a net-changes or changelog scan."?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
