You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by fo...@apache.org on 2024/04/14 04:49:07 UTC
(iceberg-python) 02/05: [Bug Fix] Allow HiveCatalog to create table with TimestamptzType (#585)
This is an automated email from the ASF dual-hosted git repository.
fokko pushed a commit to branch pyiceberg-0.6.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit fd30bc4c3cab41510d2fed64ce724047c4afd59f
Author: Honah J <ho...@apache.org>
AuthorDate: Mon Apr 8 00:11:34 2024 -0700
[Bug Fix] Allow HiveCatalog to create table with TimestamptzType (#585)
---
mkdocs/docs/configuration.md | 9 +++++++++
pyiceberg/catalog/glue.py | 4 ++--
pyiceberg/catalog/hive.py | 29 ++++++++++++++++++++++++-----
pyiceberg/table/__init__.py | 6 ++++++
tests/catalog/test_hive.py | 13 +++++++++++--
tests/integration/test_writes.py | 20 ++++++++++++++++++++
6 files changed, 72 insertions(+), 9 deletions(-)
diff --git a/mkdocs/docs/configuration.md b/mkdocs/docs/configuration.md
index 3684082a..a761e005 100644
--- a/mkdocs/docs/configuration.md
+++ b/mkdocs/docs/configuration.md
@@ -215,6 +215,15 @@ catalog:
s3.secret-access-key: password
```
+When using Hive 2.x, make sure to set the compatibility flag:
+
+```yaml
+catalog:
+ default:
+...
+ hive.hive2-compatible: true
+```
+
## Glue Catalog
Your AWS credentials can be passed directly through the Python API.
diff --git a/pyiceberg/catalog/glue.py b/pyiceberg/catalog/glue.py
index 6f93fab9..97393881 100644
--- a/pyiceberg/catalog/glue.py
+++ b/pyiceberg/catalog/glue.py
@@ -62,7 +62,7 @@ from pyiceberg.io import load_file_io
from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
from pyiceberg.schema import Schema, SchemaVisitor, visit
from pyiceberg.serializers import FromInputFile
-from pyiceberg.table import CommitTableRequest, CommitTableResponse, Table, update_table_metadata
+from pyiceberg.table import CommitTableRequest, CommitTableResponse, Table, update_table_metadata, PropertyUtil
from pyiceberg.table.metadata import TableMetadata, new_table_metadata
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties
@@ -313,7 +313,7 @@ class GlueCatalog(Catalog):
self.glue.update_table(
DatabaseName=database_name,
TableInput=table_input,
- SkipArchive=self.properties.get(GLUE_SKIP_ARCHIVE, GLUE_SKIP_ARCHIVE_DEFAULT),
+ SkipArchive=PropertyUtil.property_as_bool(self.properties, GLUE_SKIP_ARCHIVE, GLUE_SKIP_ARCHIVE_DEFAULT),
VersionId=version_id,
)
except self.glue.exceptions.EntityNotFoundException as e:
diff --git a/pyiceberg/catalog/hive.py b/pyiceberg/catalog/hive.py
index fb1f94f3..d97ddd80 100644
--- a/pyiceberg/catalog/hive.py
+++ b/pyiceberg/catalog/hive.py
@@ -74,7 +74,7 @@ from pyiceberg.io import FileIO, load_file_io
from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
from pyiceberg.schema import Schema, SchemaVisitor, visit
from pyiceberg.serializers import FromInputFile
-from pyiceberg.table import CommitTableRequest, CommitTableResponse, Table, TableProperties, update_table_metadata
+from pyiceberg.table import CommitTableRequest, CommitTableResponse, PropertyUtil, Table, TableProperties, update_table_metadata
from pyiceberg.table.metadata import new_table_metadata
from pyiceberg.table.sorting import UNSORTED_SORT_ORDER, SortOrder
from pyiceberg.typedef import EMPTY_DICT, Identifier, Properties
@@ -107,6 +107,10 @@ if TYPE_CHECKING:
COMMENT = "comment"
OWNER = "owner"
+# If set to true, HiveCatalog will operate in Hive2 compatibility mode
+HIVE2_COMPATIBLE = "hive.hive2-compatible"
+HIVE2_COMPATIBLE_DEFAULT = False
+
class _HiveClient:
"""Helper class to nicely open and close the transport."""
@@ -132,10 +136,15 @@ class _HiveClient:
self._transport.close()
-def _construct_hive_storage_descriptor(schema: Schema, location: Optional[str]) -> StorageDescriptor:
+def _construct_hive_storage_descriptor(
+ schema: Schema, location: Optional[str], hive2_compatible: bool = False
+) -> StorageDescriptor:
ser_de_info = SerDeInfo(serializationLib="org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe")
return StorageDescriptor(
- [FieldSchema(field.name, visit(field.field_type, SchemaToHiveConverter()), field.doc) for field in schema.fields],
+ [
+ FieldSchema(field.name, visit(field.field_type, SchemaToHiveConverter(hive2_compatible)), field.doc)
+ for field in schema.fields
+ ],
location,
"org.apache.hadoop.mapred.FileInputFormat",
"org.apache.hadoop.mapred.FileOutputFormat",
@@ -180,7 +189,7 @@ HIVE_PRIMITIVE_TYPES = {
DateType: "date",
TimeType: "string",
TimestampType: "timestamp",
- TimestamptzType: "timestamp",
+ TimestamptzType: "timestamp with local time zone",
StringType: "string",
UUIDType: "string",
BinaryType: "binary",
@@ -189,6 +198,11 @@ HIVE_PRIMITIVE_TYPES = {
class SchemaToHiveConverter(SchemaVisitor[str]):
+ hive2_compatible: bool
+
+ def __init__(self, hive2_compatible: bool):
+ self.hive2_compatible = hive2_compatible
+
def schema(self, schema: Schema, struct_result: str) -> str:
return struct_result
@@ -208,6 +222,9 @@ class SchemaToHiveConverter(SchemaVisitor[str]):
def primitive(self, primitive: PrimitiveType) -> str:
if isinstance(primitive, DecimalType):
return f"decimal({primitive.precision},{primitive.scale})"
+ elif self.hive2_compatible and isinstance(primitive, TimestamptzType):
+ # Hive2 doesn't support timestamp with local time zone
+ return "timestamp"
else:
return HIVE_PRIMITIVE_TYPES[type(primitive)]
@@ -296,7 +313,9 @@ class HiveCatalog(Catalog):
owner=properties[OWNER] if properties and OWNER in properties else getpass.getuser(),
createTime=current_time_millis // 1000,
lastAccessTime=current_time_millis // 1000,
- sd=_construct_hive_storage_descriptor(schema, location),
+ sd=_construct_hive_storage_descriptor(
+ schema, location, PropertyUtil.property_as_bool(self.properties, HIVE2_COMPATIBLE, HIVE2_COMPATIBLE_DEFAULT)
+ ),
tableType=EXTERNAL_TABLE,
parameters=_construct_parameters(metadata_location),
)
diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py
index 0f4e334d..c2d5409b 100644
--- a/pyiceberg/table/__init__.py
+++ b/pyiceberg/table/__init__.py
@@ -223,6 +223,12 @@ class PropertyUtil:
else:
return default
+ @staticmethod
+ def property_as_bool(properties: Dict[str, str], property_name: str, default: bool) -> bool:
+ if value := properties.get(property_name):
+ return value.lower() == "true"
+ return default
+
class Transaction:
_table: Table
diff --git a/tests/catalog/test_hive.py b/tests/catalog/test_hive.py
index 6a5e886f..396b762c 100644
--- a/tests/catalog/test_hive.py
+++ b/tests/catalog/test_hive.py
@@ -194,9 +194,14 @@ def test_check_number_of_namespaces(table_schema_simple: Schema) -> None:
catalog.create_table("table", schema=table_schema_simple)
+@pytest.mark.parametrize("hive2_compatible", [True, False])
@patch("time.time", MagicMock(return_value=12345))
-def test_create_table(table_schema_with_all_types: Schema, hive_database: HiveDatabase, hive_table: HiveTable) -> None:
+def test_create_table(
+ table_schema_with_all_types: Schema, hive_database: HiveDatabase, hive_table: HiveTable, hive2_compatible: bool
+) -> None:
catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
+ if hive2_compatible:
+ catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL, **{"hive.hive2-compatible": "true"})
catalog._client = MagicMock()
catalog._client.__enter__().create_table.return_value = None
@@ -229,7 +234,11 @@ def test_create_table(table_schema_with_all_types: Schema, hive_database: HiveDa
FieldSchema(name='date', type='date', comment=None),
FieldSchema(name='time', type='string', comment=None),
FieldSchema(name='timestamp', type='timestamp', comment=None),
- FieldSchema(name='timestamptz', type='timestamp', comment=None),
+ FieldSchema(
+ name='timestamptz',
+ type='timestamp' if hive2_compatible else 'timestamp with local time zone',
+ comment=None,
+ ),
FieldSchema(name='string', type='string', comment=None),
FieldSchema(name='uuid', type='string', comment=None),
FieldSchema(name='fixed', type='binary', comment=None),
diff --git a/tests/integration/test_writes.py b/tests/integration/test_writes.py
index 425e4cee..f1612b49 100644
--- a/tests/integration/test_writes.py
+++ b/tests/integration/test_writes.py
@@ -32,6 +32,8 @@ from pyspark.sql import SparkSession
from pytest_mock.plugin import MockerFixture
from pyiceberg.catalog import Catalog, load_catalog
+from pyiceberg.catalog import Catalog
+from pyiceberg.catalog.hive import HiveCatalog
from pyiceberg.catalog.sql import SqlCatalog
from pyiceberg.exceptions import NamespaceAlreadyExistsError, NoSuchTableError
from pyiceberg.schema import Schema
@@ -664,3 +666,21 @@ def test_duckdb_url_import(warehouse: Path, arrow_table_with_null: pa.Table) ->
b'\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11\x11',
),
]
+
+@pytest.mark.integration
+@pytest.mark.parametrize("format_version", [1, 2])
+def test_hive_catalog_storage_descriptor(
+ session_catalog_hive: HiveCatalog,
+ pa_schema: pa.Schema,
+ arrow_table_with_null: pa.Table,
+ spark: SparkSession,
+ format_version: int,
+) -> None:
+ tbl = _create_table(
+ session_catalog_hive, "default.test_storage_descriptor", {"format-version": format_version}, [arrow_table_with_null]
+ )
+
+ # check if pyiceberg can read the table
+ assert len(tbl.scan().to_arrow()) == 3
+ # check if spark can read the table
+ assert spark.sql("SELECT * FROM hive.default.test_storage_descriptor").count() == 3