Merge branch 'main' of github.com:apache/iceberg-python into fd-iceberg-python

Author: Fokko

dev/Dockerfile

@@ -36,7 +36,7 @@ ENV PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/lib/py4j-0.10.9.7-src.zip:$
 RUN mkdir -p ${HADOOP_HOME} && mkdir -p ${SPARK_HOME} && mkdir -p /home/iceberg/spark-events
 WORKDIR ${SPARK_HOME}
 
-ENV SPARK_VERSION=3.4.1
+ENV SPARK_VERSION=3.4.2
 ENV ICEBERG_SPARK_RUNTIME_VERSION=3.4_2.12
 ENV ICEBERG_VERSION=1.4.0
 ENV AWS_SDK_VERSION=2.20.18

mkdocs/docs/configuration.md

@@ -32,13 +32,19 @@ There are three ways to pass in configuration:
 - Through environment variables
 - By passing in credentials through the CLI or the Python API
 
-The configuration file is recommended since that's the most transparent way. If you prefer environment configuration:
+The configuration file is recommended since that's the easiest way to manage the credentials.
+
+Another option is through environment variables:
 
 ```sh
 export PYICEBERG_CATALOG__DEFAULT__URI=thrift://localhost:9083
+export PYICEBERG_CATALOG__DEFAULT__S3__ACCESS_KEY_ID=username
+export PYICEBERG_CATALOG__DEFAULT__S3__SECRET_ACCESS_KEY=password
 ```
 
-The environment variable picked up by Iceberg starts with `PYICEBERG_` and then follows the yaml structure below, where a double underscore `__` represents a nested field.
+The environment variable picked up by Iceberg starts with `PYICEBERG_` and then follows the yaml structure below, where a double underscore `__` represents a nested field, and the underscore `_` is converted into a dash `-`.
+
+For example, `PYICEBERG_CATALOG__DEFAULT__S3__ACCESS_KEY_ID`, sets `s3.access-key-id` on the `default` catalog.
 
 ## FileIO
 
@@ -80,8 +86,6 @@ For the FileIO there are several configuration options available:
 
 ### Azure Data lake
 
-### Azure Data lake
-
 | Key                     | Example                                                                                   | Description                                                                                                                                                                                                                                                                            |
 | ----------------------- | ----------------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | adlfs.connection-string | AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqF...;BlobEndpoint=http://localhost/ | A [connection string](https://learn.microsoft.com/en-us/azure/storage/common/storage-configure-connection-string). This could be used to use FileIO with any adlfs-compatible object storage service that has a different endpoint (like [azurite](https://github.com/azure/azurite)). |

mkdocs/requirements.txt

@@ -23,6 +23,6 @@ mkdocstrings-python==1.7.5
 mkdocs-literate-nav==0.6.1
 mkdocs-autorefs==0.5.0
 mkdocs-gen-files==0.5.0
-mkdocs-material==9.4.10
+mkdocs-material==9.4.14
 mkdocs-material-extensions==1.3.1
 mkdocs-section-index==0.3.8

poetry.lock

@@ -536,6 +536,20 @@ def identifier_to_database_and_table(
 
         return tuple_identifier[0], tuple_identifier[1]
 
+    def identifier_to_tuple_without_catalog(self, identifier: Union[str, Identifier]) -> Identifier:
+        """Convert an identifier to a tuple and drop this catalog's name from the first element.
+
+        Args:
+            identifier (str | Identifier): Table identifier.
+
+        Returns:
+            Identifier: a tuple of strings with this catalog's name removed
+        """
+        identifier_tuple = Catalog.identifier_to_tuple(identifier)
+        if len(identifier_tuple) >= 3 and identifier_tuple[0] == self.name:
+            identifier_tuple = identifier_tuple[1:]
+        return identifier_tuple
+
     def purge_table(self, identifier: Union[str, Identifier]) -> None:
         """Drop a table and purge all data and metadata files.
 
@@ -547,8 +561,9 @@ def purge_table(self, identifier: Union[str, Identifier]) -> None:
         Raises:
             NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
         """
-        table = self.load_table(identifier)
-        self.drop_table(identifier)
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
+        table = self.load_table(identifier_tuple)
+        self.drop_table(identifier_tuple)
         io = load_file_io(self.properties, table.metadata_location)
         metadata = table.metadata
         manifest_lists_to_delete = set()

pyiceberg/catalog/__init__.py

@@ -213,7 +213,8 @@ def load_table(self, identifier: Union[str, Identifier]) -> Table:
         Raises:
             NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
         """
-        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
+        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
         dynamo_table_item = self._get_iceberg_table_item(database_name=database_name, table_name=table_name)
         return self._convert_dynamo_table_item_to_iceberg_table(dynamo_table_item=dynamo_table_item)
 
@@ -226,7 +227,8 @@ def drop_table(self, identifier: Union[str, Identifier]) -> None:
         Raises:
             NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
         """
-        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
+        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
 
         try:
             self._delete_dynamo_item(
@@ -256,7 +258,8 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             NoSuchPropertyException: When from table miss some required properties.
             NoSuchNamespaceError: When the destination namespace doesn't exist.
         """
-        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier, NoSuchTableError)
+        from_identifier_tuple = self.identifier_to_tuple_without_catalog(from_identifier)
+        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier_tuple, NoSuchTableError)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
 
         from_table_item = self._get_iceberg_table_item(database_name=from_database_name, table_name=from_table_name)
@@ -287,7 +290,7 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             raise TableAlreadyExistsError(f"Table {to_database_name}.{to_table_name} already exists") from e
 
         try:
-            self.drop_table(from_identifier)
+            self.drop_table(from_identifier_tuple)
         except (NoSuchTableError, GenericDynamoDbError) as e:
             log_message = f"Failed to drop old table {from_database_name}.{from_table_name}. "
 

pyiceberg/catalog/dynamodb.py

@@ -265,7 +265,8 @@ def load_table(self, identifier: Union[str, Identifier]) -> Table:
         Raises:
             NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
         """
-        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
+        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
         try:
             load_table_response = self.glue.get_table(DatabaseName=database_name, Name=table_name)
         except self.glue.exceptions.EntityNotFoundException as e:
@@ -282,7 +283,8 @@ def drop_table(self, identifier: Union[str, Identifier]) -> None:
         Raises:
             NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
         """
-        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
+        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
         try:
             self.glue.delete_table(DatabaseName=database_name, Name=table_name)
         except self.glue.exceptions.EntityNotFoundException as e:
@@ -307,7 +309,8 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             NoSuchPropertyException: When from table miss some required properties.
             NoSuchNamespaceError: When the destination namespace doesn't exist.
         """
-        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier, NoSuchTableError)
+        from_identifier_tuple = self.identifier_to_tuple_without_catalog(from_identifier)
+        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier_tuple, NoSuchTableError)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
         try:
             get_table_response = self.glue.get_table(DatabaseName=from_database_name, Name=from_table_name)

pyiceberg/catalog/glue.py

@@ -347,7 +347,8 @@ def load_table(self, identifier: Union[str, Identifier]) -> Table:
         Raises:
             NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
         """
-        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
+        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
         try:
             with self._client as open_client:
                 hive_table = open_client.get_table(dbname=database_name, tbl_name=table_name)
@@ -366,7 +367,8 @@ def drop_table(self, identifier: Union[str, Identifier]) -> None:
         Raises:
             NoSuchTableError: If a table with the name does not exist, or the identifier is invalid.
         """
-        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
+        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
         try:
             with self._client as open_client:
                 open_client.drop_table(dbname=database_name, name=table_name, deleteData=False)
@@ -393,7 +395,8 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             NoSuchTableError: When a table with the name does not exist.
             NoSuchNamespaceError: When the destination namespace doesn't exist.
         """
-        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier, NoSuchTableError)
+        from_identifier_tuple = self.identifier_to_tuple_without_catalog(from_identifier)
+        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier_tuple, NoSuchTableError)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
         try:
             with self._client as open_client:

pyiceberg/catalog/hive.py

@@ -302,19 +302,20 @@ def _fetch_config(self) -> None:
         # Update URI based on overrides
         self.uri = config[URI]
 
-    def _split_identifier_for_path(self, identifier: Union[str, Identifier, TableIdentifier]) -> Properties:
-        if isinstance(identifier, TableIdentifier):
-            return {"namespace": NAMESPACE_SEPARATOR.join(identifier.namespace.root[1:]), "table": identifier.name}
-
+    def _identifier_to_validated_tuple(self, identifier: Union[str, Identifier]) -> Identifier:
         identifier_tuple = self.identifier_to_tuple(identifier)
         if len(identifier_tuple) <= 1:
             raise NoSuchTableError(f"Missing namespace or invalid identifier: {'.'.join(identifier_tuple)}")
+        return identifier_tuple
+
+    def _split_identifier_for_path(self, identifier: Union[str, Identifier, TableIdentifier]) -> Properties:
+        if isinstance(identifier, TableIdentifier):
+            return {"namespace": NAMESPACE_SEPARATOR.join(identifier.namespace.root[1:]), "table": identifier.name}
+        identifier_tuple = self._identifier_to_validated_tuple(identifier)
         return {"namespace": NAMESPACE_SEPARATOR.join(identifier_tuple[:-1]), "table": identifier_tuple[-1]}
 
     def _split_identifier_for_json(self, identifier: Union[str, Identifier]) -> Dict[str, Union[Identifier, str]]:
-        identifier_tuple = self.identifier_to_tuple(identifier)
-        if len(identifier_tuple) <= 1:
-            raise NoSuchTableError(f"Missing namespace or invalid identifier: {identifier_tuple}")
+        identifier_tuple = self._identifier_to_validated_tuple(identifier)
         return {"namespace": identifier_tuple[:-1], "name": identifier_tuple[-1]}
 
     def _handle_non_200_response(self, exc: HTTPError, error_handler: Dict[int, Type[Exception]]) -> None:
@@ -499,12 +500,10 @@ def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
         return [(*table.namespace, table.name) for table in ListTablesResponse(**response.json()).identifiers]
 
     def load_table(self, identifier: Union[str, Identifier]) -> Table:
-        identifier_tuple = self.identifier_to_tuple(identifier)
-
-        if len(identifier_tuple) <= 1:
-            raise NoSuchTableError(f"Missing namespace or invalid identifier: {identifier}")
-
-        response = self._session.get(self.url(Endpoints.load_table, prefixed=True, **self._split_identifier_for_path(identifier)))
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
+        response = self._session.get(
+            self.url(Endpoints.load_table, prefixed=True, **self._split_identifier_for_path(identifier_tuple))
+        )
         try:
             response.raise_for_status()
         except HTTPError as exc:
@@ -514,8 +513,11 @@ def load_table(self, identifier: Union[str, Identifier]) -> Table:
         return self._response_to_table(identifier_tuple, table_response)
 
     def drop_table(self, identifier: Union[str, Identifier], purge_requested: bool = False) -> None:
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
         response = self._session.delete(
-            self.url(Endpoints.drop_table, prefixed=True, purge=purge_requested, **self._split_identifier_for_path(identifier)),
+            self.url(
+                Endpoints.drop_table, prefixed=True, purge=purge_requested, **self._split_identifier_for_path(identifier_tuple)
+            ),
         )
         try:
             response.raise_for_status()
@@ -526,8 +528,9 @@ def purge_table(self, identifier: Union[str, Identifier]) -> None:
         self.drop_table(identifier=identifier, purge_requested=True)
 
     def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: Union[str, Identifier]) -> Table:
+        from_identifier_tuple = self.identifier_to_tuple_without_catalog(from_identifier)
         payload = {
-            "source": self._split_identifier_for_json(from_identifier),
+            "source": self._split_identifier_for_json(from_identifier_tuple),
             "destination": self._split_identifier_for_json(to_identifier),
         }
         response = self._session.post(self.url(Endpoints.rename_table), json=payload)

pyiceberg/catalog/rest.py

@@ -231,7 +231,8 @@ def load_table(self, identifier: Union[str, Identifier]) -> Table:
         Raises:
             NoSuchTableError: If a table with the name does not exist.
         """
-        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
+        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
         with Session(self.engine) as session:
             stmt = select(IcebergTables).where(
                 IcebergTables.catalog_name == self.name,
@@ -252,7 +253,8 @@ def drop_table(self, identifier: Union[str, Identifier]) -> None:
         Raises:
             NoSuchTableError: If a table with the name does not exist.
         """
-        database_name, table_name = self.identifier_to_database_and_table(identifier, NoSuchTableError)
+        identifier_tuple = self.identifier_to_tuple_without_catalog(identifier)
+        database_name, table_name = self.identifier_to_database_and_table(identifier_tuple, NoSuchTableError)
         with Session(self.engine) as session:
             res = session.execute(
                 delete(IcebergTables).where(
@@ -280,7 +282,8 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             TableAlreadyExistsError: If a table with the new name already exist.
             NoSuchNamespaceError: If the target namespace does not exist.
         """
-        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier, NoSuchTableError)
+        from_identifier_tuple = self.identifier_to_tuple_without_catalog(from_identifier)
+        from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier_tuple, NoSuchTableError)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
         if not self._namespace_exists(to_database_name):
             raise NoSuchNamespaceError(f"Namespace does not exist: {to_database_name}")