This is an automated email from the ASF dual-hosted git repository.
kevinjqliu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
The following commit(s) were added to refs/heads/main by this push:
new f507dbd1 Leverage Iceberg-Rust for all the transforms (#1833)
f507dbd1 is described below
commit f507dbd1c29c7d21d24c6f2f7b4479cde2be16de
Author: Fokko Driesprong <[email protected]>
AuthorDate: Sat Jun 7 20:40:26 2025 +0200
Leverage Iceberg-Rust for all the transforms (#1833)
# Rationale for this change
Testing out to use Iceberg Rust for all of the transforms. I think we
have some rounding error in
https://github.com/apache/iceberg-rust/pull/1128/
Closes https://github.com/apache/iceberg-python/issues/1591
# Are these changes tested?
# Are there any user-facing changes?
<!-- In the case of user-facing changes, please add the changelog label.
-->
---------
Co-authored-by: Kevin Liu <[email protected]>
---
poetry.lock | 54 ++++++++---------
pyiceberg/transforms.py | 128 ++++++++++++---------------------------
pyproject.toml | 4 +-
tests/table/test_partitioning.py | 12 +---
tests/test_transforms.py | 35 +----------
5 files changed, 73 insertions(+), 160 deletions(-)
diff --git a/poetry.lock b/poetry.lock
index 09ea4104..f41ef7d8 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -58,7 +58,7 @@ description = "Happy Eyeballs for asyncio"
optional = true
python-versions = ">=3.9"
groups = ["main"]
-markers = "extra == \"adlfs\" or extra == \"gcsfs\" or extra == \"s3fs\""
+markers = "extra == \"s3fs\" or extra == \"adlfs\" or extra == \"gcsfs\""
files = [
{file = "aiohappyeyeballs-2.6.1-py3-none-any.whl", hash =
"sha256:f349ba8f4b75cb25c99c5c2d84e997e485204d2902a9597802b0371f09331fb8"},
{file = "aiohappyeyeballs-2.6.1.tar.gz", hash =
"sha256:c3f9d0113123803ccadfdf3f0faa505bc78e6a72d1cc4806cbd719826e943558"},
@@ -71,7 +71,7 @@ description = "Async http client/server framework (asyncio)"
optional = true
python-versions = ">=3.9"
groups = ["main"]
-markers = "extra == \"adlfs\" or extra == \"gcsfs\" or extra == \"s3fs\""
+markers = "extra == \"s3fs\" or extra == \"adlfs\" or extra == \"gcsfs\""
files = [
{file = "aiohttp-3.11.14-cp310-cp310-macosx_10_9_universal2.whl", hash =
"sha256:e2bc827c01f75803de77b134afdbf74fa74b62970eafdf190f3244931d7a5c0d"},
{file = "aiohttp-3.11.14-cp310-cp310-macosx_10_9_x86_64.whl", hash =
"sha256:e365034c5cf6cf74f57420b57682ea79e19eb29033399dd3f40de4d0171998fa"},
@@ -196,7 +196,7 @@ description = "aiosignal: a list of registered asynchronous
callbacks"
optional = true
python-versions = ">=3.9"
groups = ["main"]
-markers = "(extra == \"adlfs\" or extra == \"gcsfs\" or extra == \"s3fs\") and
(extra == \"adlfs\" or extra == \"gcsfs\" or extra == \"s3fs\" or extra ==
\"ray\")"
+markers = "(extra == \"s3fs\" or extra == \"adlfs\" or extra == \"gcsfs\") and
(extra == \"s3fs\" or extra == \"adlfs\" or extra == \"gcsfs\" or extra ==
\"ray\")"
files = [
{file = "aiosignal-1.3.2-py2.py3-none-any.whl", hash =
"sha256:45cde58e409a301715980c2b01d0c28bdde3770d8290b5eb2173759d9acb31a5"},
{file = "aiosignal-1.3.2.tar.gz", hash =
"sha256:a8c255c66fafb1e499c9351d0bf32ff2d8a0321595ebac3b93713656d2436f54"},
@@ -248,7 +248,7 @@ description = "Timeout context manager for asyncio programs"
optional = true
python-versions = ">=3.8"
groups = ["main"]
-markers = "(extra == \"adlfs\" or extra == \"gcsfs\" or extra == \"s3fs\") and
python_version <= \"3.10\""
+markers = "(extra == \"s3fs\" or extra == \"adlfs\" or extra == \"gcsfs\") and
python_version <= \"3.10\""
files = [
{file = "async_timeout-5.0.1-py3-none-any.whl", hash =
"sha256:39e3809566ff85354557ec2398b55e096c8364bacac9405a7a1fa429e77fe76c"},
{file = "async_timeout-5.0.1.tar.gz", hash =
"sha256:d9321a7a3d5a6a5e187e824d2fa0793ce379a202935782d555d6e9d2735677d3"},
@@ -265,7 +265,7 @@ files = [
{file = "attrs-25.3.0-py3-none-any.whl", hash =
"sha256:427318ce031701fea540783410126f03899a97ffc6f61596ad581ac2e40e3bc3"},
{file = "attrs-25.3.0.tar.gz", hash =
"sha256:75d7cefc7fb576747b2c81b4442d4d4a1ce0900973527c011d1030fd3bf4af1b"},
]
-markers = {main = "(extra == \"adlfs\" or extra == \"gcsfs\" or extra ==
\"s3fs\") and (extra == \"adlfs\" or extra == \"gcsfs\" or extra == \"s3fs\" or
extra == \"ray\")"}
+markers = {main = "(extra == \"s3fs\" or extra == \"adlfs\" or extra ==
\"gcsfs\") and (extra == \"s3fs\" or extra == \"adlfs\" or extra == \"gcsfs\"
or extra == \"ray\")"}
[package.extras]
benchmark = ["cloudpickle ; platform_python_implementation == \"CPython\"",
"hypothesis", "mypy (>=1.11.1) ; platform_python_implementation == \"CPython\"
and python_version >= \"3.10\"", "pympler", "pytest (>=4.3.0)",
"pytest-codspeed", "pytest-mypy-plugins ; platform_python_implementation ==
\"CPython\" and python_version >= \"3.10\"", "pytest-xdist[psutil]"]
@@ -467,7 +467,7 @@ files = [
{file = "boto3-1.37.3-py3-none-any.whl", hash =
"sha256:2063b40af99fd02f6228ff52397b552ff3353831edaf8d25cc04801827ab9794"},
{file = "boto3-1.37.3.tar.gz", hash =
"sha256:21f3ce0ef111297e63a6eb998a25197b8c10982970c320d4c6e8db08be2157be"},
]
-markers = {main = "extra == \"dynamodb\" or extra == \"glue\" or extra ==
\"rest-sigv4\""}
+markers = {main = "extra == \"glue\" or extra == \"dynamodb\" or extra ==
\"rest-sigv4\""}
[package.dependencies]
botocore = ">=1.37.3,<1.38.0"
@@ -488,7 +488,7 @@ files = [
{file = "botocore-1.37.3-py3-none-any.whl", hash =
"sha256:d01bd3bf4c80e61fa88d636ad9f5c9f60a551d71549b481386c6b4efe0bb2b2e"},
{file = "botocore-1.37.3.tar.gz", hash =
"sha256:fe8403eb55a88faf9b0f9da6615e5bee7be056d75e17af66c3c8f0a3b0648da4"},
]
-markers = {main = "extra == \"dynamodb\" or extra == \"glue\" or extra ==
\"rest-sigv4\" or extra == \"s3fs\""}
+markers = {main = "extra == \"glue\" or extra == \"dynamodb\" or extra ==
\"rest-sigv4\" or extra == \"s3fs\""}
[package.dependencies]
jmespath = ">=0.7.1,<2.0.0"
@@ -1449,7 +1449,7 @@ description = "A list-like structure which implements
collections.abc.MutableSeq
optional = true
python-versions = ">=3.8"
groups = ["main"]
-markers = "(extra == \"adlfs\" or extra == \"gcsfs\" or extra == \"s3fs\") and
(extra == \"adlfs\" or extra == \"gcsfs\" or extra == \"s3fs\" or extra ==
\"ray\")"
+markers = "(extra == \"s3fs\" or extra == \"adlfs\" or extra == \"gcsfs\") and
(extra == \"s3fs\" or extra == \"adlfs\" or extra == \"gcsfs\" or extra ==
\"ray\")"
files = [
{file = "frozenlist-1.5.0-cp310-cp310-macosx_10_9_universal2.whl", hash =
"sha256:5b6a66c18b5b9dd261ca98dffcb826a525334b2f29e7caa54e182255c5f6a65a"},
{file = "frozenlist-1.5.0-cp310-cp310-macosx_10_9_x86_64.whl", hash =
"sha256:d1b3eb7b05ea246510b43a7e53ed1653e55c2121019a97e60cad7efb881a97bb"},
@@ -2162,7 +2162,7 @@ files = [
{file = "jmespath-1.0.1-py3-none-any.whl", hash =
"sha256:02e2e4cc71b5bcab88332eebf907519190dd9e6e82107fa7f83b1003a6252980"},
{file = "jmespath-1.0.1.tar.gz", hash =
"sha256:90261b206d6defd58fdd5e85f478bf633a2901798906be2ad389150c5c60edbe"},
]
-markers = {main = "extra == \"dynamodb\" or extra == \"glue\" or extra ==
\"rest-sigv4\" or extra == \"s3fs\""}
+markers = {main = "extra == \"glue\" or extra == \"dynamodb\" or extra ==
\"rest-sigv4\" or extra == \"s3fs\""}
[[package]]
name = "joserfc"
@@ -2980,7 +2980,7 @@ description = "multidict implementation"
optional = true
python-versions = ">=3.9"
groups = ["main"]
-markers = "extra == \"adlfs\" or extra == \"gcsfs\" or extra == \"s3fs\""
+markers = "extra == \"s3fs\" or extra == \"adlfs\" or extra == \"gcsfs\""
files = [
{file = "multidict-6.2.0-cp310-cp310-macosx_10_9_universal2.whl", hash =
"sha256:b9f6392d98c0bd70676ae41474e2eecf4c7150cb419237a41f8f96043fcb81d1"},
{file = "multidict-6.2.0-cp310-cp310-macosx_10_9_x86_64.whl", hash =
"sha256:3501621d5e86f1a88521ea65d5cad0a0834c77b26f193747615b7c911e5422d2"},
@@ -3574,7 +3574,7 @@ description = "Accelerated property cache"
optional = true
python-versions = ">=3.9"
groups = ["main"]
-markers = "extra == \"adlfs\" or extra == \"gcsfs\" or extra == \"s3fs\""
+markers = "extra == \"s3fs\" or extra == \"adlfs\" or extra == \"gcsfs\""
files = [
{file = "propcache-0.3.1-cp310-cp310-macosx_10_9_universal2.whl", hash =
"sha256:f27785888d2fdd918bc36de8b8739f2d6c791399552333721b58193f68ea3e98"},
{file = "propcache-0.3.1-cp310-cp310-macosx_10_9_x86_64.whl", hash =
"sha256:d4e89cde74154c7b5957f87a355bb9c8ec929c167b59c83d90654ea36aeb6180"},
@@ -3882,7 +3882,7 @@ files = [
{file = "pyarrow-20.0.0-cp39-cp39-win_amd64.whl", hash =
"sha256:9965a050048ab02409fb7cbbefeedba04d3d67f2cc899eff505cc084345959ca"},
{file = "pyarrow-20.0.0.tar.gz", hash =
"sha256:febc4a913592573c8d5805091a6c2b5064c8bd6e002131f01061797d91c783c1"},
]
-markers = {main = "extra == \"daft\" or extra == \"duckdb\" or extra ==
\"pandas\" or extra == \"pyarrow\" or extra == \"ray\""}
+markers = {main = "extra == \"pyarrow\" or extra == \"pandas\" or extra ==
\"duckdb\" or extra == \"ray\" or extra == \"daft\""}
[package.extras]
test = ["cffi", "hypothesis", "pandas", "pytest", "pytz"]
@@ -4080,19 +4080,19 @@ windows-terminal = ["colorama (>=0.4.6)"]
[[package]]
name = "pyiceberg-core"
-version = "0.4.0"
+version = "0.5.1"
description = ""
optional = true
-python-versions = "*"
+python-versions = "~=3.9"
groups = ["main"]
-markers = "extra == \"pyiceberg-core\""
+markers = "extra == \"pyarrow\" or extra == \"pyiceberg-core\""
files = [
- {file =
"pyiceberg_core-0.4.0-cp39-abi3-macosx_10_12_x86_64.macosx_11_0_arm64.macosx_10_12_universal2.whl",
hash =
"sha256:5aec569271c96e18428d542f9b7007117a7232c06017f95cb239d42e952ad3b4"},
- {file =
"pyiceberg_core-0.4.0-cp39-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl",
hash =
"sha256:5e74773e58efa4df83aba6f6265cdd41e446fa66fa4e343ca86395fed9f209ae"},
- {file =
"pyiceberg_core-0.4.0-cp39-abi3-manylinux_2_17_armv7l.manylinux2014_armv7l.whl",
hash =
"sha256:7675d21a54bf3753c740d8df78ad7efe33f438096844e479d4f3493f84830925"},
- {file =
"pyiceberg_core-0.4.0-cp39-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl",
hash =
"sha256:7058ad935a40b1838e4cdc5febd768878c1a51f83dca005d5a52a7fa280a2489"},
- {file = "pyiceberg_core-0.4.0-cp39-abi3-win_amd64.whl", hash =
"sha256:a83eb4c2307ae3dd321a9360828fb043a4add2cc9797bef0bafa20894488fb07"},
- {file = "pyiceberg_core-0.4.0.tar.gz", hash =
"sha256:d2e6138707868477b806ed354aee9c476e437913a331cb9ad9ad46b4054cd11f"},
+ {file =
"pyiceberg_core-0.5.1-cp39-abi3-macosx_10_12_x86_64.macosx_11_0_arm64.macosx_10_12_universal2.whl",
hash =
"sha256:f5bffe70f036d4409ab5f8a2a52353635dfa9afa665e84eb5fb0415d12171aaa"},
+ {file =
"pyiceberg_core-0.5.1-cp39-abi3-manylinux_2_17_armv7l.manylinux2014_armv7l.whl",
hash =
"sha256:aa5d0d00efeb7c03ad48b17577452f42398f398b41034d2695a294c872365fcd"},
+ {file =
"pyiceberg_core-0.5.1-cp39-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl",
hash =
"sha256:6ae88f65f68898a680dde4d83179903d14b9ef2ca8f7b8715fca5d91a2aa24cd"},
+ {file = "pyiceberg_core-0.5.1-cp39-abi3-manylinux_2_28_aarch64.whl", hash
= "sha256:9d02f82292d2d01cc6b43f79eadf1209e84457cf0a2b945a2ed834245d6a45f6"},
+ {file = "pyiceberg_core-0.5.1-cp39-abi3-win_amd64.whl", hash =
"sha256:aad22d3ea743fcf941336542fcf565e16b28b9131087d70e1b03924005334af3"},
+ {file = "pyiceberg_core-0.5.1.tar.gz", hash =
"sha256:5359750e690169e06e4c20b74854f0048abbb5fe74b75919247d87cf2af2fcf4"},
]
[[package]]
@@ -4392,7 +4392,7 @@ description = "World timezone definitions, modern and
historical"
optional = true
python-versions = "*"
groups = ["main"]
-markers = "extra == \"gcsfs\" or extra == \"pandas\" or extra == \"ray\""
+markers = "extra == \"pandas\" or extra == \"ray\" or extra == \"gcsfs\""
files = [
{file = "pytz-2025.2-py2.py3-none-any.whl", hash =
"sha256:5ddf76296dd8c44c26eb8f4b6f35488f3ccbf6fbbd7adee0b7262d43f0ec2f00"},
{file = "pytz-2025.2.tar.gz", hash =
"sha256:360b9e3dbb49a209c21ad61809c7fb453643e048b38924c765813546746e81c3"},
@@ -4992,7 +4992,7 @@ files = [
{file = "s3transfer-0.11.3-py3-none-any.whl", hash =
"sha256:ca855bdeb885174b5ffa95b9913622459d4ad8e331fc98eb01e6d5eb6a30655d"},
{file = "s3transfer-0.11.3.tar.gz", hash =
"sha256:edae4977e3a122445660c7c114bba949f9d191bae3b34a096f18a1c8c354527a"},
]
-markers = {main = "extra == \"dynamodb\" or extra == \"glue\" or extra ==
\"rest-sigv4\""}
+markers = {main = "extra == \"glue\" or extra == \"dynamodb\" or extra ==
\"rest-sigv4\""}
[package.dependencies]
botocore = ">=1.36.0,<2.0a.0"
@@ -5426,7 +5426,7 @@ description = "Fast, Extensible Progress Meter"
optional = true
python-versions = ">=3.7"
groups = ["main"]
-markers = "extra == \"daft\" or extra == \"hf\""
+markers = "extra == \"hf\" or extra == \"daft\""
files = [
{file = "tqdm-4.67.1-py3-none-any.whl", hash =
"sha256:26445eca388f82e72884e0d580d5464cd801a3ea01e63e5601bdff9ba6a48de2"},
{file = "tqdm-4.67.1.tar.gz", hash =
"sha256:f8aef9c52c08c13a65f30ea34f4e5aac3fd1a34959879d7e59e63027286627f2"},
@@ -5726,7 +5726,7 @@ description = "Yet another URL library"
optional = true
python-versions = ">=3.9"
groups = ["main"]
-markers = "extra == \"adlfs\" or extra == \"gcsfs\" or extra == \"s3fs\""
+markers = "extra == \"s3fs\" or extra == \"adlfs\" or extra == \"gcsfs\""
files = [
{file = "yarl-1.18.3-cp310-cp310-macosx_10_9_universal2.whl", hash =
"sha256:7df647e8edd71f000a5208fe6ff8c382a1de8edfbccdbbfe649d263de07d8c34"},
{file = "yarl-1.18.3-cp310-cp310-macosx_10_9_x86_64.whl", hash =
"sha256:c69697d3adff5aa4f874b19c0e4ed65180ceed6318ec856ebc423aa5850d84f7"},
@@ -5964,7 +5964,7 @@ hive = ["thrift"]
hive-kerberos = ["kerberos", "thrift", "thrift-sasl"]
pandas = ["pandas", "pyarrow"]
polars = ["polars"]
-pyarrow = ["pyarrow"]
+pyarrow = ["pyarrow", "pyiceberg-core"]
pyiceberg-core = ["pyiceberg-core"]
ray = ["pandas", "pyarrow", "ray", "ray"]
rest-sigv4 = ["boto3"]
@@ -5977,4 +5977,4 @@ zstandard = ["zstandard"]
[metadata]
lock-version = "2.1"
python-versions = "^3.9.2, !=3.9.7"
-content-hash =
"fb2fa0b2764e8528206e2e401ad35a12b8edf31350d130796d5bb279d8785b58"
+content-hash =
"93eda79de3991bea0b9520f69372d146f7fd3d128cd982ea5afb78e8ec5d94f2"
diff --git a/pyiceberg/transforms.py b/pyiceberg/transforms.py
index 19889a98..8dda85a4 100644
--- a/pyiceberg/transforms.py
+++ b/pyiceberg/transforms.py
@@ -111,6 +111,36 @@ def _transform_literal(func: Callable[[L], L], lit:
Literal[L]) -> Literal[L]:
return literal(func(lit.value))
+def _pyiceberg_transform_wrapper(
+ transform_func: Callable[["ArrayLike", Any], "ArrayLike"],
+ *args: Any,
+ expected_type: Optional["pa.DataType"] = None,
+) -> Callable[["ArrayLike"], "ArrayLike"]:
+ try:
+ import pyarrow as pa
+ except ModuleNotFoundError as e:
+ raise ModuleNotFoundError("For partition transforms, PyArrow needs to
be installed") from e
+
+ def _transform(array: "ArrayLike") -> "ArrayLike":
+ def _cast_if_needed(arr: "ArrayLike") -> "ArrayLike":
+ if expected_type is not None:
+ return arr.cast(expected_type)
+ else:
+ return arr
+
+ if isinstance(array, pa.Array):
+ return _cast_if_needed(transform_func(array, *args))
+ elif isinstance(array, pa.ChunkedArray):
+ result_chunks = []
+ for arr in array.iterchunks():
+ result_chunks.append(_cast_if_needed(transform_func(arr,
*args)))
+ return pa.chunked_array(result_chunks)
+ else:
+ raise ValueError(f"PyArrow array can only be of type pa.Array or
pa.ChunkedArray, but found {type(array)}")
+
+ return _transform
+
+
class Transform(IcebergRootModel[str], ABC, Generic[S, T]):
"""Transform base class for concrete transforms.
@@ -175,27 +205,6 @@ class Transform(IcebergRootModel[str], ABC, Generic[S, T]):
@abstractmethod
def pyarrow_transform(self, source: IcebergType) -> "Callable[[pa.Array],
pa.Array]": ...
- def _pyiceberg_transform_wrapper(
- self, transform_func: Callable[["ArrayLike", Any], "ArrayLike"],
*args: Any
- ) -> Callable[["ArrayLike"], "ArrayLike"]:
- try:
- import pyarrow as pa
- except ModuleNotFoundError as e:
- raise ModuleNotFoundError("For bucket/truncate transforms, PyArrow
needs to be installed") from e
-
- def _transform(array: "ArrayLike") -> "ArrayLike":
- if isinstance(array, pa.Array):
- return transform_func(array, *args)
- elif isinstance(array, pa.ChunkedArray):
- result_chunks = []
- for arr in array.iterchunks():
- result_chunks.append(transform_func(arr, *args))
- return pa.chunked_array(result_chunks)
- else:
- raise ValueError(f"PyArrow array can only be of type pa.Array
or pa.ChunkedArray, but found {type(array)}")
-
- return _transform
-
def parse_transform(v: Any) -> Transform[Any, Any]:
if isinstance(v, str):
@@ -375,7 +384,7 @@ class BucketTransform(Transform[S, int]):
def pyarrow_transform(self, source: IcebergType) -> "Callable[[pa.Array],
pa.Array]":
from pyiceberg_core import transform as pyiceberg_core_transform
- return
self._pyiceberg_transform_wrapper(pyiceberg_core_transform.bucket,
self._num_buckets)
+ return _pyiceberg_transform_wrapper(pyiceberg_core_transform.bucket,
self._num_buckets)
@property
def supports_pyarrow_transform(self) -> bool:
@@ -501,22 +510,9 @@ class YearTransform(TimeTransform[S]):
def pyarrow_transform(self, source: IcebergType) -> "Callable[[pa.Array],
pa.Array]":
import pyarrow as pa
- import pyarrow.compute as pc
-
- if isinstance(source, DateType):
- epoch = pa.scalar(datetime.EPOCH_DATE)
- elif isinstance(source, TimestampType):
- epoch = pa.scalar(datetime.EPOCH_TIMESTAMP)
- elif isinstance(source, TimestamptzType):
- epoch = pa.scalar(datetime.EPOCH_TIMESTAMPTZ)
- elif isinstance(source, TimestampNanoType):
- epoch =
pa.scalar(datetime.EPOCH_TIMESTAMP).cast(pa.timestamp("ns"))
- elif isinstance(source, TimestamptzNanoType):
- epoch =
pa.scalar(datetime.EPOCH_TIMESTAMPTZ).cast(pa.timestamp("ns"))
- else:
- raise ValueError(f"Cannot apply year transform for type: {source}")
+ from pyiceberg_core import transform as pyiceberg_core_transform
- return lambda v: pc.years_between(epoch, v) if v is not None else None
+ return _pyiceberg_transform_wrapper(pyiceberg_core_transform.year,
expected_type=pa.int32())
class MonthTransform(TimeTransform[S]):
@@ -575,28 +571,9 @@ class MonthTransform(TimeTransform[S]):
def pyarrow_transform(self, source: IcebergType) -> "Callable[[pa.Array],
pa.Array]":
import pyarrow as pa
- import pyarrow.compute as pc
-
- if isinstance(source, DateType):
- epoch = pa.scalar(datetime.EPOCH_DATE)
- elif isinstance(source, TimestampType):
- epoch = pa.scalar(datetime.EPOCH_TIMESTAMP)
- elif isinstance(source, TimestamptzType):
- epoch = pa.scalar(datetime.EPOCH_TIMESTAMPTZ)
- elif isinstance(source, TimestampNanoType):
- epoch =
pa.scalar(datetime.EPOCH_TIMESTAMP).cast(pa.timestamp("ns"))
- elif isinstance(source, TimestamptzNanoType):
- epoch =
pa.scalar(datetime.EPOCH_TIMESTAMPTZ).cast(pa.timestamp("ns"))
- else:
- raise ValueError(f"Cannot apply month transform for type:
{source}")
-
- def month_func(v: pa.Array) -> pa.Array:
- return pc.add(
- pc.multiply(pc.years_between(epoch, v), pa.scalar(12)),
- pc.add(pc.month(v), pa.scalar(-1)),
- )
+ from pyiceberg_core import transform as pyiceberg_core_transform
- return lambda v: month_func(v) if v is not None else None
+ return _pyiceberg_transform_wrapper(pyiceberg_core_transform.month,
expected_type=pa.int32())
class DayTransform(TimeTransform[S]):
@@ -663,22 +640,9 @@ class DayTransform(TimeTransform[S]):
def pyarrow_transform(self, source: IcebergType) -> "Callable[[pa.Array],
pa.Array]":
import pyarrow as pa
- import pyarrow.compute as pc
-
- if isinstance(source, DateType):
- epoch = pa.scalar(datetime.EPOCH_DATE)
- elif isinstance(source, TimestampType):
- epoch = pa.scalar(datetime.EPOCH_TIMESTAMP)
- elif isinstance(source, TimestamptzType):
- epoch = pa.scalar(datetime.EPOCH_TIMESTAMPTZ)
- elif isinstance(source, TimestampNanoType):
- epoch =
pa.scalar(datetime.EPOCH_TIMESTAMP).cast(pa.timestamp("ns"))
- elif isinstance(source, TimestamptzNanoType):
- epoch =
pa.scalar(datetime.EPOCH_TIMESTAMPTZ).cast(pa.timestamp("ns"))
- else:
- raise ValueError(f"Cannot apply day transform for type: {source}")
+ from pyiceberg_core import transform as pyiceberg_core_transform
- return lambda v: pc.days_between(epoch, v) if v is not None else None
+ return _pyiceberg_transform_wrapper(pyiceberg_core_transform.day,
expected_type=pa.int32())
class HourTransform(TimeTransform[S]):
@@ -728,21 +692,9 @@ class HourTransform(TimeTransform[S]):
return "HourTransform()"
def pyarrow_transform(self, source: IcebergType) -> "Callable[[pa.Array],
pa.Array]":
- import pyarrow as pa
- import pyarrow.compute as pc
-
- if isinstance(source, TimestampType):
- epoch = pa.scalar(datetime.EPOCH_TIMESTAMP)
- elif isinstance(source, TimestamptzType):
- epoch = pa.scalar(datetime.EPOCH_TIMESTAMPTZ)
- elif isinstance(source, TimestampNanoType):
- epoch =
pa.scalar(datetime.EPOCH_TIMESTAMP).cast(pa.timestamp("ns"))
- elif isinstance(source, TimestamptzNanoType):
- epoch =
pa.scalar(datetime.EPOCH_TIMESTAMPTZ).cast(pa.timestamp("ns"))
- else:
- raise ValueError(f"Cannot apply hour transform for type: {source}")
+ from pyiceberg_core import transform as pyiceberg_core_transform
- return lambda v: pc.hours_between(epoch, v) if v is not None else None
+ return _pyiceberg_transform_wrapper(pyiceberg_core_transform.hour)
def _base64encode(buffer: bytes) -> str:
@@ -965,7 +917,7 @@ class TruncateTransform(Transform[S, S]):
def pyarrow_transform(self, source: IcebergType) -> "Callable[[pa.Array],
pa.Array]":
from pyiceberg_core import transform as pyiceberg_core_transform
- return
self._pyiceberg_transform_wrapper(pyiceberg_core_transform.truncate,
self._width)
+ return _pyiceberg_transform_wrapper(pyiceberg_core_transform.truncate,
self._width)
@property
def supports_pyarrow_transform(self) -> bool:
diff --git a/pyproject.toml b/pyproject.toml
index d906dff6..9699f285 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -81,7 +81,7 @@ psycopg2-binary = { version = ">=2.9.6", optional = true }
sqlalchemy = { version = "^2.0.18", optional = true }
getdaft = { version = ">=0.2.12", optional = true }
cachetools = ">=5.5,<7.0"
-pyiceberg-core = { version = "^0.4.0", optional = true }
+pyiceberg-core = { version = "^0.5.1", optional = true }
polars = { version = "^1.21.0", optional = true }
thrift-sasl = { version = ">=0.4.3", optional = true }
kerberos = {version = "^1.3.1", optional = true}
@@ -289,7 +289,7 @@ generate-setup-file = false
script = "build-module.py"
[tool.poetry.extras]
-pyarrow = ["pyarrow"]
+pyarrow = ["pyarrow", "pyiceberg-core"]
pandas = ["pandas", "pyarrow"]
duckdb = ["duckdb", "pyarrow"]
ray = ["ray", "pyarrow", "pandas"]
diff --git a/tests/table/test_partitioning.py b/tests/table/test_partitioning.py
index 57ab3e32..0fe22391 100644
--- a/tests/table/test_partitioning.py
+++ b/tests/table/test_partitioning.py
@@ -186,8 +186,8 @@ def test_partition_type(table_schema_simple: Schema) ->
None:
(DecimalType(5, 9), Decimal(19.25)),
(DateType(), datetime.date(1925, 5, 22)),
(TimeType(), datetime.time(19, 25, 00)),
- (TimestampType(), datetime.datetime(19, 5, 1, 22, 1, 1)),
- (TimestamptzType(), datetime.datetime(19, 5, 1, 22, 1, 1,
tzinfo=datetime.timezone.utc)),
+ (TimestampType(), datetime.datetime(2022, 5, 1, 22, 1, 1)),
+ (TimestamptzType(), datetime.datetime(2022, 5, 1, 22, 1, 1,
tzinfo=datetime.timezone.utc)),
(StringType(), "abc"),
(UUIDType(), UUID("12345678-1234-5678-1234-567812345678").bytes),
(FixedType(5), 'b"\x8e\xd1\x87\x01"'),
@@ -208,13 +208,7 @@ def
test_transform_consistency_with_pyarrow_transform(source_type: PrimitiveType
]
for t in all_transforms:
if t.can_transform(source_type):
- try:
- assert t.transform(source_type)(value) ==
t.pyarrow_transform(source_type)(pa.array([value])).to_pylist()[0]
- except ValueError as e:
- # Skipping unsupported feature
- if "FeatureUnsupported => Unsupported data type for truncate
transform" in str(e):
- continue
- raise
+ assert t.transform(source_type)(value) ==
t.pyarrow_transform(source_type)(pa.array([value])).to_pylist()[0]
def test_deserialize_partition_field_v2() -> None:
diff --git a/tests/test_transforms.py b/tests/test_transforms.py
index d22c94cc..f8d3ac9d 100644
--- a/tests/test_transforms.py
+++ b/tests/test_transforms.py
@@ -16,7 +16,7 @@
# specific language governing permissions and limitations
# under the License.
# pylint: disable=eval-used,protected-access,redefined-outer-name
-from datetime import date, datetime
+from datetime import date
from decimal import Decimal
from typing import Annotated, Any, Callable, Optional, Union
from uuid import UUID
@@ -24,7 +24,6 @@ from uuid import UUID
import mmh3 as mmh3
import pyarrow as pa
import pytest
-import pytz
from pydantic import (
BeforeValidator,
PlainSerializer,
@@ -1654,38 +1653,6 @@ def test_bucket_pyarrow_transforms(
assert expected == transform.pyarrow_transform(source_type)(input_arr)
-# pyiceberg_core currently does not support bucket transform on timestamp_ns
and timestamptz_ns
-# https://github.com/apache/iceberg-rust/issues/1110
[email protected](
- "source_type, input_arr, num_buckets",
- [
- (
- TimestampNanoType(),
- pa.array([datetime(1970, 1, 1, 0, 0, 0), datetime(2025, 2, 26, 1,
2, 3)], type=pa.timestamp(unit="ns")),
- 10,
- ),
- (
- TimestamptzNanoType(),
- pa.array(
- [datetime(1970, 1, 1, 0, 0, 0), datetime(2025, 2, 26, 1, 2,
3)],
- type=pa.timestamp(unit="ns", tz=pytz.timezone("Etc/GMT+10")),
- ),
- 10,
- ),
- ],
-)
-def test_unsupported_bucket_pyarrow_transform(
- source_type: PrimitiveType,
- input_arr: Union[pa.Array, pa.ChunkedArray],
- num_buckets: int,
-) -> None:
- transform: Transform[Any, Any] = BucketTransform(num_buckets=num_buckets)
- with pytest.raises(ValueError) as exc_info:
- transform.pyarrow_transform(source_type)(input_arr)
-
- assert "FeatureUnsupported => Unsupported data type for bucket transform"
in str(exc_info.value)
-
-
@pytest.mark.parametrize(
"source_type, input_arr, expected, width",
[
