Update hive docker to 4.2.1 (#3924)

* Update hive docker to 4.2.1

* Use newest Hive docker image

Author: rambleraptor

dev/docker-compose-integration.yml

@@ -98,8 +98,6 @@ services:
     image: pyiceberg-hive:latest
     build:
       context: hive/
-      args:
-        MAVEN_MIRROR: ${MAVEN_MIRROR:-https://repo1.maven.org/maven2}
     container_name: pyiceberg-hive
     hostname: hive
     networks:

dev/hive/Dockerfile

@@ -13,30 +13,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM apache/hive:4.0.0
-
-# Dependency versions - changing these invalidates the JAR download layer
-ARG HADOOP_VERSION=3.3.6
-ARG AWS_SDK_BUNDLE=1.12.753
-ARG MAVEN_MIRROR=https://repo1.maven.org/maven2
+FROM apache/hive:4.2.1
 
 USER root
 
-# Install curl (separate layer - rarely changes)
-RUN apt-get update -qq && \
-    apt-get -qq -y install --no-install-recommends curl && \
-    apt-get clean && \
-    rm -rf /var/lib/apt/lists/*
-
-# Download JARs with retry logic (slow layer - only changes when versions change)
-RUN curl -fsSL --retry 3 --retry-delay 5 \
-        -o /opt/hive/lib/hadoop-aws-${HADOOP_VERSION}.jar \
-        "${MAVEN_MIRROR}/org/apache/hadoop/hadoop-aws/${HADOOP_VERSION}/hadoop-aws-${HADOOP_VERSION}.jar" && \
-    curl -fsSL --retry 3 --retry-delay 5 \
-        -o /opt/hive/lib/aws-java-sdk-bundle-${AWS_SDK_BUNDLE}.jar \
-        "${MAVEN_MIRROR}/com/amazonaws/aws-java-sdk-bundle/${AWS_SDK_BUNDLE}/aws-java-sdk-bundle-${AWS_SDK_BUNDLE}.jar"
+# Link the hadoop-aws and AWS SDK jars that the image ships into the metastore classpath
+RUN ln -s /opt/hadoop/share/hadoop/tools/lib/hadoop-aws-*.jar /opt/hive/lib/ && \
+    ln -s /opt/hadoop/share/hadoop/tools/lib/bundle-*.jar /opt/hive/lib/
 
-# Copy configuration last (changes more frequently than JARs)
-COPY core-site.xml /opt/hadoop/etc/hadoop/core-site.xml
+# The entrypoint links this directory into the Hive config directory, over its own core-site.xml
+ENV HIVE_CUSTOM_CONF_DIR=/opt/hive/custom-conf
+COPY core-site.xml ${HIVE_CUSTOM_CONF_DIR}/core-site.xml
 
 USER hive

dev/spark/Dockerfile

@@ -23,6 +23,7 @@ ARG ICEBERG_VERSION=1.11.0
 ARG ICEBERG_SPARK_RUNTIME_VERSION=4.0_2.13
 ARG HADOOP_VERSION=3.4.1
 ARG AWS_SDK_VERSION=2.24.6
+ARG HIVE_METASTORE_VERSION=4.0.1
 ARG MAVEN_MIRROR=https://repo.maven.apache.org/maven2
 
 USER root
@@ -36,6 +37,11 @@ RUN apt-get update -qq && \
     mkdir -p /home/iceberg/spark-events && \
     chown -R spark:spark /home/iceberg
 
+# Iceberg's Hive catalog uses the metastore client from the class path, and the bundled
+# Hive 2.3 one calls get_table, which Hive dropped in 4.0.1. Move it aside for a Hive 4 client.
+RUN mkdir -p "${SPARK_HOME}/hive-metastore-jars" && \
+    mv "${SPARK_HOME}"/jars/hive-metastore-*.jar "${SPARK_HOME}/hive-metastore-jars/"
+
 # Download JARs with retry logic (most cacheable - only changes when versions change)
 # This is the slowest step, so we do it before copying config files
 RUN set -e && \
@@ -44,6 +50,7 @@ RUN set -e && \
         "org/apache/iceberg/iceberg-spark-runtime-${ICEBERG_SPARK_RUNTIME_VERSION}/${ICEBERG_VERSION}/iceberg-spark-runtime-${ICEBERG_SPARK_RUNTIME_VERSION}-${ICEBERG_VERSION}.jar" \
         "org/apache/iceberg/iceberg-aws-bundle/${ICEBERG_VERSION}/iceberg-aws-bundle-${ICEBERG_VERSION}.jar" \
         "org/apache/hadoop/hadoop-aws/${HADOOP_VERSION}/hadoop-aws-${HADOOP_VERSION}.jar" \
+        "org/apache/hive/hive-standalone-metastore-common/${HIVE_METASTORE_VERSION}/hive-standalone-metastore-common-${HIVE_METASTORE_VERSION}.jar" \
         "software/amazon/awssdk/bundle/${AWS_SDK_VERSION}/bundle-${AWS_SDK_VERSION}.jar"; \
     do \
         jar_name=$(basename "${jar_path}") && \

dev/spark/spark-defaults.conf

@@ -42,6 +42,11 @@ spark.hadoop.fs.s3a.endpoint           http://minio:9000
 spark.sql.catalogImplementation        hive
 spark.sql.warehouse.dir                s3a://warehouse/hive/
 
+# Spark's own Hive client speaks the Hive 2.3 API, so keep it off the Hive 4 client on the class path
+spark.sql.hive.metastore.version       2.3.10
+spark.sql.hive.metastore.jars          path
+spark.sql.hive.metastore.jars.path     file:///opt/spark/hive-metastore-jars/*,file:///opt/spark/jars/*
+
 spark.sql.defaultCatalog               rest
 
 # Configure Spark UI and event logging

pyiceberg/catalog/hive.py

@@ -32,6 +32,8 @@
     CheckLockRequest,
     EnvironmentContext,
     FieldSchema,
+    GetTableRequest,
+    GetTablesRequest,
     InvalidOperationException,
     LockComponent,
     LockLevel,
@@ -297,6 +299,7 @@ class HiveCatalog(MetastoreCatalog):
     def __init__(self, name: str, **properties: str):
         super().__init__(name, **properties)
         self._client = self._create_hive_client(properties)
+        self._hive2_compatible = property_as_bool(properties, HIVE2_COMPATIBLE, HIVE2_COMPATIBLE_DEFAULT)
 
         self._lock_check_min_wait_time = property_as_float(properties, LOCK_CHECK_MIN_WAIT_TIME, DEFAULT_LOCK_CHECK_MIN_WAIT_TIME)
         self._lock_check_max_wait_time = property_as_float(properties, LOCK_CHECK_MAX_WAIT_TIME, DEFAULT_LOCK_CHECK_MAX_WAIT_TIME)
@@ -367,7 +370,7 @@ def _convert_iceberg_into_hive(self, table: Table) -> HiveTable:
             sd=_construct_hive_storage_descriptor(
                 table.schema(),
                 table.location(),
-                property_as_bool(self.properties, HIVE2_COMPATIBLE, HIVE2_COMPATIBLE_DEFAULT),
+                self._hive2_compatible,
             ),
             tableType=EXTERNAL_TABLE,
             parameters=_construct_parameters(metadata_location=table.metadata_location, metadata_properties=table.properties),
@@ -379,9 +382,21 @@ def _create_hive_table(self, open_client: Client, hive_table: HiveTable) -> None
         except AlreadyExistsException as e:
             raise TableAlreadyExistsError(f"Table {hive_table.dbName}.{hive_table.tableName} already exists") from e
 
+    def _fetch_hive_table(self, open_client: Client, database_name: str, table_name: str) -> HiveTable:
+        # Hive 4.0.1 removed get_table, and Hive 2 does not have get_table_req
+        if self._hive2_compatible:
+            return open_client.get_table(dbname=database_name, tbl_name=table_name)
+        return open_client.get_table_req(GetTableRequest(dbName=database_name, tblName=table_name)).table
+
+    def _fetch_hive_tables(self, open_client: Client, database_name: str) -> list[HiveTable]:
+        table_names = open_client.get_all_tables(db_name=database_name)
+        if self._hive2_compatible:
+            return open_client.get_table_objects_by_name(dbname=database_name, tbl_names=table_names)
+        return open_client.get_table_objects_by_name_req(GetTablesRequest(dbName=database_name, tblNames=table_names)).tables
+
     def _get_hive_table(self, open_client: Client, database_name: str, table_name: str) -> HiveTable:
         try:
-            return open_client.get_table(dbname=database_name, tbl_name=table_name)
+            return self._fetch_hive_table(open_client, database_name, table_name)
         except NoSuchObjectException as e:
             raise NoSuchTableError(f"Table does not exists: {table_name}") from e
 
@@ -428,7 +443,7 @@ def create_table(
 
         with self._client as open_client:
             self._create_hive_table(open_client, tbl)
-            hive_table = open_client.get_table(dbname=database_name, tbl_name=table_name)
+            hive_table = self._fetch_hive_table(open_client, database_name, table_name)
 
         return self._convert_hive_into_iceberg(hive_table)
 
@@ -474,7 +489,7 @@ def register_table(self, identifier: str | Identifier, metadata_location: str, o
         tbl = self._convert_iceberg_into_hive(staged_table)
         with self._client as open_client:
             self._create_hive_table(open_client, tbl)
-            hive_table = open_client.get_table(dbname=database_name, tbl_name=table_name)
+            hive_table = self._fetch_hive_table(open_client, database_name, table_name)
 
         return self._convert_hive_into_iceberg(hive_table)
 
@@ -603,7 +618,7 @@ def commit_table(
                     hive_table.sd = _construct_hive_storage_descriptor(
                         updated_staged_table.schema(),
                         updated_staged_table.location(),
-                        property_as_bool(self.properties, HIVE2_COMPATIBLE, HIVE2_COMPATIBLE_DEFAULT),
+                        self._hive2_compatible,
                     )
                     open_client.alter_table_with_environment_context(
                         dbname=database_name,
@@ -703,7 +718,7 @@ def rename_table(self, from_identifier: str | Identifier, to_identifier: str | I
 
         try:
             with self._client as open_client:
-                tbl = open_client.get_table(dbname=from_database_name, tbl_name=from_table_name)
+                tbl = self._fetch_hive_table(open_client, from_database_name, from_table_name)
                 tbl.dbName = to_database_name
                 tbl.tableName = to_table_name
                 open_client.alter_table_with_environment_context(
@@ -778,9 +793,7 @@ def list_tables(self, namespace: str | Identifier) -> list[Identifier]:
         with self._client as open_client:
             return [
                 (database_name, table.tableName)
-                for table in open_client.get_table_objects_by_name(
-                    dbname=database_name, tbl_names=open_client.get_all_tables(db_name=database_name)
-                )
+                for table in self._fetch_hive_tables(open_client, database_name)
                 if table.parameters.get(TABLE_TYPE, "").lower() == ICEBERG
             ]
 

tests/catalog/test_hive.py

@@ -31,6 +31,10 @@
     AlreadyExistsException,
     EnvironmentContext,
     FieldSchema,
+    GetTableRequest,
+    GetTableResult,
+    GetTablesRequest,
+    GetTablesResult,
     InvalidOperationException,
     LockResponse,
     LockState,
@@ -293,6 +297,7 @@ def test_create_table(
     catalog._client = MagicMock()
     catalog._client.__enter__().create_table.return_value = None
     catalog._client.__enter__().get_table.return_value = hive_table
+    catalog._client.__enter__().get_table_req.return_value = GetTableResult(table=hive_table)
     catalog._client.__enter__().get_database.return_value = hive_database
     catalog.create_table(("default", "table"), schema=table_schema_with_all_types, properties={"owner": "javaberg"})
 
@@ -472,6 +477,7 @@ def test_create_table_with_given_location_removes_trailing_slash(
     catalog._client = MagicMock()
     catalog._client.__enter__().create_table.return_value = None
     catalog._client.__enter__().get_table.return_value = hive_table
+    catalog._client.__enter__().get_table_req.return_value = GetTableResult(table=hive_table)
     catalog._client.__enter__().get_database.return_value = hive_database
     catalog.create_table(
         ("default", "table"), schema=table_schema_with_all_types, properties={"owner": "javaberg"}, location=f"{location}/"
@@ -645,7 +651,7 @@ def test_create_v1_table(table_schema_simple: Schema, hive_database: HiveDatabas
 
     catalog._client = MagicMock()
     catalog._client.__enter__().create_table.return_value = None
-    catalog._client.__enter__().get_table.return_value = hive_table
+    catalog._client.__enter__().get_table_req.return_value = GetTableResult(table=hive_table)
     catalog._client.__enter__().get_database.return_value = hive_database
     catalog.create_table(
         ("default", "table"), schema=table_schema_simple, properties={"owner": "javaberg", "format-version": "1"}
@@ -696,10 +702,10 @@ def test_load_table(hive_table: HiveTable) -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
 
     catalog._client = MagicMock()
-    catalog._client.__enter__().get_table.return_value = hive_table
+    catalog._client.__enter__().get_table_req.return_value = GetTableResult(table=hive_table)
     table = catalog.load_table(("default", "new_tabl2e"))
 
-    catalog._client.__enter__().get_table.assert_called_with(dbname="default", tbl_name="new_tabl2e")
+    catalog._client.__enter__().get_table_req.assert_called_with(GetTableRequest(dbName="default", tblName="new_tabl2e"))
 
     expected = TableMetadataV2(
         location="s3://bucket/test/location",
@@ -796,11 +802,11 @@ def test_load_table_from_self_identifier(hive_table: HiveTable) -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
 
     catalog._client = MagicMock()
-    catalog._client.__enter__().get_table.return_value = hive_table
+    catalog._client.__enter__().get_table_req.return_value = GetTableResult(table=hive_table)
     intermediate = catalog.load_table(("default", "new_tabl2e"))
     table = catalog.load_table(intermediate.name())
 
-    catalog._client.__enter__().get_table.assert_called_with(dbname="default", tbl_name="new_tabl2e")
+    catalog._client.__enter__().get_table_req.assert_called_with(GetTableRequest(dbName="default", tblName="new_tabl2e"))
 
     expected = TableMetadataV2(
         location="s3://bucket/test/location",
@@ -902,7 +908,10 @@ def test_rename_table(hive_table: HiveTable) -> None:
     renamed_table.tableName = "new_tabl3e"
 
     catalog._client = MagicMock()
-    catalog._client.__enter__().get_table.side_effect = [hive_table, renamed_table]
+    catalog._client.__enter__().get_table_req.side_effect = [
+        GetTableResult(table=hive_table),
+        GetTableResult(table=renamed_table),
+    ]
     catalog._client.__enter__().alter_table_with_environment_context.return_value = None
 
     from_identifier = ("default", "new_tabl2e")
@@ -911,8 +920,11 @@ def test_rename_table(hive_table: HiveTable) -> None:
 
     assert table.name() == to_identifier
 
-    calls = [call(dbname="default", tbl_name="new_tabl2e"), call(dbname="default", tbl_name="new_tabl3e")]
-    catalog._client.__enter__().get_table.assert_has_calls(calls)
+    calls = [
+        call(GetTableRequest(dbName="default", tblName="new_tabl2e")),
+        call(GetTableRequest(dbName="default", tblName="new_tabl3e")),
+    ]
+    catalog._client.__enter__().get_table_req.assert_has_calls(calls)
     catalog._client.__enter__().alter_table_with_environment_context.assert_called_with(
         dbname="default",
         tbl_name="new_tabl2e",
@@ -926,25 +938,31 @@ def test_rename_table_from_self_identifier(hive_table: HiveTable) -> None:
     catalog.table_exists = MagicMock(return_value=False)  # type: ignore[method-assign]
 
     catalog._client = MagicMock()
-    catalog._client.__enter__().get_table.return_value = hive_table
+    catalog._client.__enter__().get_table_req.return_value = GetTableResult(table=hive_table)
 
     from_identifier = ("default", "new_tabl2e")
     from_table = catalog.load_table(from_identifier)
-    catalog._client.__enter__().get_table.assert_called_with(dbname="default", tbl_name="new_tabl2e")
+    catalog._client.__enter__().get_table_req.assert_called_with(GetTableRequest(dbName="default", tblName="new_tabl2e"))
 
     renamed_table = copy.deepcopy(hive_table)
     renamed_table.dbName = "default"
     renamed_table.tableName = "new_tabl3e"
 
-    catalog._client.__enter__().get_table.side_effect = [hive_table, renamed_table]
+    catalog._client.__enter__().get_table_req.side_effect = [
+        GetTableResult(table=hive_table),
+        GetTableResult(table=renamed_table),
+    ]
     catalog._client.__enter__().alter_table_with_environment_context.return_value = None
     to_identifier = ("default", "new_tabl3e")
     table = catalog.rename_table(from_table.name(), to_identifier)
 
     assert table.name() == to_identifier
 
-    calls = [call(dbname="default", tbl_name="new_tabl2e"), call(dbname="default", tbl_name="new_tabl3e")]
-    catalog._client.__enter__().get_table.assert_has_calls(calls)
+    calls = [
+        call(GetTableRequest(dbName="default", tblName="new_tabl2e")),
+        call(GetTableRequest(dbName="default", tblName="new_tabl3e")),
+    ]
+    catalog._client.__enter__().get_table_req.assert_has_calls(calls)
     catalog._client.__enter__().alter_table_with_environment_context.assert_called_with(
         dbname="default",
         tbl_name="new_tabl2e",
@@ -1042,13 +1060,13 @@ def test_list_tables(hive_table: HiveTable) -> None:
 
     catalog._client = MagicMock()
     catalog._client.__enter__().get_all_tables.return_value = ["table1", "table2", "table3", "table4"]
-    catalog._client.__enter__().get_table_objects_by_name.return_value = [tbl1, tbl2, tbl3, tbl4]
+    catalog._client.__enter__().get_table_objects_by_name_req.return_value = GetTablesResult(tables=[tbl1, tbl2, tbl3, tbl4])
 
     got_tables = catalog.list_tables("database")
     assert got_tables == [("database", "table1"), ("database", "table2")]
     catalog._client.__enter__().get_all_tables.assert_called_with(db_name="database")
-    catalog._client.__enter__().get_table_objects_by_name.assert_called_with(
-        dbname="database", tbl_names=["table1", "table2", "table3", "table4"]
+    catalog._client.__enter__().get_table_objects_by_name_req.assert_called_with(
+        GetTablesRequest(dbName="database", tblNames=["table1", "table2", "table3", "table4"])
     )
 
 
@@ -1078,7 +1096,7 @@ def test_drop_table_from_self_identifier(hive_table: HiveTable) -> None:
     catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
 
     catalog._client = MagicMock()
-    catalog._client.__enter__().get_table.return_value = hive_table
+    catalog._client.__enter__().get_table_req.return_value = GetTableResult(table=hive_table)
     table = catalog.load_table(("default", "new_tabl2e"))
 
     catalog._client.__enter__().get_all_databases.return_value = ["namespace1", "namespace2"]