smaheshwar-pltr commented on code in PR #3512:
URL: https://github.com/apache/iceberg-python/pull/3512#discussion_r3463011136
##########
pyiceberg/table/__init__.py:
##########
@@ -2263,6 +2341,180 @@ 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:
+ 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_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
+ to_snapshot_id_inclusive: 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_exclusive: int | None = None,
+ to_snapshot_id_inclusive: 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_exclusive = from_snapshot_id_exclusive
Review Comment:
Thanks for bringing this up - I think I need to think more about the API
shape here now and in the future
(https://github.com/apache/iceberg-python/pull/3364#issuecomment-4700045818).
Thinking out loud, a couple of examples:
```py
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
```py
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).
##########
pyiceberg/table/__init__.py:
##########
@@ -2263,6 +2341,180 @@ 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:
+ 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_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
+ to_snapshot_id_inclusive: 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_exclusive: int | None = None,
+ to_snapshot_id_inclusive: 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_exclusive = from_snapshot_id_exclusive
Review Comment:
Thanks for bringing this up - I realised I need to think more about the API
shape here now and in the future
(https://github.com/apache/iceberg-python/pull/3364#issuecomment-4700045818).
Thinking out loud, a couple of examples:
```py
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
```py
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).
--
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]
