add version tuple

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,
@@ -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,10 +179,10 @@ 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)
@@ -192,7 +194,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.minor > 0)) else self.hms_v3
         client: Client = hms.Client(protocol)
         if self._ugi:
             client.set_ugi(*self._ugi)
@@ -407,14 +409,14 @@ 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 self._client._hive_version < 4:
-            return open_client.get_table(dbname=dbname, tbl_name=tbl_name)
-        return open_client.get_table_req(GetTableRequest(dbName=dbname, tblName=tbl_name)).table
+        if all((self._client._hive_version.major >= 4, self._client._hive_version.minor > 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)
 
     def _get_table_objects_by_name(self, open_client: Client, *, dbname: str, tbl_names: list[str]) -> list[HiveTable]:
-        if self._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.minor > 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,

tests/catalog/test_hive.py

@@ -57,6 +57,7 @@
     LOCK_CHECK_MIN_WAIT_TIME,
     LOCK_CHECK_RETRIES,
     HiveCatalog,
+    HiveVersion,
     _construct_hive_storage_descriptor,
     _HiveClient,
 )
@@ -258,7 +259,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):
@@ -955,7 +956,7 @@ def test_rename_table_from_does_not_exists() -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
 
     catalog._client = MagicMock()
-    catalog._client._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"
     )
@@ -970,7 +971,7 @@ def test_rename_table_to_namespace_does_not_exists() -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
 
     catalog._client = MagicMock()
-    catalog._client._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"
     )