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:05 UTC
(iceberg-python) branch pyiceberg-0.6.x updated (86ff66e3 -> 8d52993d)
This is an automated email from the ASF dual-hosted git repository.
fokko pushed a change to branch pyiceberg-0.6.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
from 86ff66e3 fix integration test
new e1a7265d add TimestamptzType to storageDescriptor construction
new fd30bc4c [Bug Fix] Allow HiveCatalog to create table with TimestamptzType (#585)
new 929d0b42 fix lint issue
new 363b2374 fix integration tests
new 8d52993d remove the new public method
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
mkdocs/docs/configuration.md | 9 ++++
pyiceberg/catalog/__init__.py | 6 +++
pyiceberg/catalog/glue.py | 5 ++-
pyiceberg/catalog/hive.py | 44 +++++++++++---------
tests/catalog/test_hive.py | 88 ++++++++++++++++++++++++++++++++++++----
tests/conftest.py | 54 ++++++++++++++++++++++++
tests/integration/test_writes.py | 41 +++++++++++++++++++
7 files changed, 217 insertions(+), 30 deletions(-)
(iceberg-python) 03/05: fix lint issue
Posted by fo...@apache.org.
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 929d0b42ee8e2aa21fbb14bdcac423c4b65f9cef
Author: HonahX <ho...@apache.org>
AuthorDate: Sat Apr 13 15:07:46 2024 -0700
fix lint issue
---
pyiceberg/catalog/glue.py | 2 +-
tests/integration/test_writes.py | 4 ++--
2 files changed, 3 insertions(+), 3 deletions(-)
diff --git a/pyiceberg/catalog/glue.py b/pyiceberg/catalog/glue.py
index 97393881..291c0667 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, PropertyUtil
+from pyiceberg.table import CommitTableRequest, CommitTableResponse, PropertyUtil, Table, update_table_metadata
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
diff --git a/tests/integration/test_writes.py b/tests/integration/test_writes.py
index f1612b49..873966ca 100644
--- a/tests/integration/test_writes.py
+++ b/tests/integration/test_writes.py
@@ -32,7 +32,6 @@ 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
@@ -667,6 +666,7 @@ def test_duckdb_url_import(warehouse: Path, arrow_table_with_null: pa.Table) ->
),
]
+
@pytest.mark.integration
@pytest.mark.parametrize("format_version", [1, 2])
def test_hive_catalog_storage_descriptor(
@@ -677,7 +677,7 @@ def test_hive_catalog_storage_descriptor(
format_version: int,
) -> None:
tbl = _create_table(
- session_catalog_hive, "default.test_storage_descriptor", {"format-version": format_version}, [arrow_table_with_null]
+ session_catalog_hive, "default.test_storage_descriptor", {"format-version": str(format_version)}, [arrow_table_with_null]
)
# check if pyiceberg can read the table
(iceberg-python) 02/05: [Bug Fix] Allow HiveCatalog to create table with TimestamptzType (#585)
Posted by fo...@apache.org.
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
(iceberg-python) 05/05: remove the new public method
Posted by fo...@apache.org.
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 8d52993deb78cfd936a6882fe407f299c80d006d
Author: HonahX <ho...@apache.org>
AuthorDate: Sat Apr 13 16:06:08 2024 -0700
remove the new public method
---
pyiceberg/catalog/__init__.py | 6 ++++++
pyiceberg/catalog/glue.py | 5 +++--
pyiceberg/catalog/hive.py | 5 +++--
pyiceberg/table/__init__.py | 6 ------
4 files changed, 12 insertions(+), 10 deletions(-)
diff --git a/pyiceberg/catalog/__init__.py b/pyiceberg/catalog/__init__.py
index e14b4c94..d12f8be4 100644
--- a/pyiceberg/catalog/__init__.py
+++ b/pyiceberg/catalog/__init__.py
@@ -257,6 +257,12 @@ def delete_data_files(io: FileIO, manifests_to_delete: List[ManifestFile]) -> No
deleted_files[path] = True
+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
+
+
@dataclass
class PropertiesUpdateSummary:
removed: List[str]
diff --git a/pyiceberg/catalog/glue.py b/pyiceberg/catalog/glue.py
index 291c0667..2d30333b 100644
--- a/pyiceberg/catalog/glue.py
+++ b/pyiceberg/catalog/glue.py
@@ -47,6 +47,7 @@ from pyiceberg.catalog import (
TABLE_TYPE,
Catalog,
PropertiesUpdateSummary,
+ _property_as_bool,
)
from pyiceberg.exceptions import (
CommitFailedException,
@@ -62,7 +63,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, PropertyUtil, Table, update_table_metadata
+from pyiceberg.table import CommitTableRequest, CommitTableResponse, Table, update_table_metadata
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 +314,7 @@ class GlueCatalog(Catalog):
self.glue.update_table(
DatabaseName=database_name,
TableInput=table_input,
- SkipArchive=PropertyUtil.property_as_bool(self.properties, GLUE_SKIP_ARCHIVE, GLUE_SKIP_ARCHIVE_DEFAULT),
+ SkipArchive=_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 d97ddd80..f5436264 100644
--- a/pyiceberg/catalog/hive.py
+++ b/pyiceberg/catalog/hive.py
@@ -60,6 +60,7 @@ from pyiceberg.catalog import (
TABLE_TYPE,
Catalog,
PropertiesUpdateSummary,
+ _property_as_bool,
)
from pyiceberg.exceptions import (
CommitFailedException,
@@ -74,7 +75,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, PropertyUtil, Table, TableProperties, update_table_metadata
+from pyiceberg.table import CommitTableRequest, CommitTableResponse, 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
@@ -314,7 +315,7 @@ class HiveCatalog(Catalog):
createTime=current_time_millis // 1000,
lastAccessTime=current_time_millis // 1000,
sd=_construct_hive_storage_descriptor(
- schema, location, PropertyUtil.property_as_bool(self.properties, HIVE2_COMPATIBLE, HIVE2_COMPATIBLE_DEFAULT)
+ schema, location, _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 c2d5409b..0f4e334d 100644
--- a/pyiceberg/table/__init__.py
+++ b/pyiceberg/table/__init__.py
@@ -223,12 +223,6 @@ 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
(iceberg-python) 01/05: add TimestamptzType to storageDescriptor construction
Posted by fo...@apache.org.
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 e1a7265dc8bffa9de24e457f5cbeec9741fe675b
Author: HonahX <ho...@apache.org>
AuthorDate: Fri Apr 5 21:37:05 2024 -0700
add TimestamptzType to storageDescriptor construction
---
pyiceberg/catalog/glue.py | 2 +-
pyiceberg/catalog/hive.py | 18 ++---------
tests/catalog/test_hive.py | 79 ++++++++++++++++++++++++++++++++++++++++------
tests/conftest.py | 54 +++++++++++++++++++++++++++++++
4 files changed, 127 insertions(+), 26 deletions(-)
diff --git a/pyiceberg/catalog/glue.py b/pyiceberg/catalog/glue.py
index a32a79a9..6f93fab9 100644
--- a/pyiceberg/catalog/glue.py
+++ b/pyiceberg/catalog/glue.py
@@ -150,7 +150,7 @@ class _IcebergSchemaToGlueType(SchemaVisitor[str]):
if isinstance(primitive, DecimalType):
return f"decimal({primitive.precision},{primitive.scale})"
if (primitive_type := type(primitive)) not in GLUE_PRIMITIVE_TYPES:
- return str(primitive_type.root)
+ return str(primitive)
return GLUE_PRIMITIVE_TYPES[primitive_type]
diff --git a/pyiceberg/catalog/hive.py b/pyiceberg/catalog/hive.py
index b383063e..fb1f94f3 100644
--- a/pyiceberg/catalog/hive.py
+++ b/pyiceberg/catalog/hive.py
@@ -95,6 +95,7 @@ from pyiceberg.types import (
StringType,
StructType,
TimestampType,
+ TimestamptzType,
TimeType,
UUIDType,
)
@@ -103,22 +104,6 @@ if TYPE_CHECKING:
import pyarrow as pa
-# Replace by visitor
-hive_types = {
- BooleanType: "boolean",
- IntegerType: "int",
- LongType: "bigint",
- FloatType: "float",
- DoubleType: "double",
- DateType: "date",
- TimeType: "string",
- TimestampType: "timestamp",
- StringType: "string",
- UUIDType: "string",
- BinaryType: "binary",
- FixedType: "binary",
-}
-
COMMENT = "comment"
OWNER = "owner"
@@ -195,6 +180,7 @@ HIVE_PRIMITIVE_TYPES = {
DateType: "date",
TimeType: "string",
TimestampType: "timestamp",
+ TimestamptzType: "timestamp",
StringType: "string",
UUIDType: "string",
BinaryType: "binary",
diff --git a/tests/catalog/test_hive.py b/tests/catalog/test_hive.py
index dc2689e0..6a5e886f 100644
--- a/tests/catalog/test_hive.py
+++ b/tests/catalog/test_hive.py
@@ -61,11 +61,24 @@ from pyiceberg.table.sorting import (
from pyiceberg.transforms import BucketTransform, IdentityTransform
from pyiceberg.typedef import UTF8
from pyiceberg.types import (
+ BinaryType,
BooleanType,
+ DateType,
+ DecimalType,
+ DoubleType,
+ FixedType,
+ FloatType,
IntegerType,
+ ListType,
LongType,
+ MapType,
NestedField,
StringType,
+ StructType,
+ TimestampType,
+ TimestamptzType,
+ TimeType,
+ UUIDType,
)
HIVE_CATALOG_NAME = "hive"
@@ -182,14 +195,14 @@ def test_check_number_of_namespaces(table_schema_simple: Schema) -> None:
@patch("time.time", MagicMock(return_value=12345))
-def test_create_table(table_schema_simple: Schema, hive_database: HiveDatabase, hive_table: HiveTable) -> None:
+def test_create_table(table_schema_with_all_types: Schema, hive_database: HiveDatabase, hive_table: HiveTable) -> None:
catalog = HiveCatalog(HIVE_CATALOG_NAME, uri=HIVE_METASTORE_FAKE_URL)
catalog._client = MagicMock()
catalog._client.__enter__().create_table.return_value = None
catalog._client.__enter__().get_table.return_value = hive_table
catalog._client.__enter__().get_database.return_value = hive_database
- catalog.create_table(("default", "table"), schema=table_schema_simple, properties={"owner": "javaberg"})
+ catalog.create_table(("default", "table"), schema=table_schema_with_all_types, properties={"owner": "javaberg"})
called_hive_table: HiveTable = catalog._client.__enter__().create_table.call_args[0][0]
# This one is generated within the function itself, so we need to extract
@@ -207,9 +220,23 @@ def test_create_table(table_schema_simple: Schema, hive_database: HiveDatabase,
retention=None,
sd=StorageDescriptor(
cols=[
- FieldSchema(name="foo", type="string", comment=None),
- FieldSchema(name="bar", type="int", comment=None),
- FieldSchema(name="baz", type="boolean", comment=None),
+ FieldSchema(name='boolean', type='boolean', comment=None),
+ FieldSchema(name='integer', type='int', comment=None),
+ FieldSchema(name='long', type='bigint', comment=None),
+ FieldSchema(name='float', type='float', comment=None),
+ FieldSchema(name='double', type='double', comment=None),
+ FieldSchema(name='decimal', type='decimal(32,3)', comment=None),
+ 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='string', type='string', comment=None),
+ FieldSchema(name='uuid', type='string', comment=None),
+ FieldSchema(name='fixed', type='binary', comment=None),
+ FieldSchema(name='binary', type='binary', comment=None),
+ FieldSchema(name='list', type='array<string>', comment=None),
+ FieldSchema(name='map', type='map<string,int>', comment=None),
+ FieldSchema(name='struct', type='struct<inner_string:string,inner_int:int>', comment=None),
],
location=f"{hive_database.locationUri}/table",
inputFormat="org.apache.hadoop.mapred.FileInputFormat",
@@ -266,12 +293,46 @@ def test_create_table(table_schema_simple: Schema, hive_database: HiveDatabase,
location=metadata.location,
table_uuid=metadata.table_uuid,
last_updated_ms=metadata.last_updated_ms,
- last_column_id=3,
+ last_column_id=22,
schemas=[
Schema(
- NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
- NestedField(field_id=2, name="bar", field_type=IntegerType(), required=True),
- NestedField(field_id=3, name="baz", field_type=BooleanType(), required=False),
+ NestedField(field_id=1, name='boolean', field_type=BooleanType(), required=True),
+ NestedField(field_id=2, name='integer', field_type=IntegerType(), required=True),
+ NestedField(field_id=3, name='long', field_type=LongType(), required=True),
+ NestedField(field_id=4, name='float', field_type=FloatType(), required=True),
+ NestedField(field_id=5, name='double', field_type=DoubleType(), required=True),
+ NestedField(field_id=6, name='decimal', field_type=DecimalType(precision=32, scale=3), required=True),
+ NestedField(field_id=7, name='date', field_type=DateType(), required=True),
+ NestedField(field_id=8, name='time', field_type=TimeType(), required=True),
+ NestedField(field_id=9, name='timestamp', field_type=TimestampType(), required=True),
+ NestedField(field_id=10, name='timestamptz', field_type=TimestamptzType(), required=True),
+ NestedField(field_id=11, name='string', field_type=StringType(), required=True),
+ NestedField(field_id=12, name='uuid', field_type=UUIDType(), required=True),
+ NestedField(field_id=13, name='fixed', field_type=FixedType(length=12), required=True),
+ NestedField(field_id=14, name='binary', field_type=BinaryType(), required=True),
+ NestedField(
+ field_id=15,
+ name='list',
+ field_type=ListType(type='list', element_id=18, element_type=StringType(), element_required=True),
+ required=True,
+ ),
+ NestedField(
+ field_id=16,
+ name='map',
+ field_type=MapType(
+ type='map', key_id=19, key_type=StringType(), value_id=20, value_type=IntegerType(), value_required=True
+ ),
+ required=True,
+ ),
+ NestedField(
+ field_id=17,
+ name='struct',
+ field_type=StructType(
+ NestedField(field_id=21, name='inner_string', field_type=StringType(), required=False),
+ NestedField(field_id=22, name='inner_int', field_type=IntegerType(), required=True),
+ ),
+ required=True,
+ ),
schema_id=0,
identifier_field_ids=[2],
)
diff --git a/tests/conftest.py b/tests/conftest.py
index 46fa4a8e..7ad12ddc 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -72,7 +72,9 @@ from pyiceberg.types import (
BinaryType,
BooleanType,
DateType,
+ DecimalType,
DoubleType,
+ FixedType,
FloatType,
IntegerType,
ListType,
@@ -81,6 +83,10 @@ from pyiceberg.types import (
NestedField,
StringType,
StructType,
+ TimestampType,
+ TimestamptzType,
+ TimeType,
+ UUIDType,
)
from pyiceberg.utils.datetime import datetime_to_millis
@@ -267,6 +273,54 @@ def table_schema_nested_with_struct_key_map() -> Schema:
)
+@pytest.fixture(scope="session")
+def table_schema_with_all_types() -> Schema:
+ return schema.Schema(
+ NestedField(field_id=1, name="boolean", field_type=BooleanType(), required=True),
+ NestedField(field_id=2, name="integer", field_type=IntegerType(), required=True),
+ NestedField(field_id=3, name="long", field_type=LongType(), required=True),
+ NestedField(field_id=4, name="float", field_type=FloatType(), required=True),
+ NestedField(field_id=5, name="double", field_type=DoubleType(), required=True),
+ NestedField(field_id=6, name="decimal", field_type=DecimalType(32, 3), required=True),
+ NestedField(field_id=7, name="date", field_type=DateType(), required=True),
+ NestedField(field_id=8, name="time", field_type=TimeType(), required=True),
+ NestedField(field_id=9, name="timestamp", field_type=TimestampType(), required=True),
+ NestedField(field_id=10, name="timestamptz", field_type=TimestamptzType(), required=True),
+ NestedField(field_id=11, name="string", field_type=StringType(), required=True),
+ NestedField(field_id=12, name="uuid", field_type=UUIDType(), required=True),
+ NestedField(field_id=14, name="fixed", field_type=FixedType(12), required=True),
+ NestedField(field_id=13, name="binary", field_type=BinaryType(), required=True),
+ NestedField(
+ field_id=15,
+ name="list",
+ field_type=ListType(element_id=16, element_type=StringType(), element_required=True),
+ required=True,
+ ),
+ NestedField(
+ field_id=17,
+ name="map",
+ field_type=MapType(
+ key_id=18,
+ key_type=StringType(),
+ value_id=19,
+ value_type=IntegerType(),
+ value_required=True,
+ ),
+ required=True,
+ ),
+ NestedField(
+ field_id=20,
+ name="struct",
+ field_type=StructType(
+ NestedField(field_id=21, name="inner_string", field_type=StringType(), required=False),
+ NestedField(field_id=22, name="inner_int", field_type=IntegerType(), required=True),
+ ),
+ ),
+ schema_id=1,
+ identifier_field_ids=[2],
+ )
+
+
@pytest.fixture(scope="session")
def pyarrow_schema_simple_without_ids() -> "pa.Schema":
import pyarrow as pa
(iceberg-python) 04/05: fix integration tests
Posted by fo...@apache.org.
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 363b23744c2d15c2da59ea438aa41743de9a7c6c
Author: HonahX <ho...@apache.org>
AuthorDate: Sat Apr 13 15:37:12 2024 -0700
fix integration tests
---
tests/integration/test_writes.py | 21 +++++++++++++++++++++
1 file changed, 21 insertions(+)
diff --git a/tests/integration/test_writes.py b/tests/integration/test_writes.py
index 873966ca..aaeefff5 100644
--- a/tests/integration/test_writes.py
+++ b/tests/integration/test_writes.py
@@ -131,6 +131,20 @@ def session_catalog() -> Catalog:
)
+@pytest.fixture(scope="session")
+def session_catalog_hive() -> Catalog:
+ return load_catalog(
+ "local",
+ **{
+ "type": "hive",
+ "uri": "http://localhost:9083",
+ "s3.endpoint": "http://localhost:9000",
+ "s3.access-key-id": "admin",
+ "s3.secret-access-key": "password",
+ },
+ )
+
+
@pytest.fixture(scope="session")
def pa_schema() -> pa.Schema:
return pa.schema([
@@ -289,6 +303,13 @@ def spark() -> SparkSession:
.config("spark.sql.catalog.integration.s3.endpoint", "http://localhost:9000")
.config("spark.sql.catalog.integration.s3.path-style-access", "true")
.config("spark.sql.defaultCatalog", "integration")
+ .config("spark.sql.catalog.hive", "org.apache.iceberg.spark.SparkCatalog")
+ .config("spark.sql.catalog.hive.type", "hive")
+ .config("spark.sql.catalog.hive.uri", "http://localhost:9083")
+ .config("spark.sql.catalog.hive.io-impl", "org.apache.iceberg.aws.s3.S3FileIO")
+ .config("spark.sql.catalog.hive.warehouse", "s3://warehouse/hive/")
+ .config("spark.sql.catalog.hive.s3.endpoint", "http://localhost:9000")
+ .config("spark.sql.catalog.hive.s3.path-style-access", "true")
.getOrCreate()
)