Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions dev/docker-compose-integration.yml
Original file line number Diff line number Diff line change
Expand Up @@ -98,8 +98,6 @@ services:
image: pyiceberg-hive:latest
build:
context: hive/
args:
MAVEN_MIRROR: ${MAVEN_MIRROR:-https://repo1.maven.org/maven2}
Comment on lines -101 to -102

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would be nice to keep this. Some folks might be behind a private mirror for vulnerability scanning.

container_name: pyiceberg-hive
hostname: hive
networks:
Expand Down
28 changes: 7 additions & 21 deletions dev/hive/Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -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
7 changes: 7 additions & 0 deletions dev/spark/Dockerfile
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 && \
Expand All @@ -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}") && \
Expand Down
5 changes: 5 additions & 0 deletions dev/spark/spark-defaults.conf
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
31 changes: 22 additions & 9 deletions pyiceberg/catalog/hive.py
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,8 @@
CheckLockRequest,
EnvironmentContext,
FieldSchema,
GetTableRequest,
GetTablesRequest,
InvalidOperationException,
LockComponent,
LockLevel,
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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),
Expand All @@ -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

Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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
]

Expand Down
52 changes: 35 additions & 17 deletions tests/catalog/test_hive.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,10 @@
AlreadyExistsException,
EnvironmentContext,
FieldSchema,
GetTableRequest,
GetTableResult,
GetTablesRequest,
GetTablesResult,
InvalidOperationException,
LockResponse,
LockState,
Expand Down Expand Up @@ -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)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

More of a meta thing, but I think most of these mock tests predate the Hive integration tests, not sure how valuable these are.

catalog._client.__enter__().get_database.return_value = hive_database
catalog.create_table(("default", "table"), schema=table_schema_with_all_types, properties={"owner": "javaberg"})

Expand Down Expand Up @@ -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}/"
Expand Down Expand Up @@ -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"}
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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")
Expand All @@ -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",
Expand All @@ -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",
Expand Down Expand Up @@ -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"])
)


Expand Down Expand Up @@ -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"]
Expand Down
Loading
Loading