You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "tusharchou (via GitHub)" <gi...@apache.org> on 2024/11/28 18:19:00 UTC
[PR] Count rows as a metadata only operation [iceberg-python]
tusharchou opened a new pull request, #1388:
URL: https://github.com/apache/iceberg-python/pull/1388
closes issue: Count rows as a metadata-only operation #1223
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Count rows as a metadata only operation [iceberg-python]
Posted by "gli-chris-hao (via GitHub)" <gi...@apache.org>.
gli-chris-hao commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1900228680
##########
pyiceberg/table/__init__.py:
##########
@@ -1594,6 +1609,29 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ """
+ Usage: calutates the total number of records in a Scan that haven't had positional deletes
+ """
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the fiter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and not len(task.delete_files):
+ # Every File has a metadata stat that stores the file record count
+ res += task.file.record_count
+ else:
+ from pyiceberg.io.pyarrow import ArrowScan
+ tbl = ArrowScan(
+ self.table_metadata, self.io, self.projection(), self.row_filter, self.case_sensitive, self.limit
+ ).to_table([task])
+ res += len(tbl)
+ return res
Review Comment:
I love this approach! My only concern is about loading too much data into memory at once, although this is loading just one file at a time, in the worst case some file could potentially be very large? Shall we define a threshold and check, for example, if `file size < 512MB`, load entire file, otherwise turn it into `pa.RecordBatchReader` and read stream of record batches for counting.
```
target_schema = schema_to_pyarrow(self.projection())
batches = ArrowScan(
self.table_metadata, self.io, self.projection(), self.row_filter, self.case_sensitive, self.limit
).to_record_batches([task])
reader = pa.RecordBatchReader.from_batches(
target_schema,
batches,
)
count = 0
for batch in reader:
count += batch.num_rows
return count
```
https://github.com/apache/iceberg-python/blob/e6465001bd8a47718ff79da4def5800962e6b895/pyiceberg/table/__init__.py#L1541-L1564
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Count rows as a metadata only operation [iceberg-python]
Posted by "tusharchou (via GitHub)" <gi...@apache.org>.
tusharchou commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1863507446
##########
pyiceberg/table/__init__.py:
##########
@@ -1493,6 +1496,13 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ res = 0
+ tasks = self.plan_files()
Review Comment:
Hi @kevinjqliu thank you for the review. I am trying to account for positional deletes, do you have a suggestion on how that can be achieved?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "tusharchou (via GitHub)" <gi...@apache.org>.
tusharchou commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1936663741
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression):
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is not None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
Review Comment:
@Fokko I would require some help here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1944716694
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1732,230 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ """Finds the residuals for an Expression the partitions in the given PartitionSpec.
+
+ A residual expression is made by partially evaluating an expression using partition values.
+ For example, if a table is partitioned by day(utc_timestamp) and is read with a filter expression
+ utc_timestamp >= a and utc_timestamp <= b, then there are 4 possible residuals expressions
Review Comment:
```suggestion
utc_timestamp > a and utc_timestamp < b, then there are 4 possible residuals expressions
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1944733940
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1732,230 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ """Finds the residuals for an Expression the partitions in the given PartitionSpec.
+
+ A residual expression is made by partially evaluating an expression using partition values.
+ For example, if a table is partitioned by day(utc_timestamp) and is read with a filter expression
+ utc_timestamp >= a and utc_timestamp <= b, then there are 4 possible residuals expressions
+ for the partition data, d:
+
+
+ 1. If d > day(a) and d < day(b), the residual is always true
+ 2. If d == day(a) and d != day(b), the residual is utc_timestamp >= a
+ 3. if d == day(b) and d != day(a), the residual is utc_timestamp <= b
+ 4. If d == day(a) == day(b), the residual is utc_timestamp >= a and utc_timestamp <= b
+
+ Partition data is passed using StructLike. Residuals are returned by residualFor(StructLike).
+
+ This class is thread-safe.
+ """
+
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+ expr: BooleanExpression
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression) -> None:
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and not math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ def struct_to_schema(struct: StructType) -> Schema:
+ return Schema(*struct.fields)
+
+ for part in parts:
+ strict_projection = part.transform.strict_project(part.name, predicate)
+ strict_result = None
+
+ if strict_projection is not None:
+ bound = strict_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound, BoundPredicate):
+ strict_result = super().visit_bound_predicate(bound)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or alwaysFalse
+ strict_result = bound
+
+ if strict_result is not None and isinstance(strict_result, AlwaysTrue):
Review Comment:
In Python we can simplify this a bit:
```suggestion
if isinstance(strict_result, AlwaysTrue):
```
>>> from pyiceberg.expressions import AlwaysTrue
>>> isinstance(None, AlwaysTrue)
False
```python
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1944751408
##########
pyiceberg/table/__init__.py:
##########
@@ -1624,6 +1659,40 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ # Usage: Calculates the total number of records in a Scan that haven't had positional deletes.
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the filter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and len(task.delete_files) == 0:
+ # Every File has a metadata stat that stores the file record count
+ res += task.file.record_count
+ else:
+ arrow_scan = ArrowScan(
+ table_metadata=self.table_metadata,
+ io=self.io,
+ projected_schema=self.projection(),
+ row_filter=self.row_filter,
+ case_sensitive=self.case_sensitive,
+ )
+ if task.file.file_size_in_bytes > 512 * 1024 * 1024:
+ target_schema = schema_to_pyarrow(self.projection())
+ batches = arrow_scan.to_record_batches([task])
+ from pyarrow import RecordBatchReader
+
+ reader = RecordBatchReader.from_batches(target_schema, batches)
+ for batch in reader:
+ res += batch.num_rows
+ else:
+ tbl = arrow_scan.to_table([task])
+ res += len(tbl)
Review Comment:
Let's keep it simple for now, I don't think we cover the other case in a test
```suggestion
tbl = arrow_scan.to_table([task])
res += len(tbl)
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1950707206
##########
pyiceberg/table/__init__.py:
##########
@@ -1466,6 +1475,25 @@ def _build_partition_evaluator(self, spec_id: int) -> Callable[[DataFile], bool]
# shared instance across multiple threads.
return lambda data_file: expression_evaluator(partition_schema, partition_expr, self.case_sensitive)(data_file.partition)
+ def _build_residual_evaluator(self, spec_id: int) -> Callable[[DataFile], ResidualEvaluator]:
+ spec = self.table_metadata.specs()[spec_id]
+
+ # The lambda created here is run in multiple threads.
+ # So we avoid creating _EvaluatorExpression methods bound to a single
+ # shared instance across multiple threads.
+ # return lambda data_file: (partition_schema, partition_expr, self.case_sensitive)(data_file.partition)
+ from pyiceberg.expressions.visitors import residual_evaluator_of
+
+ # assert self.row_filter == False
+ return lambda datafile: (
+ residual_evaluator_of(
+ spec=spec,
+ expr=self.row_filter,
+ case_sensitive=self.case_sensitive,
+ schema=self.projection(),
Review Comment:
This should not use the projected schema, but the current schema. It might be that you're filtering on a column that's not part of the projection.
```suggestion
schema=self.table_metadata.schema(),
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "tusharchou (via GitHub)" <gi...@apache.org>.
tusharchou commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1915063325
##########
pyiceberg/table/__init__.py:
##########
@@ -1522,8 +1555,9 @@ def plan_files(self) -> Iterable[FileScanTask]:
data_entry,
positional_delete_entries,
),
+ residual=residual,
Review Comment:
@Fokko as manifest isn't accessible there can we instead write -
`residual=residual_evaluators[data_entry.data_file.spec_id](data_entry.data_file.partition),`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Count rows as a metadata only operation [iceberg-python]
Posted by "gli-chris-hao (via GitHub)" <gi...@apache.org>.
gli-chris-hao commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1900228680
##########
pyiceberg/table/__init__.py:
##########
@@ -1594,6 +1609,29 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ """
+ Usage: calutates the total number of records in a Scan that haven't had positional deletes
+ """
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the fiter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and not len(task.delete_files):
+ # Every File has a metadata stat that stores the file record count
+ res += task.file.record_count
+ else:
+ from pyiceberg.io.pyarrow import ArrowScan
+ tbl = ArrowScan(
+ self.table_metadata, self.io, self.projection(), self.row_filter, self.case_sensitive, self.limit
+ ).to_table([task])
+ res += len(tbl)
+ return res
Review Comment:
I love this approach! My only concern is about loading too much data into memory at once, although this is loading just one file at a time, in the worst case some file could potentially be very large? Shall we define a threshold and check, for example, if `file size < XXX`, load entire file, otherwise turn it into `pa.RecordBatchReader` and read stream of record batches for counting.
```
target_schema = schema_to_pyarrow(self.projection())
batches = ArrowScan(
self.table_metadata, self.io, self.projection(), self.row_filter, self.case_sensitive, self.limit
).to_record_batches([task])
reader = pa.RecordBatchReader.from_batches(
target_schema,
batches,
)
count = 0
for batch in reader:
count += batch.num_rows
return count
```
https://github.com/apache/iceberg-python/blob/main/pyiceberg/table/__init__.py#L1541-L1564
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1913339866
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression):
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is None:
+ return self.visit_true()
Review Comment:
Similar to Java, I think we can return `AlwaysTrue` directly, instead of calling `visit_true()`: https://github.com/apache/iceberg/blob/5fd16b5bfeb85e12b5a9ecb4e39504389d7b72ed/api/src/main/java/org/apache/iceberg/expressions/ResidualEvaluator.java#L157
```suggestion
return AlwaysTrue()
```
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
Review Comment:
Can we copy this comment?
https://github.com/apache/iceberg/blob/5fd16b5bfeb85e12b5a9ecb4e39504389d7b72ed/api/src/main/java/org/apache/iceberg/expressions/ResidualEvaluator.java#L32
I think that would be helpful for the reader.
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression):
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is not None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ from pyiceberg.types import StructType
Review Comment:
Let's move this import to the top
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression):
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is not None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ from pyiceberg.types import StructType
+
+ def struct_to_schema(struct: StructType) -> Schema:
+ return Schema(*list(struct.fields))
+
+ for part in parts:
+ strict_projection = part.transform.strict_project(part.name, predicate)
+ strict_result = None
+
+ if strict_projection is not None:
+ bound = strict_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound, BoundPredicate):
+ strict_result = super().visit_bound_predicate(bound)
+ else:
+ strict_result = bound
+
+ if strict_result is not None and isinstance(strict_result, AlwaysTrue):
+ return AlwaysTrue()
+
+ inclusive_projection = part.transform.project(part.name, predicate)
+ inclusive_result = None
+ if inclusive_projection is not None:
+ bound_inclusive = inclusive_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound_inclusive, BoundPredicate):
+ # using predicate method specific to inclusive
+ inclusive_result = super().visit_bound_predicate(bound_inclusive)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or
+ # alwaysFalse
+ inclusive_result = bound_inclusive
+ if inclusive_result is not None and isinstance(inclusive_result, AlwaysFalse):
+ return AlwaysFalse()
+
+ return predicate
+
+ def visit_unbound_predicate(self, predicate: UnboundPredicate[L]) -> BooleanExpression:
+ bound = predicate.bind(self.schema, case_sensitive=True)
+
+ if isinstance(bound, BoundPredicate):
+ bound_residual = self.visit_bound_predicate(predicate=bound)
+ # if isinstance(bound_residual, BooleanExpression):
+ if bound_residual not in (AlwaysFalse(), AlwaysTrue()):
+ # replace inclusive original unbound predicate
+ return predicate
+
+ # use the non-predicate residual (e.g. alwaysTrue)
+ return bound_residual
+
+ # if binding didn't result in a Predicate, return the expression
+ return bound
+
+
+class ResidualEvaluator(ResidualVisitor):
+ def residual_for(self, partition_data: Record) -> BooleanExpression:
+ return self.eval(partition_data)
+
+
+class UnpartitionedResidualEvaluator(ResidualEvaluator):
+ # Finds the residuals for an Expression the partitions in the given PartitionSpec
+ def __init__(self, schema: Schema, expr: BooleanExpression):
+ from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC
Review Comment:
Let's move this import to the top as well 👍
##########
pyiceberg/table/__init__.py:
##########
@@ -1522,8 +1555,9 @@ def plan_files(self) -> Iterable[FileScanTask]:
data_entry,
positional_delete_entries,
),
+ residual=residual,
Review Comment:
Instead of carrying the residuals all the way through, I think we can compute them at this point instead:
```suggestion
residual=residual_evaluators[manifest.partition_spec_id](data_entry.data_file.partition),
```
If a whole manifest is rejected, we don't even compute the evaluator itself.
##########
pyiceberg/table/__init__.py:
##########
@@ -1596,6 +1630,43 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ # Usage: Calculates the total number of records in a Scan that haven't had positional deletes.
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the filter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and not len(task.delete_files):
+ # Every File has a metadata stat that stores the file record count
+ res += task.file.record_count
+ else:
Review Comment:
I think the positional deletes are missing from the `else:` branch. How about re-using `_task_to_record_batches` for now?
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
Review Comment:
Can we add:
```
expr: BooleanExpression
```
to the class variables as well?
##########
pyiceberg/table/__init__.py:
##########
@@ -1436,6 +1443,27 @@ def _build_partition_evaluator(self, spec_id: int) -> Callable[[DataFile], bool]
# shared instance across multiple threads.
return lambda data_file: expression_evaluator(partition_schema, partition_expr, self.case_sensitive)(data_file.partition)
+ from pyiceberg.expressions.visitors import ResidualEvaluator
+
+ def _build_residual_evaluator(self, spec_id: int) -> Callable[[DataFile], ResidualEvaluator]:
+ spec = self.table_metadata.specs()[spec_id]
+
+ # The lambda created here is run in multiple threads.
+ # So we avoid creating _EvaluatorExpression methods bound to a single
+ # shared instance across multiple threads.
+ # return lambda data_file: (partition_schema, partition_expr, self.case_sensitive)(data_file.partition)
+ from pyiceberg.expressions.visitors import residual_evaluator_of
+
+ # assert self.row_filter == False
+ return lambda datafile: (
+ residual_evaluator_of(
+ spec=spec,
+ expr=self.row_filter,
+ case_sensitive=self.case_sensitive,
+ schema=self.table_metadata.schema(),
Review Comment:
In a separate PR we want to refactor this, where we use the same schema as in `projection()`
##########
pyiceberg/table/__init__.py:
##########
@@ -1343,33 +1346,37 @@ class FileScanTask(ScanTask):
delete_files: Set[DataFile]
start: int
length: int
+ residual: BooleanExpression
def __init__(
self,
data_file: DataFile,
delete_files: Optional[Set[DataFile]] = None,
start: Optional[int] = None,
length: Optional[int] = None,
+ residual: Optional[BooleanExpression] = None,
Review Comment:
I think we should make this one required. If there is no residual, it should default to `AlwaysTrue()`.
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression):
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is not None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ from pyiceberg.types import StructType
+
+ def struct_to_schema(struct: StructType) -> Schema:
+ return Schema(*list(struct.fields))
+
+ for part in parts:
+ strict_projection = part.transform.strict_project(part.name, predicate)
+ strict_result = None
+
+ if strict_projection is not None:
+ bound = strict_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound, BoundPredicate):
+ strict_result = super().visit_bound_predicate(bound)
+ else:
+ strict_result = bound
Review Comment:
Let's keep the comments from Java in here, I think they are pretty helpful:
```suggestion
# if the result is not a predicate, then it must be a constant like alwaysTrue or alwaysFalse
strict_result = bound
```
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression):
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is not None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
Review Comment:
Java takes a different approach and checks for `NaN`:
```suggestion
if isnan(term.eval(self.struct)):
return self.visit_true()
else:
return self.visit_false()
```
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression):
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is not None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ from pyiceberg.types import StructType
+
+ def struct_to_schema(struct: StructType) -> Schema:
+ return Schema(*list(struct.fields))
Review Comment:
The conversion to a list is not needed:
```suggestion
return Schema(*struct.fields)
```
```
python3
Python 3.10.14 (main, Mar 19 2024, 21:46:16) [Clang 15.0.0 (clang-1500.3.9.4)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> def vo(*int):
... print(int)
...
>>> vo(*(1,2,3))
(1, 2, 3)
```
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression):
Review Comment:
```suggestion
def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression) -> None:
```
##########
pyiceberg/table/__init__.py:
##########
@@ -1466,6 +1494,9 @@ def plan_files(self) -> Iterable[FileScanTask]:
# the filter depends on the partition spec used to write the manifest file, so create a cache of filters for each spec id
manifest_evaluators: Dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)
+ from pyiceberg.expressions.visitors import ResidualEvaluator
Review Comment:
I think we can move this import to the top as well
##########
pyiceberg/table/__init__.py:
##########
@@ -1596,6 +1630,43 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ # Usage: Calculates the total number of records in a Scan that haven't had positional deletes.
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the filter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and not len(task.delete_files):
+ # Every File has a metadata stat that stores the file record count
+ res += task.file.record_count
+ else:
+ from pyiceberg.io.pyarrow import ArrowScan, schema_to_pyarrow
+
+ arrow_scan = ArrowScan(
+ table_metadata=self.table_metadata,
+ io=self.io,
+ projected_schema=self.projection(),
Review Comment:
We are not interested in any fields at all, so we could just pass in an empty schema. This will also relax @gli-chris-hao his concern regarding memory pressure 👍
##########
pyiceberg/table/__init__.py:
##########
@@ -1596,6 +1630,43 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ # Usage: Calculates the total number of records in a Scan that haven't had positional deletes.
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the filter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and not len(task.delete_files):
+ # Every File has a metadata stat that stores the file record count
+ res += task.file.record_count
+ else:
+ from pyiceberg.io.pyarrow import ArrowScan, schema_to_pyarrow
+
+ arrow_scan = ArrowScan(
+ table_metadata=self.table_metadata,
+ io=self.io,
+ projected_schema=self.projection(),
+ row_filter=self.row_filter,
+ case_sensitive=self.case_sensitive,
+ limit=self.limit,
Review Comment:
This will lead to incorrect results since we don't want to limit the count
```suggestion
limit=self.limit,
```
##########
pyiceberg/table/__init__.py:
##########
@@ -1596,6 +1630,43 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ # Usage: Calculates the total number of records in a Scan that haven't had positional deletes.
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the filter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and not len(task.delete_files):
Review Comment:
I think this is a bit more explicit:
```suggestion
if task.residual == AlwaysTrue() and len(task.delete_files) == 0:
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Count rows as a metadata only operation [iceberg-python]
Posted by "tusharchou (via GitHub)" <gi...@apache.org>.
tusharchou commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1896538967
##########
pyiceberg/table/__init__.py:
##########
@@ -1493,6 +1496,13 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ res = 0
+ tasks = self.plan_files()
Review Comment:
Hi @Fokko I have added Residual Evaluator with Tests.
Now I am trying to create the breaking tests for count where delete has occurred and the counts should differ
```
@pytest.mark.integration
@pytest.mark.filterwarnings("ignore:Merge on read is not yet supported, falling back to copy-on-write")
def test_delete_partitioned_table_positional_deletes(spark: SparkSession, session_catalog: RestCatalog) -> None:
identifier = "default.table_partitioned_delete"
run_spark_commands(
spark,
[
f"DROP TABLE IF EXISTS {identifier}",
f"""
CREATE TABLE {identifier} (
number_partitioned int,
number int
)
USING iceberg
PARTITIONED BY (number_partitioned)
TBLPROPERTIES(
'format-version' = 2,
'write.delete.mode'='merge-on-read',
'write.update.mode'='merge-on-read',
'write.merge.mode'='merge-on-read'
)
""",
f"""
INSERT INTO {identifier} VALUES (10, 20), (10, 30), (10, 40)
""",
# Generate a positional delete
f"""
DELETE FROM {identifier} WHERE number = 30
""",
],
)
tbl = session_catalog.load_table(identifier)
# Assert that there is just a single Parquet file, that has one merge on read file
files = list(tbl.scan().plan_files())
assert len(files) == 1
assert len(files[0].delete_files) == 1
# Will rewrite a data file without the positional delete
assert tbl.scan().to_arrow().to_pydict() == {"number_partitioned": [10, 10], "number": [20, 40]}
assert tbl.scan().count() == 2
assert tbl.scan().count() == 1
tbl.delete(EqualTo("number", 40))
# One positional delete has been added, but an OVERWRITE status is set
# https://github.com/apache/iceberg/issues/10122
assert [snapshot.summary.operation.value for snapshot in tbl.snapshots()] == ["append", "overwrite", "overwrite"]
assert tbl.scan().to_arrow().to_pydict() == {"number_partitioned": [10], "number": [20]}
assert tbl.scan().count() == 1
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1950654390
##########
tests/expressions/test_residual_evaluator.py:
##########
@@ -0,0 +1,251 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# pylint:disable=redefined-outer-name
+import pytest
+
+from pyiceberg.expressions import (
+ AlwaysFalse,
+ AlwaysTrue,
+ And,
+ EqualTo,
+ GreaterThan,
+ GreaterThanOrEqual,
+ In,
+ IsNaN,
+ IsNull,
+ LessThan,
+ NotIn,
+ NotNaN,
+ NotNull,
+ NotStartsWith,
+ Or,
+ StartsWith,
+)
+from pyiceberg.expressions.literals import literal
+from pyiceberg.expressions.visitors import residual_evaluator_of
+from pyiceberg.partitioning import PartitionField, PartitionSpec
+from pyiceberg.schema import Schema
+from pyiceberg.transforms import DayTransform, IdentityTransform
+from pyiceberg.typedef import Record
+from pyiceberg.types import DoubleType, FloatType, IntegerType, NestedField, TimestampType
+
+
+def test_identity_transform_residual() -> None:
+ schema = Schema(NestedField(50, "dateint", IntegerType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "dateint_part"))
+
+ predicate = Or(
+ Or(
+ And(EqualTo("dateint", 20170815), LessThan("hour", 12)),
+ And(LessThan("dateint", 20170815), GreaterThan("dateint", 20170801)),
+ ),
+ And(EqualTo("dateint", 20170801), GreaterThan("hour", 11)),
+ )
+ res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+
+ residual = res_eval.residual_for(Record(dateint=20170815))
+
+ # assert residual == True
+ assert isinstance(residual, LessThan)
+ assert residual.term.name == "hour" # type: ignore
+ # assert residual.term.field.name == 'hour'
+ assert residual.literal.value == 12
+ assert type(residual) is LessThan
+
+ residual = res_eval.residual_for(Record(dateint=20170801))
+
+ # assert isinstance(residual, UnboundPredicate)
+ from pyiceberg.expressions import LiteralPredicate
+
+ assert isinstance(residual, LiteralPredicate)
+ # assert isinstance(residual, GreaterThan)
+ assert residual.term.name == "hour" # type: ignore
+ # assert residual.term.
+ assert residual.literal.value == 11 # type :ignore
+ # assert type(residual) == BoundGreaterThan
+
+ residual = res_eval.residual_for(Record(dateint=20170812))
+
+ assert residual == AlwaysTrue()
+
+ residual = res_eval.residual_for(Record(dateint=20170817))
+
+ assert residual == AlwaysFalse()
+
+
+def test_case_insensitive_identity_transform_residuals() -> None:
+ schema = Schema(NestedField(50, "dateint", IntegerType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "dateint_part"))
+
+ predicate = Or(
+ Or(
+ And(EqualTo("DATEINT", 20170815), LessThan("HOUR", 12)),
+ And(LessThan("dateint", 20170815), GreaterThan("dateint", 20170801)),
+ ),
+ And(EqualTo("Dateint", 20170801), GreaterThan("hOUr", 11)),
+ )
+ res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+
+ with pytest.raises(ValueError) as e:
+ res_eval.residual_for(Record(dateint=20170815))
+ assert "Could not find field with name DATEINT, case_sensitive=True" in str(e.value)
+
+
+def test_unpartitioned_residuals() -> None:
+ expressions = [
+ AlwaysTrue(),
+ AlwaysFalse(),
+ LessThan("a", 5),
+ GreaterThanOrEqual("b", 16),
+ NotNull("c"),
+ IsNull("d"),
+ In("e", [1, 2, 3]),
+ NotIn("f", [1, 2, 3]),
+ NotNaN("g"),
+ IsNaN("h"),
+ StartsWith("data", "abcd"),
+ NotStartsWith("data", "abcd"),
+ ]
+
+ schema = Schema(
+ NestedField(50, "dateint", IntegerType()), NestedField(51, "hour", IntegerType()), NestedField(52, "a", IntegerType())
+ )
+ for expr in expressions:
+ from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC
+
+ residual_evaluator = residual_evaluator_of(UNPARTITIONED_PARTITION_SPEC, expr, True, schema=schema)
+ assert residual_evaluator.residual_for(Record()) == expr
+
+
+def test_in() -> None:
+ schema = Schema(NestedField(50, "dateint", IntegerType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "dateint_part"))
+
+ predicate = In("dateint", [20170815, 20170816, 20170817])
+
+ res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+
+ residual = res_eval.residual_for(Record(dateint=20170815))
+
+ assert residual == AlwaysTrue()
+
+
+def test_in_timestamp() -> None:
+ schema = Schema(NestedField(50, "ts", TimestampType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1000, DayTransform(), "ts_part"))
+
+ date_20191201 = literal("2019-12-01T00:00:00").to(TimestampType()).value
+ date_20191202 = literal("2019-12-02T00:00:00").to(TimestampType()).value
+
+ day = DayTransform().transform(TimestampType())
+ ts_day = day(date_20191201) # type: ignore
+
+ pred = In("ts", [date_20191202, date_20191201])
+
+ res_eval = residual_evaluator_of(spec=spec, expr=pred, case_sensitive=True, schema=schema)
+
+ residual = res_eval.residual_for(Record(ts_day))
+ assert residual == pred
+
+ residual = res_eval.residual_for(Record(ts_day + 3)) # type: ignore
+ assert residual == AlwaysFalse()
+
+
+def test_not_in() -> None:
+ schema = Schema(NestedField(50, "dateint", IntegerType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "dateint_part"))
+
+ predicate = NotIn("dateint", [20170815, 20170816, 20170817])
+
+ res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+
+ residual = res_eval.residual_for(Record(dateint=20180815))
+ assert residual == AlwaysTrue()
+
+ residual = res_eval.residual_for(Record(dateint=20170815))
+ assert residual == AlwaysFalse()
+
+
+def test_is_nan() -> None:
+ schema = Schema(NestedField(50, "double", DoubleType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "double_part"))
+
+ predicate = IsNaN("double")
+
+ res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+
+ residual = res_eval.residual_for(Record(double=None))
Review Comment:
```suggestion
residual = res_eval.residual_for(Record(double=float('nan')))
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1937048705
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression):
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is not None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
Review Comment:
I think we need to help `mypy` here a bit:
```suggestion
val = term.eval(self.struct)
if isinstance(val, SupportsFloat) and isnan(val):
return self.visit_true()
else:
return self.visit_false()
```
The `nan` check will only be done on `float` and `double` columns, but mypy doesn't know that, so we have to help it a bit :)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Count rows as a metadata only operation [iceberg-python]
Posted by "gli-chris-hao (via GitHub)" <gi...@apache.org>.
gli-chris-hao commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1900228680
##########
pyiceberg/table/__init__.py:
##########
@@ -1594,6 +1609,29 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ """
+ Usage: calutates the total number of records in a Scan that haven't had positional deletes
+ """
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the fiter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and not len(task.delete_files):
+ # Every File has a metadata stat that stores the file record count
+ res += task.file.record_count
+ else:
+ from pyiceberg.io.pyarrow import ArrowScan
+ tbl = ArrowScan(
+ self.table_metadata, self.io, self.projection(), self.row_filter, self.case_sensitive, self.limit
+ ).to_table([task])
+ res += len(tbl)
+ return res
Review Comment:
I love this approach! My only concern is about loading too much data into memory at once, although this is loading just one file at a time, in the worst case some file could potentially be very large? Shall we define a threshold and check, for example, if `file size < 512MB`, load entire file, otherwise turn it into `pa.RecordBatchReader` and read stream of record batches for counting.
```
target_schema = schema_to_pyarrow(self.projection())
batches = ArrowScan(
self.table_metadata, self.io, self.projection(), self.row_filter, self.case_sensitive, self.limit
).to_record_batches([task])
reader = pa.RecordBatchReader.from_batches(
target_schema,
batches,
)
count = 0
for batch in reader:
count += batch.num_rows
return count
```
https://github.com/apache/iceberg-python/blob/main/pyiceberg/table/__init__.py#L1541-L1564
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1915102464
##########
pyiceberg/table/__init__.py:
##########
@@ -1522,8 +1555,9 @@ def plan_files(self) -> Iterable[FileScanTask]:
data_entry,
positional_delete_entries,
),
+ residual=residual,
Review Comment:
@tusharchou I don't really think this one through. Without any additional changes, this would look like:
```python
return [
FileScanTask(
data_entry.data_file,
delete_files=_match_deletes_to_data_file(
data_entry,
positional_delete_entries,
),
residual=residual_evaluators[data_entry.data_file.spec_id](data_entry.data_file).residual_for(
data_entry.data_file.partition
),
)
for data_entry in data_entries
]
```
Where we re-create the evaluator all the time
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1922535895
##########
pyiceberg/table/__init__.py:
##########
@@ -1596,6 +1630,43 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ # Usage: Calculates the total number of records in a Scan that haven't had positional deletes.
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the filter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and not len(task.delete_files):
+ # Every File has a metadata stat that stores the file record count
+ res += task.file.record_count
+ else:
+ from pyiceberg.io.pyarrow import ArrowScan, schema_to_pyarrow
+
+ arrow_scan = ArrowScan(
+ table_metadata=self.table_metadata,
+ io=self.io,
+ projected_schema=self.projection(),
Review Comment:
I was suggesting that indeed, what kind of error? Otherwise we can leave it as is for now and do it in a subsequent PR
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Count rows as a metadata only operation [iceberg-python]
Posted by "kevinjqliu (via GitHub)" <gi...@apache.org>.
kevinjqliu commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1862579074
##########
pyiceberg/table/__init__.py:
##########
@@ -1493,6 +1496,13 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ res = 0
+ tasks = self.plan_files()
Review Comment:
this count will not be accurate when there are deletes files
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Count rows as a metadata only operation [iceberg-python]
Posted by "jayceslesar (via GitHub)" <gi...@apache.org>.
jayceslesar commented on PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#issuecomment-2512673626
Question: Does it make sense to expose this as the `__len__` dunder method because python? It would just return the `self.count()`
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "tusharchou (via GitHub)" <gi...@apache.org>.
tusharchou commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1920208710
##########
pyiceberg/table/__init__.py:
##########
@@ -1596,6 +1630,43 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ # Usage: Calculates the total number of records in a Scan that haven't had positional deletes.
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the filter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and not len(task.delete_files):
+ # Every File has a metadata stat that stores the file record count
+ res += task.file.record_count
+ else:
Review Comment:
@Fokko I have added test of positional delete and it works. `ArrowScan.to_record_batches` calls `_task_to_record_batches' internally. Can you please take another look.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "tusharchou (via GitHub)" <gi...@apache.org>.
tusharchou commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1936663741
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression):
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is not None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
Review Comment:
Hi@Fokko, I would require some help here.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1944733940
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1732,230 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ """Finds the residuals for an Expression the partitions in the given PartitionSpec.
+
+ A residual expression is made by partially evaluating an expression using partition values.
+ For example, if a table is partitioned by day(utc_timestamp) and is read with a filter expression
+ utc_timestamp >= a and utc_timestamp <= b, then there are 4 possible residuals expressions
+ for the partition data, d:
+
+
+ 1. If d > day(a) and d < day(b), the residual is always true
+ 2. If d == day(a) and d != day(b), the residual is utc_timestamp >= a
+ 3. if d == day(b) and d != day(a), the residual is utc_timestamp <= b
+ 4. If d == day(a) == day(b), the residual is utc_timestamp >= a and utc_timestamp <= b
+
+ Partition data is passed using StructLike. Residuals are returned by residualFor(StructLike).
+
+ This class is thread-safe.
+ """
+
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+ expr: BooleanExpression
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression) -> None:
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and not math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ def struct_to_schema(struct: StructType) -> Schema:
+ return Schema(*struct.fields)
+
+ for part in parts:
+ strict_projection = part.transform.strict_project(part.name, predicate)
+ strict_result = None
+
+ if strict_projection is not None:
+ bound = strict_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound, BoundPredicate):
+ strict_result = super().visit_bound_predicate(bound)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or alwaysFalse
+ strict_result = bound
+
+ if strict_result is not None and isinstance(strict_result, AlwaysTrue):
Review Comment:
In Python we can simplify this a bit:
```suggestion
if isinstance(strict_result, AlwaysTrue):
```
```python
>>> from pyiceberg.expressions import AlwaysTrue
>>> isinstance(None, AlwaysTrue)
False
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1944740942
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1732,230 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ """Finds the residuals for an Expression the partitions in the given PartitionSpec.
+
+ A residual expression is made by partially evaluating an expression using partition values.
+ For example, if a table is partitioned by day(utc_timestamp) and is read with a filter expression
+ utc_timestamp >= a and utc_timestamp <= b, then there are 4 possible residuals expressions
+ for the partition data, d:
+
+
+ 1. If d > day(a) and d < day(b), the residual is always true
+ 2. If d == day(a) and d != day(b), the residual is utc_timestamp >= a
+ 3. if d == day(b) and d != day(a), the residual is utc_timestamp <= b
+ 4. If d == day(a) == day(b), the residual is utc_timestamp >= a and utc_timestamp <= b
+
+ Partition data is passed using StructLike. Residuals are returned by residualFor(StructLike).
+
+ This class is thread-safe.
+ """
+
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+ expr: BooleanExpression
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression) -> None:
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and not math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ def struct_to_schema(struct: StructType) -> Schema:
+ return Schema(*struct.fields)
+
+ for part in parts:
+ strict_projection = part.transform.strict_project(part.name, predicate)
+ strict_result = None
+
+ if strict_projection is not None:
+ bound = strict_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
Review Comment:
```suggestion
bound = strict_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)), case_sensitive=self.case_sensitive)
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Count rows as a metadata only operation [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1863537090
##########
pyiceberg/table/__init__.py:
##########
@@ -1493,6 +1496,13 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ res = 0
+ tasks = self.plan_files()
Review Comment:
Yes, this can be widely off, not just because the merge-on-read deletes, but because `plan_files` returns all the files that (might) contain relevant rows. For example, if it cannot be determined if has relevant data, it will be returned by `plan_files`.
I think there are two ways forward:
- One is similar to how we handle deletes. For deletes, we check if the whole file matches, if this is the case, then we can simply drop the file from the metadata. You can find the code [here](https://github.com/apache/iceberg-python/blob/5bef1bfe677df7016dc37b8db87650c34faceb7a/pyiceberg/table/update/snapshot.py#L359-L427). If a file fully matches, is is valid to use `task.file.record_count`. We would need to extend this to also see if there are also merge-on-read deletes as Kevin already mentioned, or just fail when there are positional deletes.
- A cleaner option, but this is a bit more work, but pretty exciting, would be to include the `residual-predicate` in the `FileScanTask`. When we run a query, like `day_added = 2024-12-01 and user_id = 10`, then the `day_added = 2024-12-01` might be satisfied with the partitioning already. This is the case when the table is partitioned by day, and we know that all the data in the file evaluates `true` for `day_added = 2024-12-01`, then we need to open the file, and filter for `user_id = 10`. If we would leave out the `user_id = 10`, then it would be `ALWAYS_TRUE`, and then we know that we can just use `task.file.record_count`. This way we could very easily loop over the `.plan_files()`:
```python
def count(self) -> int:
res = 0
tasks = self.plan_files()
for task in tasks:
if task.residual == ALWAYS_TRUE and len(task.delete_files):
res += task.file.record_count
else:
# pseudocode, open the table, and apply the filter and deletes
res += len(_table_from_scan_task(task))
return res
```
To get to the second step, we first have to port the `ResidualEvaluator`. The java code can be found [here](https://github.com/apache/iceberg/blob/main/api/src/main/java/org/apache/iceberg/expressions/ResidualEvaluator.java), including some [excellent tests](https://github.com/apache/iceberg/blob/main/api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java).
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "tusharchou (via GitHub)" <gi...@apache.org>.
tusharchou commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1920210749
##########
pyiceberg/table/__init__.py:
##########
@@ -1596,6 +1630,43 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ # Usage: Calculates the total number of records in a Scan that haven't had positional deletes.
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the filter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and not len(task.delete_files):
+ # Every File has a metadata stat that stores the file record count
+ res += task.file.record_count
+ else:
+ from pyiceberg.io.pyarrow import ArrowScan, schema_to_pyarrow
+
+ arrow_scan = ArrowScan(
+ table_metadata=self.table_metadata,
+ io=self.io,
+ projected_schema=self.projection(),
Review Comment:
Simple question: What would an empty schema look like passing a 'Schema()` object throws error
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1944746576
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1732,230 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ """Finds the residuals for an Expression the partitions in the given PartitionSpec.
+
+ A residual expression is made by partially evaluating an expression using partition values.
+ For example, if a table is partitioned by day(utc_timestamp) and is read with a filter expression
+ utc_timestamp >= a and utc_timestamp <= b, then there are 4 possible residuals expressions
+ for the partition data, d:
+
+
+ 1. If d > day(a) and d < day(b), the residual is always true
+ 2. If d == day(a) and d != day(b), the residual is utc_timestamp >= a
+ 3. if d == day(b) and d != day(a), the residual is utc_timestamp <= b
+ 4. If d == day(a) == day(b), the residual is utc_timestamp >= a and utc_timestamp <= b
+
+ Partition data is passed using StructLike. Residuals are returned by residualFor(StructLike).
+
+ This class is thread-safe.
+ """
+
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+ expr: BooleanExpression
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression) -> None:
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and not math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ def struct_to_schema(struct: StructType) -> Schema:
+ return Schema(*struct.fields)
+
+ for part in parts:
+ strict_projection = part.transform.strict_project(part.name, predicate)
+ strict_result = None
+
+ if strict_projection is not None:
+ bound = strict_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound, BoundPredicate):
+ strict_result = super().visit_bound_predicate(bound)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or alwaysFalse
+ strict_result = bound
+
+ if strict_result is not None and isinstance(strict_result, AlwaysTrue):
+ return AlwaysTrue()
+
+ inclusive_projection = part.transform.project(part.name, predicate)
+ inclusive_result = None
+ if inclusive_projection is not None:
+ bound_inclusive = inclusive_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound_inclusive, BoundPredicate):
+ # using predicate method specific to inclusive
+ inclusive_result = super().visit_bound_predicate(bound_inclusive)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or
+ # alwaysFalse
+ inclusive_result = bound_inclusive
+ if inclusive_result is not None and isinstance(inclusive_result, AlwaysFalse):
+ return AlwaysFalse()
+
+ return predicate
+
+ def visit_unbound_predicate(self, predicate: UnboundPredicate[L]) -> BooleanExpression:
+ bound = predicate.bind(self.schema, case_sensitive=True)
+
+ if isinstance(bound, BoundPredicate):
+ bound_residual = self.visit_bound_predicate(predicate=bound)
+ # if isinstance(bound_residual, BooleanExpression):
+ if bound_residual not in (AlwaysFalse(), AlwaysTrue()):
+ # replace inclusive original unbound predicate
+ return predicate
+
+ # use the non-predicate residual (e.g. alwaysTrue)
+ return bound_residual
+
+ # if binding didn't result in a Predicate, return the expression
+ return bound
+
+
+class ResidualEvaluator(ResidualVisitor):
+ def residual_for(self, partition_data: Record) -> BooleanExpression:
+ return self.eval(partition_data)
+
+
+class UnpartitionedResidualEvaluator(ResidualEvaluator):
+ # Finds the residuals for an Expression the partitions in the given PartitionSpec
+ def __init__(self, schema: Schema, expr: BooleanExpression):
+ super().__init__(schema=schema, spec=UNPARTITIONED_PARTITION_SPEC, expr=expr, case_sensitive=False)
+ self.expr = expr
+
+ def residual_for(self, partition_data: Record) -> BooleanExpression:
+ return self.expr
+
+
+def residual_evaluator_of(
+ spec: PartitionSpec, expr: BooleanExpression, case_sensitive: bool, schema: Schema
+) -> ResidualEvaluator:
+ if len(spec.fields) != 0:
+ return ResidualEvaluator(spec=spec, expr=expr, schema=schema, case_sensitive=case_sensitive)
+ else:
+ return UnpartitionedResidualEvaluator(schema=schema, expr=expr)
Review Comment:
Just as style thing :)
```suggestion
return UnpartitionedResidualEvaluator(schema=schema, expr=expr) if spec.is_unpartitioned() else ResidualEvaluator(spec=spec, expr=expr, schema=schema, case_sensitive=case_sensitive)
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko merged PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1944743154
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1732,230 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ """Finds the residuals for an Expression the partitions in the given PartitionSpec.
+
+ A residual expression is made by partially evaluating an expression using partition values.
+ For example, if a table is partitioned by day(utc_timestamp) and is read with a filter expression
+ utc_timestamp >= a and utc_timestamp <= b, then there are 4 possible residuals expressions
+ for the partition data, d:
+
+
+ 1. If d > day(a) and d < day(b), the residual is always true
+ 2. If d == day(a) and d != day(b), the residual is utc_timestamp >= a
+ 3. if d == day(b) and d != day(a), the residual is utc_timestamp <= b
+ 4. If d == day(a) == day(b), the residual is utc_timestamp >= a and utc_timestamp <= b
+
+ Partition data is passed using StructLike. Residuals are returned by residualFor(StructLike).
+
+ This class is thread-safe.
+ """
+
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+ expr: BooleanExpression
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression) -> None:
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and not math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ def struct_to_schema(struct: StructType) -> Schema:
+ return Schema(*struct.fields)
+
+ for part in parts:
+ strict_projection = part.transform.strict_project(part.name, predicate)
+ strict_result = None
+
+ if strict_projection is not None:
+ bound = strict_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound, BoundPredicate):
+ strict_result = super().visit_bound_predicate(bound)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or alwaysFalse
+ strict_result = bound
+
+ if strict_result is not None and isinstance(strict_result, AlwaysTrue):
+ return AlwaysTrue()
+
+ inclusive_projection = part.transform.project(part.name, predicate)
+ inclusive_result = None
+ if inclusive_projection is not None:
+ bound_inclusive = inclusive_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound_inclusive, BoundPredicate):
+ # using predicate method specific to inclusive
+ inclusive_result = super().visit_bound_predicate(bound_inclusive)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or
+ # alwaysFalse
+ inclusive_result = bound_inclusive
+ if inclusive_result is not None and isinstance(inclusive_result, AlwaysFalse):
Review Comment:
```suggestion
if isinstance(inclusive_result, AlwaysFalse):
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1944716167
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1732,230 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ """Finds the residuals for an Expression the partitions in the given PartitionSpec.
+
+ A residual expression is made by partially evaluating an expression using partition values.
+ For example, if a table is partitioned by day(utc_timestamp) and is read with a filter expression
+ utc_timestamp >= a and utc_timestamp <= b, then there are 4 possible residuals expressions
+ for the partition data, d:
+
+
+ 1. If d > day(a) and d < day(b), the residual is always true
+ 2. If d == day(a) and d != day(b), the residual is utc_timestamp >= a
+ 3. if d == day(b) and d != day(a), the residual is utc_timestamp <= b
+ 4. If d == day(a) == day(b), the residual is utc_timestamp >= a and utc_timestamp <= b
+
+ Partition data is passed using StructLike. Residuals are returned by residualFor(StructLike).
+
+ This class is thread-safe.
Review Comment:
```suggestion
1. If d > day(a) and d < day(b), the residual is always true
2. If d == day(a) and d != day(b), the residual is utc_timestamp > a
3. if d == day(b) and d != day(a), the residual is utc_timestamp < b
4. If d == day(a) == day(b), the residual is utc_timestamp > a and utc_timestamp < b
Partition data is passed using StructLike. Residuals are returned by residualFor(StructLike).
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1950682935
##########
tests/expressions/test_residual_evaluator.py:
##########
@@ -0,0 +1,251 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+# pylint:disable=redefined-outer-name
+import pytest
+
+from pyiceberg.expressions import (
+ AlwaysFalse,
+ AlwaysTrue,
+ And,
+ EqualTo,
+ GreaterThan,
+ GreaterThanOrEqual,
+ In,
+ IsNaN,
+ IsNull,
+ LessThan,
+ NotIn,
+ NotNaN,
+ NotNull,
+ NotStartsWith,
+ Or,
+ StartsWith,
+)
+from pyiceberg.expressions.literals import literal
+from pyiceberg.expressions.visitors import residual_evaluator_of
+from pyiceberg.partitioning import PartitionField, PartitionSpec
+from pyiceberg.schema import Schema
+from pyiceberg.transforms import DayTransform, IdentityTransform
+from pyiceberg.typedef import Record
+from pyiceberg.types import DoubleType, FloatType, IntegerType, NestedField, TimestampType
+
+
+def test_identity_transform_residual() -> None:
+ schema = Schema(NestedField(50, "dateint", IntegerType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "dateint_part"))
+
+ predicate = Or(
+ Or(
+ And(EqualTo("dateint", 20170815), LessThan("hour", 12)),
+ And(LessThan("dateint", 20170815), GreaterThan("dateint", 20170801)),
+ ),
+ And(EqualTo("dateint", 20170801), GreaterThan("hour", 11)),
+ )
+ res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+
+ residual = res_eval.residual_for(Record(dateint=20170815))
+
+ # assert residual == True
+ assert isinstance(residual, LessThan)
+ assert residual.term.name == "hour" # type: ignore
+ # assert residual.term.field.name == 'hour'
+ assert residual.literal.value == 12
+ assert type(residual) is LessThan
+
+ residual = res_eval.residual_for(Record(dateint=20170801))
+
+ # assert isinstance(residual, UnboundPredicate)
+ from pyiceberg.expressions import LiteralPredicate
+
+ assert isinstance(residual, LiteralPredicate)
+ # assert isinstance(residual, GreaterThan)
+ assert residual.term.name == "hour" # type: ignore
+ # assert residual.term.
+ assert residual.literal.value == 11 # type :ignore
+ # assert type(residual) == BoundGreaterThan
+
+ residual = res_eval.residual_for(Record(dateint=20170812))
+
+ assert residual == AlwaysTrue()
+
+ residual = res_eval.residual_for(Record(dateint=20170817))
+
+ assert residual == AlwaysFalse()
+
+
+def test_case_insensitive_identity_transform_residuals() -> None:
+ schema = Schema(NestedField(50, "dateint", IntegerType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "dateint_part"))
+
+ predicate = Or(
+ Or(
+ And(EqualTo("DATEINT", 20170815), LessThan("HOUR", 12)),
+ And(LessThan("dateint", 20170815), GreaterThan("dateint", 20170801)),
+ ),
+ And(EqualTo("Dateint", 20170801), GreaterThan("hOUr", 11)),
+ )
+ res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+
+ with pytest.raises(ValueError) as e:
+ res_eval.residual_for(Record(dateint=20170815))
+ assert "Could not find field with name DATEINT, case_sensitive=True" in str(e.value)
+
+
+def test_unpartitioned_residuals() -> None:
+ expressions = [
+ AlwaysTrue(),
+ AlwaysFalse(),
+ LessThan("a", 5),
+ GreaterThanOrEqual("b", 16),
+ NotNull("c"),
+ IsNull("d"),
+ In("e", [1, 2, 3]),
+ NotIn("f", [1, 2, 3]),
+ NotNaN("g"),
+ IsNaN("h"),
+ StartsWith("data", "abcd"),
+ NotStartsWith("data", "abcd"),
+ ]
+
+ schema = Schema(
+ NestedField(50, "dateint", IntegerType()), NestedField(51, "hour", IntegerType()), NestedField(52, "a", IntegerType())
+ )
+ for expr in expressions:
+ from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC
+
+ residual_evaluator = residual_evaluator_of(UNPARTITIONED_PARTITION_SPEC, expr, True, schema=schema)
+ assert residual_evaluator.residual_for(Record()) == expr
+
+
+def test_in() -> None:
+ schema = Schema(NestedField(50, "dateint", IntegerType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "dateint_part"))
+
+ predicate = In("dateint", [20170815, 20170816, 20170817])
+
+ res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+
+ residual = res_eval.residual_for(Record(dateint=20170815))
+
+ assert residual == AlwaysTrue()
+
+
+def test_in_timestamp() -> None:
+ schema = Schema(NestedField(50, "ts", TimestampType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1000, DayTransform(), "ts_part"))
+
+ date_20191201 = literal("2019-12-01T00:00:00").to(TimestampType()).value
+ date_20191202 = literal("2019-12-02T00:00:00").to(TimestampType()).value
+
+ day = DayTransform().transform(TimestampType())
+ ts_day = day(date_20191201) # type: ignore
+
+ pred = In("ts", [date_20191202, date_20191201])
+
+ res_eval = residual_evaluator_of(spec=spec, expr=pred, case_sensitive=True, schema=schema)
+
+ residual = res_eval.residual_for(Record(ts_day))
+ assert residual == pred
+
+ residual = res_eval.residual_for(Record(ts_day + 3)) # type: ignore
+ assert residual == AlwaysFalse()
+
+
+def test_not_in() -> None:
+ schema = Schema(NestedField(50, "dateint", IntegerType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "dateint_part"))
+
+ predicate = NotIn("dateint", [20170815, 20170816, 20170817])
+
+ res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+
+ residual = res_eval.residual_for(Record(dateint=20180815))
+ assert residual == AlwaysTrue()
+
+ residual = res_eval.residual_for(Record(dateint=20170815))
+ assert residual == AlwaysFalse()
+
+
+def test_is_nan() -> None:
+ schema = Schema(NestedField(50, "double", DoubleType()), NestedField(51, "hour", IntegerType()))
+
+ spec = PartitionSpec(PartitionField(50, 1050, IdentityTransform(), "double_part"))
+
+ predicate = IsNaN("double")
+
+ res_eval = residual_evaluator_of(spec=spec, expr=predicate, case_sensitive=True, schema=schema)
+
+ residual = res_eval.residual_for(Record(double=float('nan')))
Review Comment:
```suggestion
residual = res_eval.residual_for(Record(double=float("nan")))
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#issuecomment-2650619004
Thanks @tusharchou for working on this 🚀
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Count rows as a metadata only operation [iceberg-python]
Posted by "gli-chris-hao (via GitHub)" <gi...@apache.org>.
gli-chris-hao commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1900228680
##########
pyiceberg/table/__init__.py:
##########
@@ -1594,6 +1609,29 @@ def to_ray(self) -> ray.data.dataset.Dataset:
return ray.data.from_arrow(self.to_arrow())
+ def count(self) -> int:
+ """
+ Usage: calutates the total number of records in a Scan that haven't had positional deletes
+ """
+ res = 0
+ # every task is a FileScanTask
+ tasks = self.plan_files()
+
+ for task in tasks:
+ # task.residual is a Boolean Expression if the fiter condition is fully satisfied by the
+ # partition value and task.delete_files represents that positional delete haven't been merged yet
+ # hence those files have to read as a pyarrow table applying the filter and deletes
+ if task.residual == AlwaysTrue() and not len(task.delete_files):
+ # Every File has a metadata stat that stores the file record count
+ res += task.file.record_count
+ else:
+ from pyiceberg.io.pyarrow import ArrowScan
+ tbl = ArrowScan(
+ self.table_metadata, self.io, self.projection(), self.row_filter, self.case_sensitive, self.limit
+ ).to_table([task])
+ res += len(tbl)
+ return res
Review Comment:
I love this approach! My only concern is about loading too much data into memory at once, although this is loading just one file at a time, in the worst case some file could potentially be very large? Shall we define a threshold and check, for example, if `file size < 512MB`, load entire file, otherwise turn it into `pa.RecordBatchReader` and read stream of record batches for counting.
```
target_schema = schema_to_pyarrow(self.projection())
batches = ArrowScan(
self.table_metadata, self.io, self.projection(), self.row_filter, self.case_sensitive, self.limit
).to_record_batches([task])
reader = pa.RecordBatchReader.from_batches(
target_schema,
batches,
)
count = 0
for batch in reader:
count += batch.num_rows
return count
```
https://github.com/apache/iceberg-python/blob/e6465001bd8a47718ff79da4def5800962e6b895/pyiceberg/table/__init__.py#L1557-L1564
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "mrutunjay-kinagi (via GitHub)" <gi...@apache.org>.
mrutunjay-kinagi commented on PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#issuecomment-2651843972
🚀
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "tusharchou (via GitHub)" <gi...@apache.org>.
tusharchou commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1920238317
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1731,214 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression):
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if val is not None:
+ return self.visit_true()
+ else:
+ return self.visit_false()
Review Comment:
This change is causing a type casting problem as `term.eval()` returns a `L` with can be of typre `str, bytes, UUID` with isnan doesn't support
```
pyiceberg/expressions/visitors.py:1799: error: Argument 1 to "isnan" has incompatible type "str"; expected "Union[SupportsFloat, SupportsIndex]" [arg-type]
pyiceberg/expressions/visitors.py:1799: error: Argument 1 to "isnan" has incompatible type "bytes"; expected "Union[SupportsFloat, SupportsIndex]" [arg-type]
pyiceberg/expressions/visitors.py:1799: error: Argument 1 to "isnan" has incompatible type "UUID"; expected "Union[SupportsFloat, SupportsIndex]" [arg-type]
pyiceberg/expressions/visitors.py:1805: error: Argument 1 to "isnan" has incompatible type "str"; expected "Union[SupportsFloat, SupportsIndex]" [arg-type]
pyiceberg/expressions/visitors.py:1805: error: Argument 1 to "isnan" has incompatible type "bytes"; expected "Union[SupportsFloat, SupportsIndex]" [arg-type]
pyiceberg/expressions/visitors.py:1805: error: Argument 1 to "isnan" has incompatible type "UUID"; expected "Union[SupportsFloat, SupportsIndex]" [arg-type]
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1944744615
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1732,230 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ """Finds the residuals for an Expression the partitions in the given PartitionSpec.
+
+ A residual expression is made by partially evaluating an expression using partition values.
+ For example, if a table is partitioned by day(utc_timestamp) and is read with a filter expression
+ utc_timestamp >= a and utc_timestamp <= b, then there are 4 possible residuals expressions
+ for the partition data, d:
+
+
+ 1. If d > day(a) and d < day(b), the residual is always true
+ 2. If d == day(a) and d != day(b), the residual is utc_timestamp >= a
+ 3. if d == day(b) and d != day(a), the residual is utc_timestamp <= b
+ 4. If d == day(a) == day(b), the residual is utc_timestamp >= a and utc_timestamp <= b
+
+ Partition data is passed using StructLike. Residuals are returned by residualFor(StructLike).
+
+ This class is thread-safe.
+ """
+
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+ expr: BooleanExpression
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression) -> None:
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and not math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ def struct_to_schema(struct: StructType) -> Schema:
+ return Schema(*struct.fields)
+
+ for part in parts:
+ strict_projection = part.transform.strict_project(part.name, predicate)
+ strict_result = None
+
+ if strict_projection is not None:
+ bound = strict_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound, BoundPredicate):
+ strict_result = super().visit_bound_predicate(bound)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or alwaysFalse
+ strict_result = bound
+
+ if strict_result is not None and isinstance(strict_result, AlwaysTrue):
+ return AlwaysTrue()
+
+ inclusive_projection = part.transform.project(part.name, predicate)
+ inclusive_result = None
+ if inclusive_projection is not None:
+ bound_inclusive = inclusive_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound_inclusive, BoundPredicate):
+ # using predicate method specific to inclusive
+ inclusive_result = super().visit_bound_predicate(bound_inclusive)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or
+ # alwaysFalse
+ inclusive_result = bound_inclusive
+ if inclusive_result is not None and isinstance(inclusive_result, AlwaysFalse):
+ return AlwaysFalse()
+
+ return predicate
+
+ def visit_unbound_predicate(self, predicate: UnboundPredicate[L]) -> BooleanExpression:
+ bound = predicate.bind(self.schema, case_sensitive=True)
+
+ if isinstance(bound, BoundPredicate):
+ bound_residual = self.visit_bound_predicate(predicate=bound)
+ # if isinstance(bound_residual, BooleanExpression):
+ if bound_residual not in (AlwaysFalse(), AlwaysTrue()):
Review Comment:
```suggestion
if not isinstance(bound_residual, (AlwaysFalse, AlwaysTrue)):
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1944742544
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1732,230 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ """Finds the residuals for an Expression the partitions in the given PartitionSpec.
+
+ A residual expression is made by partially evaluating an expression using partition values.
+ For example, if a table is partitioned by day(utc_timestamp) and is read with a filter expression
+ utc_timestamp >= a and utc_timestamp <= b, then there are 4 possible residuals expressions
+ for the partition data, d:
+
+
+ 1. If d > day(a) and d < day(b), the residual is always true
+ 2. If d == day(a) and d != day(b), the residual is utc_timestamp >= a
+ 3. if d == day(b) and d != day(a), the residual is utc_timestamp <= b
+ 4. If d == day(a) == day(b), the residual is utc_timestamp >= a and utc_timestamp <= b
+
+ Partition data is passed using StructLike. Residuals are returned by residualFor(StructLike).
+
+ This class is thread-safe.
+ """
+
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+ expr: BooleanExpression
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression) -> None:
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and not math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ def struct_to_schema(struct: StructType) -> Schema:
+ return Schema(*struct.fields)
+
+ for part in parts:
+ strict_projection = part.transform.strict_project(part.name, predicate)
+ strict_result = None
+
+ if strict_projection is not None:
+ bound = strict_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound, BoundPredicate):
+ strict_result = super().visit_bound_predicate(bound)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or alwaysFalse
+ strict_result = bound
+
+ if strict_result is not None and isinstance(strict_result, AlwaysTrue):
+ return AlwaysTrue()
+
+ inclusive_projection = part.transform.project(part.name, predicate)
+ inclusive_result = None
+ if inclusive_projection is not None:
+ bound_inclusive = inclusive_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound_inclusive, BoundPredicate):
+ # using predicate method specific to inclusive
+ inclusive_result = super().visit_bound_predicate(bound_inclusive)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or
+ # alwaysFalse
+ inclusive_result = bound_inclusive
+ if inclusive_result is not None and isinstance(inclusive_result, AlwaysFalse):
+ return AlwaysFalse()
+
+ return predicate
+
+ def visit_unbound_predicate(self, predicate: UnboundPredicate[L]) -> BooleanExpression:
+ bound = predicate.bind(self.schema, case_sensitive=True)
Review Comment:
```suggestion
bound = predicate.bind(self.schema, case_sensitive=self.case_sensitive)
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Add ResidualVisitor to compute residuals [iceberg-python]
Posted by "Fokko (via GitHub)" <gi...@apache.org>.
Fokko commented on code in PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#discussion_r1944741812
##########
pyiceberg/expressions/visitors.py:
##########
@@ -1731,3 +1732,230 @@ def _can_contain_nulls(self, field_id: int) -> bool:
def _can_contain_nans(self, field_id: int) -> bool:
return (nan_count := self.nan_counts.get(field_id)) is not None and nan_count > 0
+
+
+class ResidualVisitor(BoundBooleanExpressionVisitor[BooleanExpression], ABC):
+ """Finds the residuals for an Expression the partitions in the given PartitionSpec.
+
+ A residual expression is made by partially evaluating an expression using partition values.
+ For example, if a table is partitioned by day(utc_timestamp) and is read with a filter expression
+ utc_timestamp >= a and utc_timestamp <= b, then there are 4 possible residuals expressions
+ for the partition data, d:
+
+
+ 1. If d > day(a) and d < day(b), the residual is always true
+ 2. If d == day(a) and d != day(b), the residual is utc_timestamp >= a
+ 3. if d == day(b) and d != day(a), the residual is utc_timestamp <= b
+ 4. If d == day(a) == day(b), the residual is utc_timestamp >= a and utc_timestamp <= b
+
+ Partition data is passed using StructLike. Residuals are returned by residualFor(StructLike).
+
+ This class is thread-safe.
+ """
+
+ schema: Schema
+ spec: PartitionSpec
+ case_sensitive: bool
+ expr: BooleanExpression
+
+ def __init__(self, schema: Schema, spec: PartitionSpec, case_sensitive: bool, expr: BooleanExpression) -> None:
+ self.schema = schema
+ self.spec = spec
+ self.case_sensitive = case_sensitive
+ self.expr = expr
+
+ def eval(self, partition_data: Record) -> BooleanExpression:
+ self.struct = partition_data
+ return visit(self.expr, visitor=self)
+
+ def visit_true(self) -> BooleanExpression:
+ return AlwaysTrue()
+
+ def visit_false(self) -> BooleanExpression:
+ return AlwaysFalse()
+
+ def visit_not(self, child_result: BooleanExpression) -> BooleanExpression:
+ return Not(child_result)
+
+ def visit_and(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return And(left_result, right_result)
+
+ def visit_or(self, left_result: BooleanExpression, right_result: BooleanExpression) -> BooleanExpression:
+ return Or(left_result, right_result)
+
+ def visit_is_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_null(self, term: BoundTerm[L]) -> BooleanExpression:
+ if term.eval(self.struct) is not None:
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_is_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_nan(self, term: BoundTerm[L]) -> BooleanExpression:
+ val = term.eval(self.struct)
+ if isinstance(val, SupportsFloat) and not math.isnan(val):
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) < literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_less_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) <= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) > literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_greater_than_or_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) >= literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) == literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_equal(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if term.eval(self.struct) != literal.value:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_not_in(self, term: BoundTerm[L], literals: Set[L]) -> BooleanExpression:
+ if term.eval(self.struct) not in literals:
+ return self.visit_true()
+ else:
+ return self.visit_false()
+
+ def visit_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ eval_res = term.eval(self.struct)
+ if eval_res is not None and str(eval_res).startswith(str(literal.value)):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_not_starts_with(self, term: BoundTerm[L], literal: Literal[L]) -> BooleanExpression:
+ if not self.visit_starts_with(term, literal):
+ return AlwaysTrue()
+ else:
+ return AlwaysFalse()
+
+ def visit_bound_predicate(self, predicate: BoundPredicate[Any]) -> BooleanExpression:
+ """
+ If there is no strict projection or if it evaluates to false, then return the predicate.
+
+ Get the strict projection and inclusive projection of this predicate in partition data,
+ then use them to determine whether to return the original predicate. The strict projection
+ returns true iff the original predicate would have returned true, so the predicate can be
+ eliminated if the strict projection evaluates to true. Similarly the inclusive projection
+ returns false iff the original predicate would have returned false, so the predicate can
+ also be eliminated if the inclusive projection evaluates to false.
+
+ """
+ parts = self.spec.fields_by_source_id(predicate.term.ref().field.field_id)
+ if parts == []:
+ return predicate
+
+ def struct_to_schema(struct: StructType) -> Schema:
+ return Schema(*struct.fields)
+
+ for part in parts:
+ strict_projection = part.transform.strict_project(part.name, predicate)
+ strict_result = None
+
+ if strict_projection is not None:
+ bound = strict_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
+ if isinstance(bound, BoundPredicate):
+ strict_result = super().visit_bound_predicate(bound)
+ else:
+ # if the result is not a predicate, then it must be a constant like alwaysTrue or alwaysFalse
+ strict_result = bound
+
+ if strict_result is not None and isinstance(strict_result, AlwaysTrue):
+ return AlwaysTrue()
+
+ inclusive_projection = part.transform.project(part.name, predicate)
+ inclusive_result = None
+ if inclusive_projection is not None:
+ bound_inclusive = inclusive_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)))
Review Comment:
```suggestion
bound_inclusive = inclusive_projection.bind(struct_to_schema(self.spec.partition_type(self.schema)), case_sensitive=self.case_sensitive)
```
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Count rows as a metadata only operation [iceberg-python]
Posted by "tusharchou (via GitHub)" <gi...@apache.org>.
tusharchou commented on PR #1388:
URL: https://github.com/apache/iceberg-python/pull/1388#issuecomment-2572804832
Hi @Fokko @kevinjqliu @gli-chris-hao ,
I have implemented these suggestions with my best understanding.
- [x] residual evaluator
- [x] positional deletes
- [x] batch processing of files larger than 512mb
It would be helpful to get fresh review
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org