merge main, fix integration and unit tests

Author: igorvoltaic

dev/Dockerfile

@@ -40,7 +40,6 @@ WORKDIR ${SPARK_HOME}
 ENV SPARK_VERSION=3.5.6
 ENV ICEBERG_SPARK_RUNTIME_VERSION=3.5_2.12
 ENV ICEBERG_VERSION=1.9.1
-ENV PYICEBERG_VERSION=0.9.1
 
 RUN curl --retry 5 -s -C - https://archive.apache.org/dist/spark/spark-${SPARK_VERSION}/spark-${SPARK_VERSION}-bin-hadoop3.tgz -o spark-${SPARK_VERSION}-bin-hadoop3.tgz \
  && tar xzf spark-${SPARK_VERSION}-bin-hadoop3.tgz --directory /opt/spark --strip-components 1 \
@@ -55,18 +54,30 @@ RUN curl --retry 5 -s https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-
 RUN curl --retry 5 -s https://repo1.maven.org/maven2/org/apache/iceberg/iceberg-aws-bundle/${ICEBERG_VERSION}/iceberg-aws-bundle-${ICEBERG_VERSION}.jar \
       -Lo /opt/spark/jars/iceberg-aws-bundle-${ICEBERG_VERSION}.jar
 
-COPY spark-defaults.conf /opt/spark/conf
+COPY dev/spark-defaults.conf /opt/spark/conf
 ENV PATH="/opt/spark/sbin:/opt/spark/bin:${PATH}"
 
 RUN chmod u+x /opt/spark/sbin/* && \
     chmod u+x /opt/spark/bin/*
 
 RUN pip3 install -q ipython
 
-RUN pip3 install "pyiceberg[s3fs,hive,pyarrow]==${PYICEBERG_VERSION}"
+# Copy the local pyiceberg source code and install locally
+COPY pyiceberg/ /tmp/pyiceberg/pyiceberg
+COPY pyproject.toml /tmp/pyiceberg/
+COPY build-module.py /tmp/pyiceberg/
+COPY vendor/ /tmp/pyiceberg/vendor
+COPY README.md /tmp/pyiceberg/
+COPY NOTICE /tmp/pyiceberg/
 
-COPY entrypoint.sh .
-COPY provision.py .
+# Install pyiceberg from the copied source
+RUN cd /tmp/pyiceberg && pip3 install ".[s3fs,hive,pyarrow]"
+
+# Clean up
+RUN rm -rf /tmp/pyiceberg
+
+COPY dev/entrypoint.sh ${SPARK_HOME}/
+COPY dev/provision.py ${SPARK_HOME}/
 
 ENTRYPOINT ["./entrypoint.sh"]
 CMD ["notebook"]

dev/docker-compose-integration.yml

@@ -19,7 +19,9 @@ services:
   spark-iceberg:
     image: python-integration
     container_name: pyiceberg-spark
-    build: .
+    build:
+      context: ..
+      dockerfile: dev/Dockerfile
     networks:
       iceberg_net:
     depends_on:
@@ -53,7 +55,6 @@ services:
       - CATALOG_WAREHOUSE=s3://warehouse/
       - CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO
       - CATALOG_S3_ENDPOINT=http://minio:9000
-      - CATALOG_JDBC_STRICT__MODE=true
   minio:
     image: minio/minio
     container_name: pyiceberg-minio

pyiceberg/catalog/hive.py

@@ -19,6 +19,7 @@
 import logging
 import socket
 import time
+from collections import namedtuple
 from types import TracebackType
 from typing import (
     TYPE_CHECKING,
@@ -33,8 +34,8 @@
 )
 from urllib.parse import urlparse
 
-from hive_metastore.v4.ThriftHiveMetastore import Client
-from hive_metastore.v4.ttypes import (
+from hive_metastore.v3.ThriftHiveMetastore import Client
+from hive_metastore.v3.ttypes import (
     AlreadyExistsException,
     CheckLockRequest,
     EnvironmentContext,
@@ -54,10 +55,10 @@
     StorageDescriptor,
     UnlockRequest,
 )
-from hive_metastore.v4.ttypes import (
+from hive_metastore.v3.ttypes import (
     Database as HiveDatabase,
 )
-from hive_metastore.v4.ttypes import (
+from hive_metastore.v3.ttypes import (
     Table as HiveTable,
 )
 from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential
@@ -148,6 +149,7 @@
 DEFAULT_LOCK_CHECK_RETRIES = 4
 DO_NOT_UPDATE_STATS = "DO_NOT_UPDATE_STATS"
 DO_NOT_UPDATE_STATS_DEFAULT = "true"
+HiveVersion = namedtuple("HiveVersion", "major minor patch")
 
 logger = logging.getLogger(__name__)
 
@@ -157,7 +159,7 @@ class _HiveClient:
 
     _transport: TTransport
     _ugi: Optional[List[str]]
-    _hive_version: int = 4
+    _hive_version: HiveVersion = HiveVersion(4, 0, 0)
     _hms_v3: object
     _hms_v4: object
 
@@ -177,13 +179,15 @@ def __init__(
         self.hms_v4 = importlib.import_module("hive_metastore.v4.ThriftHiveMetastore")
         self._hive_version = self._get_hive_version()
 
-    def _get_hive_version(self) -> int:
+    def _get_hive_version(self) -> HiveVersion:
         with self as open_client:
-            major, *_ = open_client.getVersion().split(".")
-            return int(major)
+            version = map(int, open_client.getVersion().split("."))
+            return HiveVersion(*version)
 
     def _init_thrift_transport(self) -> TTransport:
         url_parts = urlparse(self._uri)
+        if not url_parts.hostname or not url_parts.port:
+            raise ValueError("hive hostname and port must be set")
         socket = TSocket.TSocket(url_parts.hostname, url_parts.port)
         if not self._kerberos_auth:
             return TTransport.TBufferedTransport(socket)
@@ -192,7 +196,7 @@ def _init_thrift_transport(self) -> TTransport:
 
     def _client(self) -> Client:
         protocol = TBinaryProtocol.TBinaryProtocol(self._transport)
-        hms = self.hms_v3 if self._hive_version < 4 else self.hms_v4
+        hms = self.hms_v4 if all((self._hive_version.major >= 4, self._hive_version.patch > 0)) else self.hms_v3
         client: Client = hms.Client(protocol)
         if self._ugi:
             client.set_ugi(*self._ugi)
@@ -407,14 +411,17 @@ def _create_hive_table(self, open_client: Client, hive_table: HiveTable) -> None
             raise TableAlreadyExistsError(f"Table {hive_table.dbName}.{hive_table.tableName} already exists") from e
 
     def _get_hive_table(self, open_client: Client, *, dbname: str, tbl_name: str) -> HiveTable:
-        if open_client._hive_version < 4:
+        try:
+            if all((self._client._hive_version.major >= 4, self._client._hive_version.patch > 0)):
+                return open_client.get_table_req(GetTableRequest(dbName=dbname, tblName=tbl_name)).table
             return open_client.get_table(dbname=dbname, tbl_name=tbl_name)
-        return open_client.get_table_req(GetTableRequest(dbName=dbname, tblName=tbl_name)).table
+        except NoSuchObjectException as e:
+            raise NoSuchTableError(f"Table does not exists: {tbl_name}") from e
 
     def _get_table_objects_by_name(self, open_client: Client, *, dbname: str, tbl_names: list[str]) -> list[HiveTable]:
-        if open_client._hive_version < 4:
-            return open_client.get_table_objects_by_name(dbname=dbname, tbl_names=tbl_names)
-        return open_client.get_table_objects_by_name_req(GetTablesRequest(dbName=dbname, tblNames=tbl_names)).tables
+        if all((self._client._hive_version.major >= 4, self._client._hive_version.patch > 0)):
+            return open_client.get_table_objects_by_name_req(GetTablesRequest(dbName=dbname, tblNames=tbl_names)).tables
+        return open_client.get_table_objects_by_name(dbname=dbname, tbl_names=tbl_names)
 
     def create_table(
         self,
@@ -458,7 +465,7 @@ def create_table(
 
         with self._client as open_client:
             self._create_hive_table(open_client, tbl)
-            hive_table: HiveTable = self._get_hive_table(open_client, dbname=database_name, tbl_name=table_name)
+            hive_table = self._get_hive_table(open_client, dbname=database_name, tbl_name=table_name)
 
         return self._convert_hive_into_iceberg(hive_table)
 
@@ -488,7 +495,7 @@ def register_table(self, identifier: Union[str, Identifier], metadata_location:
         tbl = self._convert_iceberg_into_hive(staged_table)
         with self._client as open_client:
             self._create_hive_table(open_client, tbl)
-            hive_table: HiveTable = self._get_hive_table(open_client, dbname=database_name, tbl_name=table_name)
+            hive_table = self._get_hive_table(open_client, dbname=database_name, tbl_name=table_name)
 
         return self._convert_hive_into_iceberg(hive_table)
 
@@ -674,12 +681,17 @@ def rename_table(self, from_identifier: Union[str, Identifier], to_identifier: U
             ValueError: When from table identifier is invalid.
             NoSuchTableError: When a table with the name does not exist.
             NoSuchNamespaceError: When the destination namespace doesn't exist.
+            TableAlreadyExistsError: When the destination table already exists.
         """
         from_database_name, from_table_name = self.identifier_to_database_and_table(from_identifier, NoSuchTableError)
         to_database_name, to_table_name = self.identifier_to_database_and_table(to_identifier)
+
+        if self.table_exists(to_identifier):
+            raise TableAlreadyExistsError(f"Table already exists: {to_table_name}")
+
         try:
             with self._client as open_client:
-                tbl: HiveTable = self._get_hive_table(open_client, dbname=from_database_name, tbl_name=from_table_name)
+                tbl = self._get_hive_table(open_client, dbname=from_database_name, tbl_name=from_table_name)
                 tbl.dbName = to_database_name
                 tbl.tableName = to_table_name
                 open_client.alter_table_with_environment_context(
@@ -823,7 +835,7 @@ def update_namespace_properties(
             if removals:
                 for key in removals:
                     if key in parameters:
-                        parameters[key] = None
+                        parameters.pop(key)
                         removed.add(key)
             if updates:
                 for key, value in updates.items():

tests/catalog/test_hive.py

@@ -27,7 +27,7 @@
 
 import pytest
 import thrift.transport.TSocket
-from hive_metastore.v4.ttypes import (
+from hive_metastore.v3.ttypes import (
     AlreadyExistsException,
     EnvironmentContext,
     FieldSchema,
@@ -40,10 +40,10 @@
     SkewedInfo,
     StorageDescriptor,
 )
-from hive_metastore.v4.ttypes import (
+from hive_metastore.v3.ttypes import (
     Database as HiveDatabase,
 )
-from hive_metastore.v4.ttypes import (
+from hive_metastore.v3.ttypes import (
     Table as HiveTable,
 )
 
@@ -57,6 +57,7 @@
     LOCK_CHECK_MIN_WAIT_TIME,
     LOCK_CHECK_RETRIES,
     HiveCatalog,
+    HiveVersion,
     _construct_hive_storage_descriptor,
     _HiveClient,
 )
@@ -65,6 +66,7 @@
     NamespaceNotEmptyError,
     NoSuchNamespaceError,
     NoSuchTableError,
+    TableAlreadyExistsError,
     WaitingForLockException,
 )
 from pyiceberg.partitioning import PartitionField, PartitionSpec
@@ -258,7 +260,7 @@ def test_no_uri_supplied() -> None:
 
 
 def test_check_number_of_namespaces(table_schema_simple: Schema) -> None:
-    _HiveClient._get_hive_version = MagicMock(return_value=3)  # type: ignore
+    _HiveClient._get_hive_version = MagicMock(return_value=HiveVersion(4, 0, 0))  # type: ignore
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
 
     with pytest.raises(ValueError):
@@ -889,6 +891,7 @@ def test_load_table_from_self_identifier(hive_table: HiveTable) -> None:
 
 def test_rename_table(hive_table: HiveTable) -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
+    catalog.table_exists = MagicMock(return_value=False)  # type: ignore[method-assign]
 
     renamed_table = copy.deepcopy(hive_table)
     renamed_table.dbName = "default"
@@ -919,6 +922,7 @@ def test_rename_table(hive_table: HiveTable) -> None:
 
 def test_rename_table_from_self_identifier(hive_table: HiveTable) -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
+    catalog.table_exists = MagicMock(return_value=False)  # type: ignore[method-assign]
 
     catalog._client = MagicMock()
     catalog._get_hive_table = MagicMock(return_value=hive_table)  # type: ignore
@@ -953,9 +957,10 @@ def test_rename_table_from_self_identifier(hive_table: HiveTable) -> None:
 
 def test_rename_table_from_does_not_exists() -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
+    catalog.table_exists = MagicMock(return_value=False)  # type: ignore[method-assign]
 
     catalog._client = MagicMock()
-    catalog._client.__enter__()._hive_version = 3
+    catalog._client._hive_version = HiveVersion(4, 0, 0)
     catalog._client.__enter__().alter_table_with_environment_context.side_effect = NoSuchObjectException(
         message="hive.default.does_not_exists table not found"
     )
@@ -966,11 +971,22 @@ def test_rename_table_from_does_not_exists() -> None:
     assert "Table does not exist: does_not_exists" in str(exc_info.value)
 
 
+def test_rename_table_to_table_already_exists(hive_table: HiveTable) -> None:
+    catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
+    catalog.load_table = MagicMock(return_value=hive_table)  # type: ignore[method-assign]
+
+    with pytest.raises(TableAlreadyExistsError) as exc_info:
+        catalog.rename_table(("default", "some_table"), ("default", "new_tabl2e"))
+
+    assert "Table already exists: new_tabl2e" in str(exc_info.value)
+
+
 def test_rename_table_to_namespace_does_not_exists() -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
+    catalog.table_exists = MagicMock(return_value=False)  # type: ignore[method-assign]
 
     catalog._client = MagicMock()
-    catalog._client.__enter__()._hive_version = 3
+    catalog._client._hive_version = HiveVersion(4, 0, 0)
     catalog._client.__enter__().alter_table_with_environment_context.side_effect = InvalidOperationException(
         message="Unable to change partition or table. Database default does not exist Check metastore logs for detailed stack.does_not_exists"
     )
@@ -1171,7 +1187,7 @@ def test_update_namespace_properties(hive_database: HiveDatabase) -> None:
             name="default",
             description=None,
             locationUri=hive_database.locationUri,
-            parameters={"test": None, "label": "core"},
+            parameters={"label": "core"},
             privileges=None,
             ownerName=None,
             ownerType=1,

tests/integration/test_reads.py

@@ -26,7 +26,7 @@
 import pyarrow as pa
 import pyarrow.parquet as pq
 import pytest
-from hive_metastore.v4.ttypes import LockRequest, LockResponse, LockState, UnlockRequest
+from hive_metastore.v3.ttypes import LockRequest, LockResponse, LockState, UnlockRequest
 from pyarrow.fs import S3FileSystem
 from pydantic_core import ValidationError
 from pyspark.sql import SparkSession
@@ -115,22 +115,21 @@ def test_table_properties(catalog: Catalog) -> None:
 
 @pytest.mark.integration
 @pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog_hive")])
-def test_hive_properties(catalog: Catalog) -> None:
+def test_hive_properties(catalog: HiveCatalog) -> None:
     table = create_table(catalog)
     table.transaction().set_properties({"abc": "def", "p1": "123"}).commit_transaction()
-
     hive_client: _HiveClient = _HiveClient(catalog.properties["uri"])
 
     with hive_client as open_client:
-        hive_table = open_client.get_table(*TABLE_NAME)
+        hive_table = catalog._get_hive_table(open_client, dbname="default", tbl_name="t1")
         assert hive_table.parameters.get("abc") == "def"
         assert hive_table.parameters.get("p1") == "123"
         assert hive_table.parameters.get("not_exist_parameter") is None
 
     table.transaction().remove_properties("abc").commit_transaction()
 
     with hive_client as open_client:
-        hive_table = open_client.get_table(*TABLE_NAME)
+        hive_table = catalog._get_hive_table(open_client, dbname="default", tbl_name="t1")
         assert hive_table.parameters.get("abc") is None