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

Author: Fokko

dev/hive/core-site.xml

@@ -50,4 +50,9 @@
       <name>fs.s3a.path.style.access</name>
       <value>true</value>
     </property>
+    <property>
+      <name>hive.metastore.disallow.incompatible.col.type.changes</name>
+      <value>false</value>
+    </property>
+
 </configuration>

dev/provision.py

@@ -273,13 +273,9 @@
         """
         )
 
-    # There is an issue with CREATE OR REPLACE
-    # https://github.com/apache/iceberg/issues/8756
-    spark.sql(f"DROP TABLE IF EXISTS {catalog_name}.default.test_table_version")
-
     spark.sql(
         f"""
-    CREATE TABLE {catalog_name}.default.test_table_version (
+    CREATE OR REPLACE TABLE {catalog_name}.default.test_table_version (
         dt     date,
         number integer,
         letter string

mkdocs/docs/configuration.md

@@ -351,6 +351,7 @@ catalog:
 | rest.signing-name   | execute-api                      | The service signing name to use when SigV4 signing a request                                       |
 | oauth2-server-uri   | <https://auth-service/cc>          | Authentication URL to use for client credentials authentication (default: uri + 'v1/oauth/tokens') |
 | snapshot-loading-mode | refs                             | The snapshots to return in the body of the metadata. Setting the value to `all` would return the full set of snapshots currently valid for the table. Setting the value to `refs` would load all snapshots referenced by branches or tags. |
+| warehouse          | myWarehouse                         | Warehouse location or identifier to request from the catalog service. May be used to determine server-side overrides, such as the warehouse location. |
 
 <!-- markdown-link-check-enable-->
 
@@ -438,10 +439,11 @@ catalog:
     s3.secret-access-key: password
 ```
 
-| Key                          | Example | Description                       |
-|------------------------------| ------- | --------------------------------- |
-| hive.hive2-compatible        | true    | Using Hive 2.x compatibility mode |
-| hive.kerberos-authentication | true    | Using authentication via Kerberos |
+| Key                          | Example | Description                          |
+|------------------------------| ------- | ------------------------------------ |
+| hive.hive2-compatible        | true    | Using Hive 2.x compatibility mode    |
+| hive.kerberos-authentication | true    | Using authentication via Kerberos    |
+| hive.kerberos-service-name   | hive    | Kerberos service name (default hive) |
 
 When using Hive 2.x, make sure to set the compatibility flag:
 

poetry.lock

@@ -140,12 +140,20 @@
 
 
 def _construct_parameters(
-    metadata_location: str, glue_table: Optional["TableTypeDef"] = None, prev_metadata_location: Optional[str] = None
+    metadata_location: str,
+    glue_table: Optional["TableTypeDef"] = None,
+    prev_metadata_location: Optional[str] = None,
+    metadata_properties: Optional[Properties] = None,
 ) -> Properties:
     new_parameters = glue_table.get("Parameters", {}) if glue_table else {}
     new_parameters.update({TABLE_TYPE: ICEBERG.upper(), METADATA_LOCATION: metadata_location})
     if prev_metadata_location:
         new_parameters[PREVIOUS_METADATA_LOCATION] = prev_metadata_location
+
+    if metadata_properties:
+        for key, value in metadata_properties.items():
+            new_parameters[key] = str(value)
+
     return new_parameters
 
 
@@ -236,7 +244,7 @@ def _construct_table_input(
     table_input: "TableInputTypeDef" = {
         "Name": table_name,
         "TableType": EXTERNAL_TABLE,
-        "Parameters": _construct_parameters(metadata_location, glue_table, prev_metadata_location),
+        "Parameters": _construct_parameters(metadata_location, glue_table, prev_metadata_location, properties),
         "StorageDescriptor": {
             "Columns": _to_columns(metadata),
             "Location": metadata.location,

pyiceberg/catalog/glue.py

@@ -130,6 +130,8 @@
 
 HIVE_KERBEROS_AUTH = "hive.kerberos-authentication"
 HIVE_KERBEROS_AUTH_DEFAULT = False
+HIVE_KERBEROS_SERVICE_NAME = "hive.kerberos-service-name"
+HIVE_KERBEROS_SERVICE_NAME_DEFAULT = "hive"
 
 LOCK_CHECK_MIN_WAIT_TIME = "lock-check-min-wait-time"
 LOCK_CHECK_MAX_WAIT_TIME = "lock-check-max-wait-time"
@@ -149,9 +151,16 @@ class _HiveClient:
     _transport: TTransport
     _ugi: Optional[List[str]]
 
-    def __init__(self, uri: str, ugi: Optional[str] = None, kerberos_auth: Optional[bool] = HIVE_KERBEROS_AUTH_DEFAULT):
+    def __init__(
+        self,
+        uri: str,
+        ugi: Optional[str] = None,
+        kerberos_auth: Optional[bool] = HIVE_KERBEROS_AUTH_DEFAULT,
+        kerberos_service_name: Optional[str] = HIVE_KERBEROS_SERVICE_NAME,
+    ):
         self._uri = uri
         self._kerberos_auth = kerberos_auth
+        self._kerberos_service_name = kerberos_service_name
         self._ugi = ugi.split(":") if ugi else None
         self._transport = self._init_thrift_transport()
 
@@ -161,7 +170,7 @@ def _init_thrift_transport(self) -> TTransport:
         if not self._kerberos_auth:
             return TTransport.TBufferedTransport(socket)
         else:
-            return TTransport.TSaslClientTransport(socket, host=url_parts.hostname, service="hive")
+            return TTransport.TSaslClientTransport(socket, host=url_parts.hostname, service=self._kerberos_service_name)
 
     def _client(self) -> Client:
         protocol = TBinaryProtocol.TBinaryProtocol(self._transport)
@@ -314,6 +323,7 @@ def _create_hive_client(properties: Dict[str, str]) -> _HiveClient:
                     uri,
                     properties.get("ugi"),
                     property_as_bool(properties, HIVE_KERBEROS_AUTH, HIVE_KERBEROS_AUTH_DEFAULT),
+                    properties.get(HIVE_KERBEROS_SERVICE_NAME, HIVE_KERBEROS_SERVICE_NAME_DEFAULT),
                 )
             except BaseException as e:
                 last_exception = e
@@ -551,6 +561,12 @@ def commit_table(
                         previous_metadata_location=current_table.metadata_location,
                         metadata_properties=updated_staged_table.properties,
                     )
+                    # Update hive's schema and properties
+                    hive_table.sd = _construct_hive_storage_descriptor(
+                        updated_staged_table.schema(),
+                        updated_staged_table.location(),
+                        property_as_bool(updated_staged_table.properties, HIVE2_COMPATIBLE, HIVE2_COMPATIBLE_DEFAULT),
+                    )
                     open_client.alter_table_with_environment_context(
                         dbname=database_name,
                         tbl_name=table_name,

pyiceberg/catalog/hive.py

@@ -178,8 +178,8 @@ class RegisterTableRequest(IcebergBaseModel):
 
 
 class ConfigResponse(IcebergBaseModel):
-    defaults: Properties = Field()
-    overrides: Properties = Field()
+    defaults: Optional[Properties] = Field(default_factory=dict)
+    overrides: Optional[Properties] = Field(default_factory=dict)
 
 
 class ListNamespaceResponse(IcebergBaseModel):

pyiceberg/catalog/rest/__init__.py

@@ -33,7 +33,7 @@
 from pyiceberg.cli.output import ConsoleOutput, JsonOutput, Output
 from pyiceberg.exceptions import NoSuchNamespaceError, NoSuchPropertyException, NoSuchTableError
 from pyiceberg.table import TableProperties
-from pyiceberg.table.refs import SnapshotRef
+from pyiceberg.table.refs import SnapshotRef, SnapshotRefType
 from pyiceberg.utils.properties import property_as_int
 
 
@@ -417,7 +417,7 @@ def list_refs(ctx: Context, identifier: str, type: str, verbose: bool) -> None:
     refs = table.refs()
     if type:
         type = type.lower()
-        if type not in {"branch", "tag"}:
+        if type not in {SnapshotRefType.BRANCH, SnapshotRefType.TAG}:
             raise ValueError(f"Type must be either branch or tag, got: {type}")
 
     relevant_refs = [
@@ -431,7 +431,7 @@ def list_refs(ctx: Context, identifier: str, type: str, verbose: bool) -> None:
 
 def _retention_properties(ref: SnapshotRef, table_properties: Dict[str, str]) -> Dict[str, str]:
     retention_properties = {}
-    if ref.snapshot_ref_type == "branch":
+    if ref.snapshot_ref_type == SnapshotRefType.BRANCH:
         default_min_snapshots_to_keep = property_as_int(
             table_properties,
             TableProperties.MIN_SNAPSHOTS_TO_KEEP,