Merge branch 'main' into drop-python-39

Author: Aniketsy

.github/workflows/pypi-build-artifacts.yml

@@ -61,7 +61,7 @@ jobs:
         if: startsWith(matrix.os, 'ubuntu')
 
       - name: Build wheels
-        uses: pypa/cibuildwheel@v3.2.0
+        uses: pypa/cibuildwheel@v3.2.1
         with:
           output-dir: wheelhouse
           config-file: "pyproject.toml"

.github/workflows/svn-build-artifacts.yml

@@ -56,7 +56,7 @@ jobs:
         if: startsWith(matrix.os, 'ubuntu')
 
       - name: Build wheels
-        uses: pypa/cibuildwheel@v3.2.0
+        uses: pypa/cibuildwheel@v3.2.1
         with:
           output-dir: wheelhouse
           config-file: "pyproject.toml"

mkdocs/docs/configuration.md

@@ -539,6 +539,18 @@ catalog:
     py-io-impl: pyiceberg.io.fsspec.FsspecFileIO
 ```
 
+##### Apache Gravitino
+
+```yaml
+catalog:
+  gravitino_catalog:
+    type: rest
+    uri: <gravitino-catalog-uri>
+    header.X-Iceberg-Access-Delegation: vended-credentials
+    auth:
+      type: noop
+```
+
 ### SQL Catalog
 
 The SQL catalog requires a database for its backend. PyIceberg supports PostgreSQL and SQLite through psycopg2. The database connection has to be configured using the `uri` property. The init_catalog_tables is optional and defaults to True. If it is set to False, the catalog tables will not be created when the SQLCatalog is initialized. See SQLAlchemy's [documentation for URL format](https://docs.sqlalchemy.org/en/20/core/engines.html#backend-specific-urls):

poetry.lock

@@ -743,12 +743,18 @@ def as_bound(self) -> Type[BoundNotIn[L]]:
         return BoundNotIn[L]
 
 
-class LiteralPredicate(UnboundPredicate[L], ABC):
-    literal: Literal[L]
+class LiteralPredicate(IcebergBaseModel, UnboundPredicate[L], ABC):
+    type: TypingLiteral["lt", "lt-eq", "gt", "gt-eq", "eq", "not-eq", "starts-with", "not-starts-with"] = Field(alias="type")
+    term: UnboundTerm[Any]
+    value: Literal[L] = Field()
+    model_config = ConfigDict(populate_by_name=True, frozen=True, arbitrary_types_allowed=True)
 
-    def __init__(self, term: Union[str, UnboundTerm[Any]], literal: Union[L, Literal[L]]):  # pylint: disable=W0621
-        super().__init__(term)
-        self.literal = _to_literal(literal)  # pylint: disable=W0621
+    def __init__(self, term: Union[str, UnboundTerm[Any]], literal: Union[L, Literal[L]]):
+        super().__init__(term=_to_unbound_term(term), value=_to_literal(literal))  # type: ignore[call-arg]
+
+    @property
+    def literal(self) -> Literal[L]:
+        return self.value
 
     def bind(self, schema: Schema, case_sensitive: bool = True) -> BoundLiteralPredicate[L]:
         bound_term = self.term.bind(schema, case_sensitive)
@@ -773,6 +779,10 @@ def __eq__(self, other: Any) -> bool:
             return self.term == other.term and self.literal == other.literal
         return False
 
+    def __str__(self) -> str:
+        """Return the string representation of the LiteralPredicate class."""
+        return f"{str(self.__class__.__name__)}(term={repr(self.term)}, literal={repr(self.literal)})"
+
     def __repr__(self) -> str:
         """Return the string representation of the LiteralPredicate class."""
         return f"{str(self.__class__.__name__)}(term={repr(self.term)}, literal={repr(self.literal)})"
@@ -886,6 +896,8 @@ def as_unbound(self) -> Type[NotStartsWith[L]]:
 
 
 class EqualTo(LiteralPredicate[L]):
+    type: TypingLiteral["eq"] = Field(default="eq", alias="type")
+
     def __invert__(self) -> NotEqualTo[L]:
         """Transform the Expression into its negated version."""
         return NotEqualTo[L](self.term, self.literal)
@@ -896,6 +908,8 @@ def as_bound(self) -> Type[BoundEqualTo[L]]:
 
 
 class NotEqualTo(LiteralPredicate[L]):
+    type: TypingLiteral["not-eq"] = Field(default="not-eq", alias="type")
+
     def __invert__(self) -> EqualTo[L]:
         """Transform the Expression into its negated version."""
         return EqualTo[L](self.term, self.literal)
@@ -906,6 +920,8 @@ def as_bound(self) -> Type[BoundNotEqualTo[L]]:
 
 
 class LessThan(LiteralPredicate[L]):
+    type: TypingLiteral["lt"] = Field(default="lt", alias="type")
+
     def __invert__(self) -> GreaterThanOrEqual[L]:
         """Transform the Expression into its negated version."""
         return GreaterThanOrEqual[L](self.term, self.literal)
@@ -916,6 +932,8 @@ def as_bound(self) -> Type[BoundLessThan[L]]:
 
 
 class GreaterThanOrEqual(LiteralPredicate[L]):
+    type: TypingLiteral["gt-eq"] = Field(default="gt-eq", alias="type")
+
     def __invert__(self) -> LessThan[L]:
         """Transform the Expression into its negated version."""
         return LessThan[L](self.term, self.literal)
@@ -926,6 +944,8 @@ def as_bound(self) -> Type[BoundGreaterThanOrEqual[L]]:
 
 
 class GreaterThan(LiteralPredicate[L]):
+    type: TypingLiteral["gt"] = Field(default="gt", alias="type")
+
     def __invert__(self) -> LessThanOrEqual[L]:
         """Transform the Expression into its negated version."""
         return LessThanOrEqual[L](self.term, self.literal)
@@ -936,6 +956,8 @@ def as_bound(self) -> Type[BoundGreaterThan[L]]:
 
 
 class LessThanOrEqual(LiteralPredicate[L]):
+    type: TypingLiteral["lt-eq"] = Field(default="lt-eq", alias="type")
+
     def __invert__(self) -> GreaterThan[L]:
         """Transform the Expression into its negated version."""
         return GreaterThan[L](self.term, self.literal)
@@ -946,6 +968,8 @@ def as_bound(self) -> Type[BoundLessThanOrEqual[L]]:
 
 
 class StartsWith(LiteralPredicate[L]):
+    type: TypingLiteral["starts-with"] = Field(default="starts-with", alias="type")
+
     def __invert__(self) -> NotStartsWith[L]:
         """Transform the Expression into its negated version."""
         return NotStartsWith[L](self.term, self.literal)
@@ -956,6 +980,8 @@ def as_bound(self) -> Type[BoundStartsWith[L]]:
 
 
 class NotStartsWith(LiteralPredicate[L]):
+    type: TypingLiteral["not-starts-with"] = Field(default="not-starts-with", alias="type")
+
     def __invert__(self) -> StartsWith[L]:
         """Transform the Expression into its negated version."""
         return StartsWith[L](self.term, self.literal)

pyiceberg/expressions/__init__.py

@@ -31,6 +31,7 @@
     Callable,
     Dict,
     Iterable,
+    Iterator,
     List,
     Optional,
     Set,
@@ -1942,11 +1943,11 @@ def _check_sequence_number(min_sequence_number: int, manifest: ManifestFile) ->
             and (manifest.sequence_number or INITIAL_SEQUENCE_NUMBER) >= min_sequence_number
         )
 
-    def plan_files(self) -> Iterable[FileScanTask]:
-        """Plans the relevant files by filtering on the PartitionSpecs.
+    def scan_plan_helper(self) -> Iterator[List[ManifestEntry]]:
+        """Filter and return manifest entries based on partition and metrics evaluators.
 
         Returns:
-            List of FileScanTasks that contain both data and delete files.
+            Iterator of ManifestEntry objects that match the scan's partition filter.
         """
         snapshot = self.snapshot()
         if not snapshot:
@@ -1957,8 +1958,6 @@ def plan_files(self) -> Iterable[FileScanTask]:
 
         manifest_evaluators: Dict[int, Callable[[ManifestFile], bool]] = KeyDefaultDict(self._build_manifest_evaluator)
 
-        residual_evaluators: Dict[int, Callable[[DataFile], ResidualEvaluator]] = KeyDefaultDict(self._build_residual_evaluator)
-
         manifests = [
             manifest_file
             for manifest_file in snapshot.manifests(self.io)
@@ -1972,25 +1971,34 @@ def plan_files(self) -> Iterable[FileScanTask]:
 
         min_sequence_number = _min_sequence_number(manifests)
 
+        executor = ExecutorFactory.get_or_create()
+
+        return executor.map(
+            lambda args: _open_manifest(*args),
+            [
+                (
+                    self.io,
+                    manifest,
+                    partition_evaluators[manifest.partition_spec_id],
+                    self._build_metrics_evaluator(),
+                )
+                for manifest in manifests
+                if self._check_sequence_number(min_sequence_number, manifest)
+            ],
+        )
+
+    def plan_files(self) -> Iterable[FileScanTask]:
+        """Plans the relevant files by filtering on the PartitionSpecs.
+
+        Returns:
+            List of FileScanTasks that contain both data and delete files.
+        """
         data_entries: List[ManifestEntry] = []
         positional_delete_entries = SortedList(key=lambda entry: entry.sequence_number or INITIAL_SEQUENCE_NUMBER)
 
-        executor = ExecutorFactory.get_or_create()
-        for manifest_entry in chain(
-            *executor.map(
-                lambda args: _open_manifest(*args),
-                [
-                    (
-                        self.io,
-                        manifest,
-                        partition_evaluators[manifest.partition_spec_id],
-                        self._build_metrics_evaluator(),
-                    )
-                    for manifest in manifests
-                    if self._check_sequence_number(min_sequence_number, manifest)
-                ],
-            )
-        ):
+        residual_evaluators: Dict[int, Callable[[DataFile], ResidualEvaluator]] = KeyDefaultDict(self._build_residual_evaluator)
+
+        for manifest_entry in chain.from_iterable(self.scan_plan_helper()):
             data_file = manifest_entry.data_file
             if data_file.content == DataFileContent.DATA:
                 data_entries.append(manifest_entry)