def visit_bound_predicate()

in pyiceberg/expressions/visitors.py [0:0]

• 32 lines of code
• 13 McCabe index (conditional complexity)

    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)), case_sensitive=self.case_sensitive
                )
                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 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)), case_sensitive=self.case_sensitive
                )
                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 isinstance(inclusive_result, AlwaysFalse):
                return AlwaysFalse()

        return predicate