Merge branch 'main' into feat/rest-catalog-pagination

Author: jayceslesar

.pre-commit-config.yaml

@@ -27,13 +27,13 @@ repos:
       - id: check-yaml
       - id: check-ast
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.8.6
+    rev: v0.11.13
     hooks:
       - id: ruff
         args: [ --fix, --exit-non-zero-on-fix ]
       - id: ruff-format
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.14.1
+    rev: v1.16.0
     hooks:
       - id: mypy
         args:

pyiceberg/catalog/__init__.py

@@ -262,6 +262,10 @@ def load_catalog(name: Optional[str] = None, **properties: Optional[str]) -> Cat
     raise ValueError(f"Could not initialize catalog with the following properties: {properties}")
 
 
+def list_catalogs() -> List[str]:
+    return _ENV_CONFIG.get_known_catalogs()
+
+
 def delete_files(io: FileIO, files_to_delete: Set[str], file_type: str) -> None:
     """Delete files.
 
@@ -923,6 +927,20 @@ def _resolve_table_location(self, location: Optional[str], database_name: str, t
         return location.rstrip("/")
 
     def _get_default_warehouse_location(self, database_name: str, table_name: str) -> str:
+        """Return the default warehouse location using the convention of `warehousePath/databaseName/tableName`."""
+        database_properties = self.load_namespace_properties(database_name)
+        if database_location := database_properties.get(LOCATION):
+            database_location = database_location.rstrip("/")
+            return f"{database_location}/{table_name}"
+
+        if warehouse_path := self.properties.get(WAREHOUSE_LOCATION):
+            warehouse_path = warehouse_path.rstrip("/")
+            return f"{warehouse_path}/{database_name}/{table_name}"
+
+        raise ValueError("No default path is set, please specify a location when creating a table")
+
+    def _get_hive_style_warehouse_location(self, database_name: str, table_name: str) -> str:
+        """Return the default warehouse location following the Hive convention of `warehousePath/databaseName.db/tableName`."""
         database_properties = self.load_namespace_properties(database_name)
         if database_location := database_properties.get(LOCATION):
             database_location = database_location.rstrip("/")

pyiceberg/catalog/dynamodb.py

@@ -664,6 +664,10 @@ def _convert_dynamo_table_item_to_iceberg_table(self, dynamo_table_item: Dict[st
             catalog=self,
         )
 
+    def _get_default_warehouse_location(self, database_name: str, table_name: str) -> str:
+        """Override the default warehouse location to follow Hive-style conventions."""
+        return self._get_hive_style_warehouse_location(database_name, table_name)
+
 
 def _get_create_table_item(database_name: str, table_name: str, properties: Properties, metadata_location: str) -> Dict[str, Any]:
     current_timestamp_ms = str(round(time() * 1000))

pyiceberg/catalog/glue.py

@@ -821,3 +821,7 @@ def view_exists(self, identifier: Union[str, Identifier]) -> bool:
     @staticmethod
     def __is_iceberg_table(table: "TableTypeDef") -> bool:
         return table.get("Parameters", {}).get(TABLE_TYPE, "").lower() == ICEBERG
+
+    def _get_default_warehouse_location(self, database_name: str, table_name: str) -> str:
+        """Override the default warehouse location to follow Hive-style conventions."""
+        return self._get_hive_style_warehouse_location(database_name, table_name)

pyiceberg/catalog/hive.py

@@ -211,11 +211,18 @@ def _construct_hive_storage_descriptor(
 DEFAULT_PROPERTIES = {TableProperties.PARQUET_COMPRESSION: TableProperties.PARQUET_COMPRESSION_DEFAULT}
 
 
-def _construct_parameters(metadata_location: str, previous_metadata_location: Optional[str] = None) -> Dict[str, Any]:
+def _construct_parameters(
+    metadata_location: str, previous_metadata_location: Optional[str] = None, metadata_properties: Optional[Properties] = None
+) -> Dict[str, Any]:
     properties = {PROP_EXTERNAL: "TRUE", PROP_TABLE_TYPE: "ICEBERG", PROP_METADATA_LOCATION: metadata_location}
     if previous_metadata_location:
         properties[PROP_PREVIOUS_METADATA_LOCATION] = previous_metadata_location
 
+    if metadata_properties:
+        for key, value in metadata_properties.items():
+            if key not in properties:
+                properties[key] = str(value)
+
     return properties
 
 
@@ -360,7 +367,7 @@ def _convert_iceberg_into_hive(self, table: Table) -> HiveTable:
                 property_as_bool(self.properties, HIVE2_COMPATIBLE, HIVE2_COMPATIBLE_DEFAULT),
             ),
             tableType=EXTERNAL_TABLE,
-            parameters=_construct_parameters(table.metadata_location),
+            parameters=_construct_parameters(metadata_location=table.metadata_location, metadata_properties=table.properties),
         )
 
     def _create_hive_table(self, open_client: Client, hive_table: HiveTable) -> None:
@@ -541,6 +548,7 @@ def commit_table(
                     hive_table.parameters = _construct_parameters(
                         metadata_location=updated_staged_table.metadata_location,
                         previous_metadata_location=current_table.metadata_location,
+                        metadata_properties=updated_staged_table.properties,
                     )
                     open_client.alter_table_with_environment_context(
                         dbname=database_name,
@@ -790,3 +798,7 @@ def update_namespace_properties(
 
     def drop_view(self, identifier: Union[str, Identifier]) -> None:
         raise NotImplementedError
+
+    def _get_default_warehouse_location(self, database_name: str, table_name: str) -> str:
+        """Override the default warehouse location to follow Hive-style conventions."""
+        return self._get_hive_style_warehouse_location(database_name, table_name)

pyiceberg/expressions/literals.py

@@ -262,7 +262,7 @@ def __init__(self, value: bool) -> None:
         super().__init__(value, bool)
 
     @singledispatchmethod
-    def to(self, type_var: IcebergType) -> Literal[bool]:  # type: ignore
+    def to(self, type_var: IcebergType) -> Literal[bool]:
         raise TypeError(f"Cannot convert BooleanLiteral into {type_var}")
 
     @to.register(BooleanType)

pyiceberg/io/fsspec.py

@@ -91,7 +91,7 @@
 
 
 def s3v4_rest_signer(properties: Properties, request: "AWSRequest", **_: Any) -> "AWSRequest":
-    signer_url = properties.get(S3_SIGNER_URI, properties["uri"]).rstrip("/")
+    signer_url = properties.get(S3_SIGNER_URI, properties["uri"]).rstrip("/")  # type: ignore
     signer_endpoint = properties.get(S3_SIGNER_ENDPOINT, S3_SIGNER_ENDPOINT_DEFAULT)
 
     signer_headers = {}

pyiceberg/manifest.py

@@ -912,7 +912,7 @@ def _inherit_from_manifest(entry: ManifestEntry, manifest: ManifestFile) -> Mani
     """
     # Inherit sequence numbers.
     # The snapshot_id is required in V1, inherit with V2 when null
-    if entry.snapshot_id is None:
+    if entry.snapshot_id is None and manifest.added_snapshot_id is not None:
         entry.snapshot_id = manifest.added_snapshot_id
 
     # in v1 tables, the sequence number is not persisted and can be safely defaulted to 0

pyiceberg/transforms.py

@@ -1049,11 +1049,11 @@ def _truncate_number(
         raise ValueError(f"Expected a numeric literal, got: {type(boundary)}")
 
     if isinstance(pred, BoundLessThan):
-        return LessThanOrEqual(Reference(name), _transform_literal(transform, boundary.decrement()))  # type: ignore
+        return LessThanOrEqual(Reference(name), _transform_literal(transform, boundary.decrement()))
     elif isinstance(pred, BoundLessThanOrEqual):
         return LessThanOrEqual(Reference(name), _transform_literal(transform, boundary))
     elif isinstance(pred, BoundGreaterThan):
-        return GreaterThanOrEqual(Reference(name), _transform_literal(transform, boundary.increment()))  # type: ignore
+        return GreaterThanOrEqual(Reference(name), _transform_literal(transform, boundary.increment()))
     elif isinstance(pred, BoundGreaterThanOrEqual):
         return GreaterThanOrEqual(Reference(name), _transform_literal(transform, boundary))
     elif isinstance(pred, BoundEqualTo):
@@ -1073,11 +1073,11 @@ def _truncate_number_strict(
     if isinstance(pred, BoundLessThan):
         return LessThan(Reference(name), _transform_literal(transform, boundary))
     elif isinstance(pred, BoundLessThanOrEqual):
-        return LessThan(Reference(name), _transform_literal(transform, boundary.increment()))  # type: ignore
+        return LessThan(Reference(name), _transform_literal(transform, boundary.increment()))
     elif isinstance(pred, BoundGreaterThan):
         return GreaterThan(Reference(name), _transform_literal(transform, boundary))
     elif isinstance(pred, BoundGreaterThanOrEqual):
-        return GreaterThan(Reference(name), _transform_literal(transform, boundary.decrement()))  # type: ignore
+        return GreaterThan(Reference(name), _transform_literal(transform, boundary.decrement()))
     elif isinstance(pred, BoundNotEqualTo):
         return NotEqualTo(Reference(name), _transform_literal(transform, boundary))
     elif isinstance(pred, BoundEqualTo):

pyiceberg/utils/config.py

@@ -159,6 +159,12 @@ def get_catalog_config(self, catalog_name: str) -> Optional[RecursiveDict]:
                 return catalog_conf
         return None
 
+    def get_known_catalogs(self) -> List[str]:
+        catalogs = self.config.get(CATALOG, {})
+        if not isinstance(catalogs, dict):
+            raise ValueError("Catalog configurations needs to be an object")
+        return list(catalogs.keys())
+
     def get_int(self, key: str) -> Optional[int]:
         if (val := self.config.get(key)) is not None:
             try: