Merge branch 'main' into fd-1-11

Author: Fokko

.github/workflows/codeql.yml

@@ -46,11 +46,11 @@ jobs:
         persist-credentials: false
 
     - name: Initialize CodeQL
-      uses: github/codeql-action/init@e46ed2cbd01164d986452f91f178727624ae40d7 # v4.35.3
+      uses: github/codeql-action/init@9e0d7b8d25671d64c341c19c0152d693099fb5ba # v4.35.5
       with:
         languages: actions
 
     - name: Perform CodeQL Analysis
-      uses: github/codeql-action/analyze@e46ed2cbd01164d986452f91f178727624ae40d7 # v4.35.3
+      uses: github/codeql-action/analyze@9e0d7b8d25671d64c341c19c0152d693099fb5ba # v4.35.5
       with:
         category: "/language:actions"

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

@@ -89,7 +89,7 @@ jobs:
         if: matrix.os == 'ubuntu-latest'
         run: ls -lah dist/* && cp dist/* wheelhouse/
 
-      - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+      - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:
           name: "pypi-release-candidate-${{ matrix.os }}"
           path: ./wheelhouse/*

.github/workflows/python-ci.yml

@@ -101,7 +101,7 @@ jobs:
       if: ${{ failure() }}
       run: docker compose -f dev/docker-compose-integration.yml logs
     - name: Upload coverage data
-      uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+      uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
       with:
         name: coverage-integration
         path: .coverage*
@@ -130,7 +130,7 @@ jobs:
       if: ${{ failure() }}
       run: docker compose -f dev/docker-compose.yml logs
     - name: Upload coverage data
-      uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+      uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
       with:
         name: coverage-s3
         path: .coverage*
@@ -159,7 +159,7 @@ jobs:
       if: ${{ failure() }}
       run: docker compose -f dev/docker-compose-azurite.yml logs
     - name: Upload coverage data
-      uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+      uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
       with:
         name: coverage-adls
         path: .coverage*
@@ -188,7 +188,7 @@ jobs:
       if: ${{ failure() }}
       run: docker compose -f dev/docker-compose-gcs-server.yml logs
     - name: Upload coverage data
-      uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+      uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
       with:
         name: coverage-gcs
         path: .coverage*

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

@@ -80,7 +80,7 @@ jobs:
         if: matrix.os == 'ubuntu-latest'
         run: ls -lah dist/* && cp dist/* wheelhouse/
 
-      - uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+      - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:
           name: "svn-release-candidate-${{ matrix.os }}"
           path: ./wheelhouse/*

Makefile

@@ -16,7 +16,7 @@
 # under the License.
 .PHONY: help install install-uv check-license lint \
         test test-integration test-integration-setup test-integration-exec test-integration-cleanup test-integration-rebuild \
-        test-s3 test-adls test-gcs test-coverage coverage-report \
+        test-s3 test-adls test-gcs test-coverage coverage-report test test-notebook\
         docs-serve docs-build notebook notebook-infra \
         clean
 
@@ -150,6 +150,9 @@ coverage-report: ## Combine and report coverage
 	uv run $(PYTHON_ARG) coverage html
 	uv run $(PYTHON_ARG) coverage xml
 
+test-notebook: ## Run notebook tests (pyiceberg_example and spark_integration_example) via papermill
+	$(TEST_RUNNER) pytest tests/notebooks/test_pyiceberg_example.py tests/notebooks/test_spark_integration_example.py -m notebook $(PYTEST_ARGS)
+
 # ================
 # Documentation
 # ================

mkdocs/docs/api.md

@@ -365,6 +365,17 @@ for buf in tbl.scan().to_arrow_batch_reader():
     print(f"Buffer contains {len(buf)} rows")
 ```
 
+### Streaming writes from a `RecordBatchReader`
+
+`tbl.append()` and `tbl.overwrite()` also accept a `pyarrow.RecordBatchReader` directly, which lets you write datasets that don't fit in memory without materialising them as a `pa.Table` first. PyIceberg consumes the reader once and microbatches it into Parquet files of approximately `write.target-file-size-bytes` (default 512 MiB), keeping memory usage bounded by the target size. All files are committed in a single snapshot.
+
+```python
+reader = pa.RecordBatchReader.from_batches(schema, batch_iter)
+tbl.append(reader)
+```
+
+Streaming writes are currently only supported on **unpartitioned** tables. For a partitioned table, materialise the reader as a `pa.Table` first, or follow [#2152](https://github.com/apache/iceberg-python/issues/2152) for the partitioned support tracked as a follow-up.
+
 To avoid any type inconsistencies during writing, you can convert the Iceberg table schema to Arrow:
 
 ```python
@@ -425,7 +436,7 @@ You can overwrite the record of `Paris` with a record of `New York`:
 from pyiceberg.expressions import EqualTo
 df = pa.Table.from_pylist(
     [
-        {"city": "New York", "lat": 40.7128, "long": 74.0060},
+        {"city": "New York", "lat": 40.7128, "long": -74.0060},
     ]
 )
 tbl.overwrite(df, overwrite_filter=EqualTo('city', "Paris"))
@@ -441,7 +452,7 @@ long: double
 ----
 city: [["New York"],["Amsterdam","San Francisco","Drachten"]]
 lat: [[40.7128],[52.371807,37.773972,53.11254]]
-long: [[74.006],[4.896029,-122.431297,6.0989]]
+long: [[-74.006],[4.896029,-122.431297,6.0989]]
 ```
 
 If the PyIceberg table is partitioned, you can use `tbl.dynamic_partition_overwrite(df)` to replace the existing partitions with new ones provided in the dataframe. The partitions to be replaced are detected automatically from the provided arrow table.
@@ -1529,6 +1540,17 @@ catalog = load_catalog("default")
 catalog.view_exists("default.bar")
 ```
 
+## Register a view
+
+To register a view using existing metadata:
+
+```python
+catalog.register_view(
+    identifier="docs_example.bids",
+    metadata_location="s3://warehouse/path/to/metadata.json"
+)
+```
+
 ## Table Statistics Management
 
 Manage table statistics with operations through the `Table` API:

pyiceberg/catalog/__init__.py

@@ -31,6 +31,8 @@
     cast,
 )
 
+from typing_extensions import override
+
 from pyiceberg.exceptions import (
     NamespaceAlreadyExistsError,
     NoSuchNamespaceError,
@@ -690,6 +692,22 @@ def update_namespace_properties(
             ValueError: If removals and updates have overlapping keys.
         """
 
+    @abstractmethod
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        """Register a new view using existing metadata.
+
+        Args:
+            identifier (Union[str, Identifier]): View identifier for the view
+            metadata_location (str): The location to the metadata
+
+        Returns:
+            View: The newly registered view
+
+        Raises:
+            ViewAlreadyExistsError: If the view already exists.
+            TableAlreadyExistsError: If a table with the same name already exists.
+        """
+
     @abstractmethod
     def drop_view(self, identifier: str | Identifier) -> None:
         """Drop a view.
@@ -888,9 +906,11 @@ class MetastoreCatalog(Catalog, ABC):
     def __init__(self, name: str, **properties: str):
         super().__init__(name, **properties)
 
+    @override
     def supports_server_side_planning(self) -> bool:
         return False
 
+    @override
     def create_table_transaction(
         self,
         identifier: str | Identifier,
@@ -904,13 +924,15 @@ def create_table_transaction(
             self._create_staged_table(identifier, schema, location, partition_spec, sort_order, properties)
         )
 
+    @override
     def table_exists(self, identifier: str | Identifier) -> bool:
         try:
             self.load_table(identifier)
             return True
         except NoSuchTableError:
             return False
 
+    @override
     def namespace_exists(self, namespace: str | Identifier) -> bool:
         """Check if a namespace exists.
 
@@ -926,6 +948,7 @@ def namespace_exists(self, namespace: str | Identifier) -> bool:
         except NoSuchNamespaceError:
             return False
 
+    @override
     def purge_table(self, identifier: str | Identifier) -> None:
         table = self.load_table(identifier)
         self.drop_table(identifier)
@@ -946,6 +969,7 @@ def purge_table(self, identifier: str | Identifier) -> None:
         delete_files(io, prev_metadata_files, PREVIOUS_METADATA)
         delete_files(io, {table.metadata_location}, METADATA)
 
+    @override
     def create_view(
         self,
         identifier: str | Identifier,

pyiceberg/catalog/bigquery_metastore.py

@@ -26,6 +26,7 @@
 from google.cloud.bigquery.schema import SerDeInfo, StorageDescriptor
 from google.cloud.exceptions import Conflict
 from google.oauth2 import service_account
+from typing_extensions import override
 
 from pyiceberg.catalog import WAREHOUSE_LOCATION, MetastoreCatalog, PropertiesUpdateSummary
 from pyiceberg.exceptions import NamespaceAlreadyExistsError, NoSuchNamespaceError, NoSuchTableError, TableAlreadyExistsError
@@ -101,6 +102,7 @@ def __init__(self, name: str, **properties: str):
         self.location = location
         self.project_id = project_id
 
+    @override
     def create_table(
         self,
         identifier: str | Identifier,
@@ -156,6 +158,7 @@ def create_table(
 
         return self.load_table(identifier=identifier)
 
+    @override
     def create_namespace(self, namespace: str | Identifier, properties: Properties = EMPTY_DICT) -> None:
         """Create a namespace in the catalog.
 
@@ -177,6 +180,7 @@ def create_namespace(self, namespace: str | Identifier, properties: Properties =
         except Conflict as e:
             raise NamespaceAlreadyExistsError("Namespace {database_name} already exists") from e
 
+    @override
     def load_table(self, identifier: str | Identifier) -> Table:
         """
         Load the table's metadata and returns the table instance.
@@ -205,6 +209,7 @@ def load_table(self, identifier: str | Identifier) -> Table:
         except NotFound as e:
             raise NoSuchTableError(f"Table does not exist: {dataset_name}.{table_name}") from e
 
+    @override
     def drop_table(self, identifier: str | Identifier) -> None:
         """Drop a table.
 
@@ -225,14 +230,17 @@ def drop_table(self, identifier: str | Identifier) -> None:
         except NoSuchTableError as e:
             raise NoSuchTableError(f"Table does not exist: {dataset_name}.{table_name}") from e
 
+    @override
     def commit_table(
         self, table: Table, requirements: tuple[TableRequirement, ...], updates: tuple[TableUpdate, ...]
     ) -> CommitTableResponse:
         raise NotImplementedError
 
+    @override
     def rename_table(self, from_identifier: str | Identifier, to_identifier: str | Identifier) -> Table:
         raise NotImplementedError
 
+    @override
     def drop_namespace(self, namespace: str | Identifier) -> None:
         database_name = self.identifier_to_database(namespace)
 
@@ -243,6 +251,7 @@ def drop_namespace(self, namespace: str | Identifier) -> None:
         except NotFound as e:
             raise NoSuchNamespaceError(f"Namespace {namespace} does not exist.") from e
 
+    @override
     def list_tables(self, namespace: str | Identifier) -> list[Identifier]:
         database_name = self.identifier_to_database(namespace)
         iceberg_tables: list[Identifier] = []
@@ -257,6 +266,7 @@ def list_tables(self, namespace: str | Identifier) -> list[Identifier]:
             raise NoSuchNamespaceError(f"Namespace (dataset) '{database_name}' not found.") from None
         return iceberg_tables
 
+    @override
     def list_namespaces(self, namespace: str | Identifier = ()) -> list[Identifier]:
         # Since this catalog only supports one-level namespaces, it always returns an empty list unless
         # passed an empty namespace to list all namespaces within the catalog.
@@ -267,6 +277,7 @@ def list_namespaces(self, namespace: str | Identifier = ()) -> list[Identifier]:
         datasets_iterator = self.client.list_datasets()
         return [(dataset.dataset_id,) for dataset in datasets_iterator]
 
+    @override
     def register_table(self, identifier: str | Identifier, metadata_location: str, overwrite: bool = False) -> Table:
         """Register a new table using existing metadata.
 
@@ -302,18 +313,27 @@ def register_table(self, identifier: str | Identifier, metadata_location: str, o
 
         return self.load_table(identifier=identifier)
 
+    @override
     def list_views(self, namespace: str | Identifier) -> list[Identifier]:
         raise NotImplementedError
 
+    @override
+    def register_view(self, identifier: str | Identifier, metadata_location: str) -> View:
+        raise NotImplementedError
+
+    @override
     def drop_view(self, identifier: str | Identifier) -> None:
         raise NotImplementedError
 
+    @override
     def view_exists(self, identifier: str | Identifier) -> bool:
         raise NotImplementedError
 
+    @override
     def load_view(self, identifier: str | Identifier) -> View:
         raise NotImplementedError
 
+    @override
     def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
         dataset_name = self.identifier_to_database(namespace)
 
@@ -326,6 +346,7 @@ def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
             raise NoSuchNamespaceError(f"Namespace {namespace} not found") from e
         return {}
 
+    @override
     def update_namespace_properties(
         self, namespace: str | Identifier, removals: set[str] | None = None, updates: Properties = EMPTY_DICT
     ) -> PropertiesUpdateSummary: