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 2025/04/17 12:40:27 UTC
(iceberg-python) branch pyiceberg-0.9.x updated (8bfb16cf -> 562de407)
This is an automated email from the ASF dual-hosted git repository.
fokko pushed a change to branch pyiceberg-0.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
from 8bfb16cf fix loading `in-memory` catalog (#1725)
new 998ce7cc Remove unused argument (#1734)
new b989197f Fix thrift client connection for Kerberos Hive Client (#1747)
new 9da7bd2b Use a balanced tree instead of unbalanced one (#1830)
new 903fb917 Fix decimal physicial type mapping (#1839)
new cfb4f251 fix `upsert` with null values (#1861)
new 29e94e75 Set field-id when needed (#1867)
new bc7e1cc7 Fix the snapshot summary of a partial overwrite (#1879)
new e9745d5f Fix creation of Bucket Transforms with `pydantic>=2.11.0` (#1881)
new e5d1bb87 Temporary fix for filtering on empty batches (#1901)
new ddd72250 Fix for metadata entries table for MOR tables containing Delete Files. (#1902)
new 562de407 Fix `add_files` with non-identity transforms (#1925)
The 11 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:
pyiceberg/catalog/hive.py | 36 +++++---
pyiceberg/expressions/__init__.py | 47 +++++++++-
pyiceberg/io/pyarrow.py | 53 +++++++----
pyiceberg/table/__init__.py | 2 -
pyiceberg/table/inspect.py | 6 +-
pyiceberg/table/snapshots.py | 6 ++
pyiceberg/table/update/snapshot.py | 1 -
pyiceberg/table/upsert_util.py | 19 +++-
pyiceberg/transforms.py | 2 +-
tests/expressions/test_expressions.py | 4 +-
tests/expressions/test_visitors.py | 48 +++++-----
tests/integration/test_add_files.py | 33 ++++++-
tests/integration/test_deletes.py | 20 ++--
tests/integration/test_writes/test_writes.py | 131 ++++++++++++++++++++++++++-
tests/io/test_pyarrow_stats.py | 26 +++++-
tests/io/test_pyarrow_visitor.py | 2 +-
tests/table/test_snapshots.py | 20 ++--
tests/table/test_upsert.py | 43 +++++++++
18 files changed, 395 insertions(+), 104 deletions(-)
(iceberg-python) 11/11: Fix `add_files` with non-identity transforms (#1925)
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.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit 562de407d0dbf5e1212f0a541001a12543a645f3
Author: Fokko Driesprong <fo...@apache.org>
AuthorDate: Wed Apr 16 20:35:36 2025 +0200
Fix `add_files` with non-identity transforms (#1925)
<!--
Thanks for opening a pull request!
-->
<!-- In the case this PR will resolve an issue, please replace
${GITHUB_ISSUE_ID} below with the actual Github issue id. -->
<!-- Closes #${GITHUB_ISSUE_ID} -->
Found out I broke this myself after doing a `git bisect`:
```
36d383dcb676ae5ef59c34cc2910d16a8e30a80c is the first bad commit
commit 36d383dcb676ae5ef59c34cc2910d16a8e30a80c
Author: Fokko Driesprong <fo...@apache.org>
Date: Thu Jan 23 07:50:54 2025 +0100
PyArrow: Avoid buffer-overflow by avoid doing a sort (#1555)
Second attempt of https://github.com/apache/iceberg-python/pull/1539
This was already being discussed back here:
https://github.com/apache/iceberg-python/issues/208#issuecomment-1889891973
This PR changes from doing a sort, and then a single pass over the table
to the approach where we determine the unique partition tuples filter on
them individually.
Fixes https://github.com/apache/iceberg-python/issues/1491
Because the sort caused buffers to be joined where it would overflow in
Arrow. I think this is an issue on the Arrow side, and it should
automatically break up into smaller buffers. The `combine_chunks` method
does this correctly.
Now:
```
0.42877754200890195
Run 1 took: 0.2507691659993725
Run 2 took: 0.24833179199777078
Run 3 took: 0.24401691700040828
Run 4 took: 0.2419595829996979
Average runtime of 0.28 seconds
```
Before:
```
Run 0 took: 1.0768639159941813
Run 1 took: 0.8784021250030492
Run 2 took: 0.8486490420036716
Run 3 took: 0.8614017910003895
Run 4 took: 0.8497851670108503
Average runtime of 0.9 seconds
```
So it comes with a nice speedup as well :)
---------
Co-authored-by: Kevin Liu <ke...@users.noreply.github.com>
pyiceberg/io/pyarrow.py | 129 ++-
pyiceberg/partitioning.py | 39 +-
pyiceberg/table/__init__.py | 6 +-
pyproject.toml | 1 +
tests/benchmark/test_benchmark.py | 72 ++
tests/integration/test_partitioning_key.py | 1299 ++++++++++++++--------------
tests/table/test_locations.py | 2 +-
7 files changed, 805 insertions(+), 743 deletions(-)
create mode 100644 tests/benchmark/test_benchmark.py
```
Closes #1917
<!-- In the case of user-facing changes, please add the changelog label.
-->
---
pyiceberg/io/pyarrow.py | 31 +++++++++++++++++++------------
tests/integration/test_add_files.py | 33 +++++++++++++++++++++++++++++++--
2 files changed, 50 insertions(+), 14 deletions(-)
diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py
index 90309546..cf7bcc10 100644
--- a/pyiceberg/io/pyarrow.py
+++ b/pyiceberg/io/pyarrow.py
@@ -2205,29 +2205,36 @@ class DataFileStatistics:
if partition_field.source_id not in self.column_aggregates:
return None
- if not partition_field.transform.preserves_order:
+ source_field = schema.find_field(partition_field.source_id)
+ iceberg_transform = partition_field.transform
+
+ if not iceberg_transform.preserves_order:
raise ValueError(
f"Cannot infer partition value from parquet metadata for a non-linear Partition Field: {partition_field.name} with transform {partition_field.transform}"
)
- lower_value = partition_record_value(
- partition_field=partition_field,
- value=self.column_aggregates[partition_field.source_id].current_min,
- schema=schema,
+ transform_func = iceberg_transform.transform(source_field.field_type)
+
+ lower_value = transform_func(
+ partition_record_value(
+ partition_field=partition_field,
+ value=self.column_aggregates[partition_field.source_id].current_min,
+ schema=schema,
+ )
)
- upper_value = partition_record_value(
- partition_field=partition_field,
- value=self.column_aggregates[partition_field.source_id].current_max,
- schema=schema,
+ upper_value = transform_func(
+ partition_record_value(
+ partition_field=partition_field,
+ value=self.column_aggregates[partition_field.source_id].current_max,
+ schema=schema,
+ )
)
if lower_value != upper_value:
raise ValueError(
f"Cannot infer partition value from parquet metadata as there are more than one partition values for Partition Field: {partition_field.name}. {lower_value=}, {upper_value=}"
)
- source_field = schema.find_field(partition_field.source_id)
- transform = partition_field.transform.transform(source_field.field_type)
- return transform(lower_value)
+ return lower_value
def partition(self, partition_spec: PartitionSpec, schema: Schema) -> Record:
return Record(**{field.name: self._partition_value(field, schema) for field in partition_spec.fields})
diff --git a/tests/integration/test_add_files.py b/tests/integration/test_add_files.py
index 87136152..a1480375 100644
--- a/tests/integration/test_add_files.py
+++ b/tests/integration/test_add_files.py
@@ -30,11 +30,12 @@ from pytest_mock.plugin import MockerFixture
from pyiceberg.catalog import Catalog
from pyiceberg.exceptions import NoSuchTableError
from pyiceberg.io import FileIO
-from pyiceberg.io.pyarrow import UnsupportedPyArrowTypeException, _pyarrow_schema_ensure_large_types
+from pyiceberg.io.pyarrow import UnsupportedPyArrowTypeException, schema_to_pyarrow
+from pyiceberg.io.pyarrow import _pyarrow_schema_ensure_large_types
from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionField, PartitionSpec
from pyiceberg.schema import Schema
from pyiceberg.table import Table
-from pyiceberg.transforms import BucketTransform, IdentityTransform, MonthTransform
+from pyiceberg.transforms import BucketTransform, HourTransform, IdentityTransform, MonthTransform
from pyiceberg.types import (
BooleanType,
DateType,
@@ -42,6 +43,7 @@ from pyiceberg.types import (
LongType,
NestedField,
StringType,
+ TimestampType,
TimestamptzType,
)
@@ -850,3 +852,30 @@ def test_add_files_that_referenced_by_current_snapshot_with_check_duplicate_file
with pytest.raises(ValueError) as exc_info:
tbl.add_files(file_paths=[existing_files_in_table], check_duplicate_files=True)
assert f"Cannot add files that are already referenced by table, files: {existing_files_in_table}" in str(exc_info.value)
+
+
+@pytest.mark.integration
+def test_add_files_hour_transform(session_catalog: Catalog) -> None:
+ identifier = "default.test_add_files_hour_transform"
+
+ schema = Schema(NestedField(1, "hourly", TimestampType()))
+ schema_arrow = schema_to_pyarrow(schema, include_field_ids=False)
+ spec = PartitionSpec(PartitionField(source_id=1, field_id=1000, transform=HourTransform(), name="spec_hour"))
+
+ tbl = _create_table(session_catalog, identifier, format_version=1, schema=schema, partition_spec=spec)
+
+ file_path = "s3://warehouse/default/test_add_files_hour_transform/test.parquet"
+
+ from pyiceberg.utils.datetime import micros_to_timestamp
+
+ arrow_table = pa.Table.from_pylist(
+ [{"hourly": micros_to_timestamp(1743465600155254)}, {"hourly": micros_to_timestamp(1743469198047855)}],
+ schema=schema_arrow,
+ )
+
+ fo = tbl.io.new_output(file_path)
+ with fo.create(overwrite=True) as fos:
+ with pq.ParquetWriter(fos, schema=schema_arrow) as writer:
+ writer.write_table(arrow_table)
+
+ tbl.add_files(file_paths=[file_path])
(iceberg-python) 09/11: Temporary fix for filtering on empty batches (#1901)
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.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit e5d1bb87ba0a358cc2abdf430f48fd033549564e
Author: koenvo <in...@koenvossen.nl>
AuthorDate: Thu Apr 10 00:35:34 2025 +0200
Temporary fix for filtering on empty batches (#1901)
Potential fix for https://github.com/apache/iceberg-python/issues/1804
Might want to write a test, but not sure yet how to reproduce without
using glue.
Closes https://github.com/apache/iceberg-python/issues/1804
---
pyiceberg/io/pyarrow.py | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py
index 839b16ad..90309546 100644
--- a/pyiceberg/io/pyarrow.py
+++ b/pyiceberg/io/pyarrow.py
@@ -1421,11 +1421,15 @@ def _task_to_record_batches(
# Apply the user filter
if pyarrow_filter is not None:
- current_batch = current_batch.filter(pyarrow_filter)
+ # Temporary fix until PyArrow 21 is released ( https://github.com/apache/arrow/pull/46057 )
+ table = pa.Table.from_batches([current_batch])
+ table = table.filter(pyarrow_filter)
# skip empty batches
- if current_batch.num_rows == 0:
+ if table.num_rows == 0:
continue
+ current_batch = table.combine_chunks().to_batches()[0]
+
result_batch = _to_requested_schema(
projected_schema,
file_project_schema,
(iceberg-python) 04/11: Fix decimal physicial type mapping (#1839)
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.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit 903fb917310f4a113809e50fcb8a0b696440d7d9
Author: Amit Kumar Singh <am...@gmail.com>
AuthorDate: Mon Mar 31 19:07:58 2025 +0530
Fix decimal physicial type mapping (#1839)
This pull request addresses the handling of decimal physical type
matching in Parquet. It implements rules such that:
For precision ≤ 9, values are stored as `int32`.
For precision ≤ 18, values are stored as `int64`.
For higher precision, values are stored as a `FIXED_LEN_BYTE_ARRAY`.
Closes #1789
---------
Co-authored-by: redpheonixx <am...@192.168.1.5>
Co-authored-by: redpheonixx <am...@192.168.1.10>
---
pyiceberg/io/pyarrow.py | 12 +++++++++---
tests/io/test_pyarrow_stats.py | 26 +++++++++++++++++++++-----
2 files changed, 30 insertions(+), 8 deletions(-)
diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py
index f7e3c7c0..25f81ed9 100644
--- a/pyiceberg/io/pyarrow.py
+++ b/pyiceberg/io/pyarrow.py
@@ -171,6 +171,7 @@ from pyiceberg.types import (
from pyiceberg.utils.concurrent import ExecutorFactory
from pyiceberg.utils.config import Config
from pyiceberg.utils.datetime import millis_to_datetime
+from pyiceberg.utils.decimal import unscaled_to_decimal
from pyiceberg.utils.properties import get_first_property_value, property_as_bool, property_as_int
from pyiceberg.utils.singleton import Singleton
from pyiceberg.utils.truncate import truncate_upper_bound_binary_string, truncate_upper_bound_text_string
@@ -1861,7 +1862,7 @@ class PrimitiveToPhysicalType(SchemaVisitorPerPrimitiveType[str]):
return "FIXED_LEN_BYTE_ARRAY"
def visit_decimal(self, decimal_type: DecimalType) -> str:
- return "FIXED_LEN_BYTE_ARRAY"
+ return "INT32" if decimal_type.precision <= 9 else "INT64" if decimal_type.precision <= 18 else "FIXED_LEN_BYTE_ARRAY"
def visit_boolean(self, boolean_type: BooleanType) -> str:
return "BOOLEAN"
@@ -2326,8 +2327,13 @@ def data_file_statistics_from_parquet_metadata(
stats_col.iceberg_type, statistics.physical_type, stats_col.mode.length
)
- col_aggs[field_id].update_min(statistics.min)
- col_aggs[field_id].update_max(statistics.max)
+ if isinstance(stats_col.iceberg_type, DecimalType) and statistics.physical_type != "FIXED_LEN_BYTE_ARRAY":
+ scale = stats_col.iceberg_type.scale
+ col_aggs[field_id].update_min(unscaled_to_decimal(statistics.min_raw, scale))
+ col_aggs[field_id].update_max(unscaled_to_decimal(statistics.max_raw, scale))
+ else:
+ col_aggs[field_id].update_min(statistics.min)
+ col_aggs[field_id].update_max(statistics.max)
except pyarrow.lib.ArrowNotImplementedError as e:
invalidate_col.add(field_id)
diff --git a/tests/io/test_pyarrow_stats.py b/tests/io/test_pyarrow_stats.py
index 78889171..09c9eb4a 100644
--- a/tests/io/test_pyarrow_stats.py
+++ b/tests/io/test_pyarrow_stats.py
@@ -27,6 +27,7 @@ from datetime import (
timedelta,
timezone,
)
+from decimal import Decimal
from typing import (
Any,
Dict,
@@ -446,6 +447,9 @@ def construct_test_table_primitive_types() -> Tuple[pq.FileMetaData, Union[Table
{"id": 10, "name": "strings", "required": False, "type": "string"},
{"id": 11, "name": "uuids", "required": False, "type": "uuid"},
{"id": 12, "name": "binaries", "required": False, "type": "binary"},
+ {"id": 13, "name": "decimal8", "required": False, "type": "decimal(5, 2)"},
+ {"id": 14, "name": "decimal16", "required": False, "type": "decimal(16, 6)"},
+ {"id": 15, "name": "decimal32", "required": False, "type": "decimal(19, 6)"},
],
},
],
@@ -470,6 +474,9 @@ def construct_test_table_primitive_types() -> Tuple[pq.FileMetaData, Union[Table
strings = ["hello", "world"]
uuids = [uuid.uuid3(uuid.NAMESPACE_DNS, "foo").bytes, uuid.uuid3(uuid.NAMESPACE_DNS, "bar").bytes]
binaries = [b"hello", b"world"]
+ decimal8 = pa.array([Decimal("123.45"), Decimal("678.91")], pa.decimal128(8, 2))
+ decimal16 = pa.array([Decimal("12345679.123456"), Decimal("67891234.678912")], pa.decimal128(16, 6))
+ decimal32 = pa.array([Decimal("1234567890123.123456"), Decimal("9876543210703.654321")], pa.decimal128(19, 6))
table = pa.Table.from_pydict(
{
@@ -485,6 +492,9 @@ def construct_test_table_primitive_types() -> Tuple[pq.FileMetaData, Union[Table
"strings": strings,
"uuids": uuids,
"binaries": binaries,
+ "decimal8": decimal8,
+ "decimal16": decimal16,
+ "decimal32": decimal32,
},
schema=arrow_schema,
)
@@ -492,7 +502,7 @@ def construct_test_table_primitive_types() -> Tuple[pq.FileMetaData, Union[Table
metadata_collector: List[Any] = []
with pa.BufferOutputStream() as f:
- with pq.ParquetWriter(f, table.schema, metadata_collector=metadata_collector) as writer:
+ with pq.ParquetWriter(f, table.schema, metadata_collector=metadata_collector, store_decimal_as_integer=True) as writer:
writer.write_table(table)
return metadata_collector[0], table_metadata
@@ -510,13 +520,13 @@ def test_metrics_primitive_types() -> None:
)
datafile = DataFile(**statistics.to_serialized_dict())
- assert len(datafile.value_counts) == 12
- assert len(datafile.null_value_counts) == 12
+ assert len(datafile.value_counts) == 15
+ assert len(datafile.null_value_counts) == 15
assert len(datafile.nan_value_counts) == 0
tz = timezone(timedelta(seconds=19800))
- assert len(datafile.lower_bounds) == 12
+ assert len(datafile.lower_bounds) == 15
assert datafile.lower_bounds[1] == STRUCT_BOOL.pack(False)
assert datafile.lower_bounds[2] == STRUCT_INT32.pack(23)
assert datafile.lower_bounds[3] == STRUCT_INT64.pack(2)
@@ -529,8 +539,11 @@ def test_metrics_primitive_types() -> None:
assert datafile.lower_bounds[10] == b"he"
assert datafile.lower_bounds[11] == uuid.uuid3(uuid.NAMESPACE_DNS, "foo").bytes
assert datafile.lower_bounds[12] == b"he"
+ assert datafile.lower_bounds[13][::-1].ljust(4, b"\x00") == STRUCT_INT32.pack(12345)
+ assert datafile.lower_bounds[14][::-1].ljust(8, b"\x00") == STRUCT_INT64.pack(12345679123456)
+ assert str(int.from_bytes(datafile.lower_bounds[15], byteorder="big", signed=True)).encode("utf-8") == b"1234567890123123456"
- assert len(datafile.upper_bounds) == 12
+ assert len(datafile.upper_bounds) == 15
assert datafile.upper_bounds[1] == STRUCT_BOOL.pack(True)
assert datafile.upper_bounds[2] == STRUCT_INT32.pack(89)
assert datafile.upper_bounds[3] == STRUCT_INT64.pack(54)
@@ -543,6 +556,9 @@ def test_metrics_primitive_types() -> None:
assert datafile.upper_bounds[10] == b"wp"
assert datafile.upper_bounds[11] == uuid.uuid3(uuid.NAMESPACE_DNS, "bar").bytes
assert datafile.upper_bounds[12] == b"wp"
+ assert datafile.upper_bounds[13][::-1].ljust(4, b"\x00") == STRUCT_INT32.pack(67891)
+ assert datafile.upper_bounds[14][::-1].ljust(8, b"\x00") == STRUCT_INT64.pack(67891234678912)
+ assert str(int.from_bytes(datafile.upper_bounds[15], byteorder="big", signed=True)).encode("utf-8") == b"9876543210703654321"
def construct_test_table_invalid_upper_bound() -> Tuple[pq.FileMetaData, Union[TableMetadataV1, TableMetadataV2]]:
(iceberg-python) 07/11: Fix the snapshot summary of a partial overwrite (#1879)
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.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit bc7e1cc7054e4fde4b156db5e1c8c5c7626fe0c6
Author: Fokko Driesprong <fo...@apache.org>
AuthorDate: Wed Apr 16 21:55:10 2025 +0200
Fix the snapshot summary of a partial overwrite (#1879)
# Rationale for this change
@kevinjqliu PTAL. I took the liberty of providing a fix for this since I
was curious where this was coming from, hope you don't mind! I've
cherry-picked your commit with the test.
Java produces:
```json
{
"added-data-files": "1",
"added-files-size": "707",
"added-records": "2",
"app-id": "local-1743678304626",
"changed-partition-count": "1",
"deleted-data-files": "1",
"deleted-records": "3",
"engine-name": "spark",
"engine-version": "3.5.5",
"iceberg-version": "Apache Iceberg 1.8.1 (commit 9ce0fcf0af7becf25ad9fc996c3bad2afdcfd33d)",
"removed-files-size": "693",
"spark.app.id": "local-1743678304626",
"total-data-files": "3",
"total-delete-files": "0",
"total-equality-deletes": "0",
"total-files-size": "1993",
"total-position-deletes": "0",
"total-records": "4"
}
```
# 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 <ke...@gmail.com>
Co-authored-by: Kevin Liu <ke...@users.noreply.github.com>
---
pyiceberg/table/snapshots.py | 6 ++
pyiceberg/table/update/snapshot.py | 1 -
tests/integration/test_deletes.py | 20 +++---
tests/integration/test_writes/test_writes.py | 95 +++++++++++++++++++++++++++-
tests/table/test_snapshots.py | 20 ++----
5 files changed, 115 insertions(+), 27 deletions(-)
diff --git a/pyiceberg/table/snapshots.py b/pyiceberg/table/snapshots.py
index a5515f12..af3f0404 100644
--- a/pyiceberg/table/snapshots.py
+++ b/pyiceberg/table/snapshots.py
@@ -28,6 +28,7 @@ from pyiceberg.io import FileIO
from pyiceberg.manifest import DataFile, DataFileContent, ManifestFile, _manifests
from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec
from pyiceberg.schema import Schema
+from pyiceberg.utils.deprecated import deprecation_message
if TYPE_CHECKING:
from pyiceberg.table.metadata import TableMetadata
@@ -356,6 +357,11 @@ def update_snapshot_summaries(
raise ValueError(f"Operation not implemented: {summary.operation}")
if truncate_full_table and summary.operation == Operation.OVERWRITE and previous_summary is not None:
+ deprecation_message(
+ deprecated_in="0.10.0",
+ removed_in="0.11.0",
+ help_message="The truncate-full-table shouldn't be used.",
+ )
summary = _truncate_table_summary(summary, previous_summary)
if not previous_summary:
diff --git a/pyiceberg/table/update/snapshot.py b/pyiceberg/table/update/snapshot.py
index f21c5017..0a09726d 100644
--- a/pyiceberg/table/update/snapshot.py
+++ b/pyiceberg/table/update/snapshot.py
@@ -236,7 +236,6 @@ class _SnapshotProducer(UpdateTableMetadata[U], Generic[U]):
return update_snapshot_summaries(
summary=Summary(operation=self._operation, **ssc.build(), **snapshot_properties),
previous_summary=previous_snapshot.summary if previous_snapshot is not None else None,
- truncate_full_table=self._operation == Operation.OVERWRITE,
)
def _commit(self) -> UpdatesAndRequirements:
diff --git a/tests/integration/test_deletes.py b/tests/integration/test_deletes.py
index ae03beea..527f6596 100644
--- a/tests/integration/test_deletes.py
+++ b/tests/integration/test_deletes.py
@@ -467,21 +467,19 @@ def test_partitioned_table_positional_deletes_sequence_number(spark: SparkSessio
assert snapshots[2].summary == Summary(
Operation.OVERWRITE,
**{
- "added-files-size": snapshots[2].summary["total-files-size"],
"added-data-files": "1",
+ "added-files-size": snapshots[2].summary["added-files-size"],
"added-records": "2",
"changed-partition-count": "1",
- "total-files-size": snapshots[2].summary["total-files-size"],
- "total-delete-files": "0",
- "total-data-files": "1",
- "total-position-deletes": "0",
- "total-records": "2",
- "total-equality-deletes": "0",
- "deleted-data-files": "2",
- "removed-delete-files": "1",
- "deleted-records": "5",
+ "deleted-data-files": "1",
+ "deleted-records": "3",
"removed-files-size": snapshots[2].summary["removed-files-size"],
- "removed-position-deletes": "1",
+ "total-data-files": "2",
+ "total-delete-files": "1",
+ "total-equality-deletes": "0",
+ "total-files-size": snapshots[2].summary["total-files-size"],
+ "total-position-deletes": "1",
+ "total-records": "4",
},
)
diff --git a/tests/integration/test_writes/test_writes.py b/tests/integration/test_writes/test_writes.py
index f64e225b..dcfd0156 100644
--- a/tests/integration/test_writes/test_writes.py
+++ b/tests/integration/test_writes/test_writes.py
@@ -249,7 +249,7 @@ def test_summaries(spark: SparkSession, session_catalog: Catalog, arrow_table_wi
"total-records": "0",
}
- # Overwrite
+ # Append
assert summaries[3] == {
"added-data-files": "1",
"added-files-size": str(file_size),
@@ -263,6 +263,99 @@ def test_summaries(spark: SparkSession, session_catalog: Catalog, arrow_table_wi
}
+@pytest.mark.integration
+def test_summaries_partial_overwrite(spark: SparkSession, session_catalog: Catalog) -> None:
+ identifier = "default.test_summaries_partial_overwrite"
+ TEST_DATA = {
+ "id": [1, 2, 3, 1, 1],
+ "name": ["AB", "CD", "EF", "CD", "EF"],
+ }
+ pa_schema = pa.schema(
+ [
+ pa.field("id", pa.int32()),
+ pa.field("name", pa.string()),
+ ]
+ )
+ arrow_table = pa.Table.from_pydict(TEST_DATA, schema=pa_schema)
+ tbl = _create_table(session_catalog, identifier, {"format-version": "2"}, schema=pa_schema)
+ with tbl.update_spec() as txn:
+ txn.add_identity("id")
+ tbl.append(arrow_table)
+
+ assert len(tbl.inspect.data_files()) == 3
+
+ tbl.delete(delete_filter="id == 1 and name = 'AB'") # partial overwrite data from 1 data file
+
+ rows = spark.sql(
+ f"""
+ SELECT operation, summary
+ FROM {identifier}.snapshots
+ ORDER BY committed_at ASC
+ """
+ ).collect()
+
+ operations = [row.operation for row in rows]
+ assert operations == ["append", "overwrite"]
+
+ summaries = [row.summary for row in rows]
+
+ file_size = int(summaries[0]["added-files-size"])
+ assert file_size > 0
+
+ # APPEND
+ assert summaries[0] == {
+ "added-data-files": "3",
+ "added-files-size": "2570",
+ "added-records": "5",
+ "changed-partition-count": "3",
+ "total-data-files": "3",
+ "total-delete-files": "0",
+ "total-equality-deletes": "0",
+ "total-files-size": "2570",
+ "total-position-deletes": "0",
+ "total-records": "5",
+ }
+ # Java produces:
+ # {
+ # "added-data-files": "1",
+ # "added-files-size": "707",
+ # "added-records": "2",
+ # "app-id": "local-1743678304626",
+ # "changed-partition-count": "1",
+ # "deleted-data-files": "1",
+ # "deleted-records": "3",
+ # "engine-name": "spark",
+ # "engine-version": "3.5.5",
+ # "iceberg-version": "Apache Iceberg 1.8.1 (commit 9ce0fcf0af7becf25ad9fc996c3bad2afdcfd33d)",
+ # "removed-files-size": "693",
+ # "spark.app.id": "local-1743678304626",
+ # "total-data-files": "3",
+ # "total-delete-files": "0",
+ # "total-equality-deletes": "0",
+ # "total-files-size": "1993",
+ # "total-position-deletes": "0",
+ # "total-records": "4"
+ # }
+ files = tbl.inspect.data_files()
+ assert len(files) == 3
+ assert summaries[1] == {
+ "added-data-files": "1",
+ "added-files-size": "859",
+ "added-records": "2",
+ "changed-partition-count": "1",
+ "deleted-data-files": "1",
+ "deleted-records": "3",
+ "removed-files-size": "866",
+ "total-data-files": "3",
+ "total-delete-files": "0",
+ "total-equality-deletes": "0",
+ "total-files-size": "2563",
+ "total-position-deletes": "0",
+ "total-records": "4",
+ }
+ assert len(tbl.scan().to_pandas()) == 4
+
+
@pytest.mark.integration
def test_data_files(spark: SparkSession, session_catalog: Catalog, arrow_table_with_null: pa.Table) -> None:
identifier = "default.arrow_data_files"
diff --git a/tests/table/test_snapshots.py b/tests/table/test_snapshots.py
index b4dde217..f9f7fb02 100644
--- a/tests/table/test_snapshots.py
+++ b/tests/table/test_snapshots.py
@@ -289,7 +289,6 @@ def test_merge_snapshot_summaries_overwrite_summary() -> None:
"total-position-deletes": "1",
"total-records": "1",
},
- truncate_full_table=True,
)
expected = {
@@ -299,18 +298,12 @@ def test_merge_snapshot_summaries_overwrite_summary() -> None:
"added-files-size": "4",
"added-position-deletes": "5",
"added-records": "6",
- "total-data-files": "1",
- "total-records": "6",
- "total-delete-files": "2",
- "total-equality-deletes": "3",
- "total-files-size": "4",
- "total-position-deletes": "5",
- "deleted-data-files": "1",
- "removed-delete-files": "1",
- "deleted-records": "1",
- "removed-files-size": "1",
- "removed-position-deletes": "1",
- "removed-equality-deletes": "1",
+ "total-data-files": "2",
+ "total-delete-files": "3",
+ "total-records": "7",
+ "total-files-size": "5",
+ "total-position-deletes": "6",
+ "total-equality-deletes": "4",
}
assert actual.additional_properties == expected
@@ -337,7 +330,6 @@ def test_invalid_type() -> None:
},
),
previous_summary={"total-data-files": "abc"}, # should be a number
- truncate_full_table=True,
)
assert "Could not parse summary property total-data-files to an int: abc" in str(e.value)
(iceberg-python) 06/11: Set field-id when needed (#1867)
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.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit 29e94e75f85735eee13c04db7345afa21bd04694
Author: Fokko Driesprong <fo...@apache.org>
AuthorDate: Tue Apr 1 17:24:57 2025 +0200
Set field-id when needed (#1867)
Fixes #1798
<!--
Thanks for opening a pull request!
-->
<!-- In the case this PR will resolve an issue, please replace
${GITHUB_ISSUE_ID} below with the actual Github issue id. -->
<!-- Closes #${GITHUB_ISSUE_ID} -->
# Rationale for this change
# Are these changes tested?
# Are there any user-facing changes?
<!-- In the case of user-facing changes, please add the changelog label.
-->
---
pyiceberg/io/pyarrow.py | 2 +-
tests/integration/test_writes/test_writes.py | 36 ++++++++++++++++++++++++++++
2 files changed, 37 insertions(+), 1 deletion(-)
diff --git a/pyiceberg/io/pyarrow.py b/pyiceberg/io/pyarrow.py
index 25f81ed9..839b16ad 100644
--- a/pyiceberg/io/pyarrow.py
+++ b/pyiceberg/io/pyarrow.py
@@ -1750,7 +1750,7 @@ class ArrowProjectionVisitor(SchemaWithPartnerVisitor[pa.Array, Optional[pa.Arra
field_arrays.append(array)
fields.append(self._construct_field(field, array.type))
elif field.optional:
- arrow_type = schema_to_pyarrow(field.field_type, include_field_ids=False)
+ arrow_type = schema_to_pyarrow(field.field_type, include_field_ids=self._include_field_ids)
field_arrays.append(pa.nulls(len(struct_array), type=arrow_type))
fields.append(self._construct_field(field, arrow_type))
else:
diff --git a/tests/integration/test_writes/test_writes.py b/tests/integration/test_writes/test_writes.py
index 1fe29c68..f64e225b 100644
--- a/tests/integration/test_writes/test_writes.py
+++ b/tests/integration/test_writes/test_writes.py
@@ -51,6 +51,7 @@ from pyiceberg.types import (
DateType,
DoubleType,
IntegerType,
+ ListType,
LongType,
NestedField,
StringType,
@@ -1639,3 +1640,38 @@ def test_abort_table_transaction_on_exception(
# Validate the transaction is aborted and no partial update is applied
assert len(tbl.scan().to_pandas()) == table_size # type: ignore
+
+
+@pytest.mark.integration
+def test_write_optional_list(session_catalog: Catalog) -> None:
+ identifier = "default.test_write_optional_list"
+ schema = Schema(
+ NestedField(field_id=1, name="name", field_type=StringType(), required=False),
+ NestedField(
+ field_id=3,
+ name="my_list",
+ field_type=ListType(element_id=45, element=StringType(), element_required=False),
+ required=False,
+ ),
+ )
+ session_catalog.create_table_if_not_exists(identifier, schema)
+
+ df_1 = pa.Table.from_pylist(
+ [
+ {"name": "one", "my_list": ["test"]},
+ {"name": "another", "my_list": ["test"]},
+ ]
+ )
+ session_catalog.load_table(identifier).append(df_1)
+
+ assert len(session_catalog.load_table(identifier).scan().to_arrow()) == 2
+
+ df_2 = pa.Table.from_pylist(
+ [
+ {"name": "one"},
+ {"name": "another"},
+ ]
+ )
+ session_catalog.load_table(identifier).append(df_2)
+
+ assert len(session_catalog.load_table(identifier).scan().to_arrow()) == 4
(iceberg-python) 01/11: Remove unused argument (#1734)
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.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit 998ce7cc33c2f50fbb160d62ab1f386a8cd7812e
Author: Fokko Driesprong <fo...@apache.org>
AuthorDate: Thu Feb 27 17:30:47 2025 +0100
Remove unused argument (#1734)
Less is more!
---
pyiceberg/table/__init__.py | 2 --
1 file changed, 2 deletions(-)
diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py
index 679f74d1..ee50a8b5 100644
--- a/pyiceberg/table/__init__.py
+++ b/pyiceberg/table/__init__.py
@@ -1549,7 +1549,6 @@ def _open_manifest(
io: FileIO,
manifest: ManifestFile,
partition_filter: Callable[[DataFile], bool],
- residual_evaluator: Callable[[Record], BooleanExpression],
metrics_evaluator: Callable[[DataFile], bool],
) -> List[ManifestEntry]:
"""Open a manifest file and return matching manifest entries.
@@ -1716,7 +1715,6 @@ class DataScan(TableScan):
self.io,
manifest,
partition_evaluators[manifest.partition_spec_id],
- residual_evaluators[manifest.partition_spec_id],
self._build_metrics_evaluator(),
)
for manifest in manifests
(iceberg-python) 08/11: Fix creation of Bucket Transforms with `pydantic>=2.11.0` (#1881)
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.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit e9745d5f63f4841ba725ca76343a8e3ac5574e6b
Author: b-rick <92...@users.noreply.github.com>
AuthorDate: Fri Apr 4 17:44:28 2025 +1100
Fix creation of Bucket Transforms with `pydantic>=2.11.0` (#1881)
## Rationale For This Change
When using pydantic>=2.11.0, we get an error when creating bucket
transforms. In this version, it's illegal to access self before calling
super. To fix this, we just need to ensure we call `super().__init__`
before setting field properties on `self`.
---
## Before
Using `pydantic==2.11.0` the test
`test_transforms.py::test_bucket_hash_values` fails
```python
tests/test_transforms.py:None (tests/test_transforms.py)
test_transforms.py:179: in <module>
(BucketTransform(2).transform(IntegerType()), 0, 0),
../pyiceberg/transforms.py:237: in __init__
self._num_buckets = num_buckets
../../../miniforge3/envs/pyiceberg/lib/python3.12/site-packages/pydantic/main.py:991: in __setattr__
setattr_handler(self, name, value) # call here to not memo on possibly unknown fields
../../../miniforge3/envs/pyiceberg/lib/python3.12/site-packages/pydantic/main.py:105: in <lambda>
'private': lambda model, name, val: model.__pydantic_private__.__setitem__(name, val), # pyright: ignore[reportOptionalMemberAccess]
E AttributeError: 'NoneType' object has no attribute '__setitem__'. Did you mean: '__setattr__'?
```
---
## After
Using `pydantic==2.11.0` the test
`test_transforms.py::test_bucket_hash_values` succeeds.
---
✅ Are these changes tested?
Yes
- No test was added to reproduce the bug because it is already
reproducible with the existing test
Co-authored-by: Zac Sanchez <za...@imc.com>
---
pyiceberg/transforms.py | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pyiceberg/transforms.py b/pyiceberg/transforms.py
index b8f0b975..7dfe7514 100644
--- a/pyiceberg/transforms.py
+++ b/pyiceberg/transforms.py
@@ -232,8 +232,8 @@ class BucketTransform(Transform[S, int]):
_num_buckets: PositiveInt = PrivateAttr()
def __init__(self, num_buckets: int, **data: Any) -> None:
- self._num_buckets = num_buckets
super().__init__(f"bucket[{num_buckets}]", **data)
+ self._num_buckets = num_buckets
@property
def num_buckets(self) -> int:
(iceberg-python) 02/11: Fix thrift client connection for Kerberos Hive Client (#1747)
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.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit b989197f94093fcbd0bbe8d58928dda9d1888b27
Author: Kevin Liu <ke...@users.noreply.github.com>
AuthorDate: Wed Apr 16 14:37:23 2025 -0400
Fix thrift client connection for Kerberos Hive Client (#1747)
Closes #1744
`TSaslClientTransport` cannot be reopen. This PR changes the behavior to
recreate a `TSaslClientTransport` when its already closed.
Note, `_HiveClient` should be used with context manager, but can be used
without.
---
pyiceberg/catalog/hive.py | 36 ++++++++++++++++++++++--------------
1 file changed, 22 insertions(+), 14 deletions(-)
diff --git a/pyiceberg/catalog/hive.py b/pyiceberg/catalog/hive.py
index 7189b12c..cad89e10 100644
--- a/pyiceberg/catalog/hive.py
+++ b/pyiceberg/catalog/hive.py
@@ -18,6 +18,7 @@ import getpass
import logging
import socket
import time
+from functools import cached_property
from types import TracebackType
from typing import (
TYPE_CHECKING,
@@ -142,40 +143,47 @@ class _HiveClient:
"""Helper class to nicely open and close the transport."""
_transport: TTransport
- _client: Client
_ugi: Optional[List[str]]
def __init__(self, uri: str, ugi: Optional[str] = None, kerberos_auth: Optional[bool] = HIVE_KERBEROS_AUTH_DEFAULT):
self._uri = uri
self._kerberos_auth = kerberos_auth
self._ugi = ugi.split(":") if ugi else None
+ self._transport = self._init_thrift_transport()
- self._init_thrift_client()
-
- def _init_thrift_client(self) -> None:
+ def _init_thrift_transport(self) -> TTransport:
url_parts = urlparse(self._uri)
-
socket = TSocket.TSocket(url_parts.hostname, url_parts.port)
-
if not self._kerberos_auth:
- self._transport = TTransport.TBufferedTransport(socket)
+ return TTransport.TBufferedTransport(socket)
else:
- self._transport = TTransport.TSaslClientTransport(socket, host=url_parts.hostname, service="hive")
+ return TTransport.TSaslClientTransport(socket, host=url_parts.hostname, service="hive")
+ @cached_property
+ def _client(self) -> Client:
protocol = TBinaryProtocol.TBinaryProtocol(self._transport)
-
- self._client = Client(protocol)
+ client = Client(protocol)
+ if self._ugi:
+ client.set_ugi(*self._ugi)
+ return client
def __enter__(self) -> Client:
- self._transport.open()
- if self._ugi:
- self._client.set_ugi(*self._ugi)
+ """Make sure the transport is initialized and open."""
+ if not self._transport.isOpen():
+ try:
+ self._transport.open()
+ except TTransport.TTransportException:
+ # reinitialize _transport
+ self._transport = self._init_thrift_transport()
+ self._transport.open()
return self._client
def __exit__(
self, exctype: Optional[Type[BaseException]], excinst: Optional[BaseException], exctb: Optional[TracebackType]
) -> None:
- self._transport.close()
+ """Close transport if it was opened."""
+ if self._transport.isOpen():
+ self._transport.close()
def _construct_hive_storage_descriptor(
(iceberg-python) 03/11: Use a balanced tree instead of unbalanced one (#1830)
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.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit 9da7bd2b416917cb2407981b8d8fb735299230ec
Author: koenvo <in...@koenvossen.nl>
AuthorDate: Mon Mar 31 14:24:48 2025 +0200
Use a balanced tree instead of unbalanced one (#1830)
**Use a balanced tree instead of an unbalanced one to prevent recursion
error in `create_match_filter`**
<!-- Closes #1776 -->
## Rationale for this change
In the `create_match_filter` function, the previous implementation used
`functools.reduce(operator.or_, filters)` to combine expressions. This
approach constructed a right-heavy, unbalanced tree, which could lead to
a `RecursionError` when dealing with a large number of expressions
(e.g., over 1,000).
To address this, we've introduced the `_build_balanced_tree` function.
This utility constructs a balanced binary tree of expressions, reducing
the maximum depth to O(log n) and thereby preventing potential recursion
errors. This makes expression construction more stable and scalable,
especially when working with large datasets.
```python
# Past behavior
Or(*[A, B, C, D]) = Or(A, Or(B, Or(C, D))
# New behavior
Or(*[A, B, C, D]) = Or(Or(A, B), Or(C, D))
```
## Are these changes tested?
Yes, existing tests cover the functionality of `Or`. Additional testing
was done with large expression sets (e.g., 10,000 items) to ensure that
balanced tree construction avoids recursion errors.
## Are there any user-facing changes?
No, there are no user-facing changes. This is an internal implementation
improvement that does not affect the public API.
Closes #1759
Closes #1785
<!-- In the case of user-facing changes, please add the changelog label.
-->
---
pyiceberg/expressions/__init__.py | 47 +++++++++++++++++++++++++++++++---
pyiceberg/table/upsert_util.py | 8 +++++-
tests/expressions/test_expressions.py | 4 +--
tests/expressions/test_visitors.py | 48 +++++++++++++++++------------------
tests/io/test_pyarrow_visitor.py | 2 +-
5 files changed, 78 insertions(+), 31 deletions(-)
diff --git a/pyiceberg/expressions/__init__.py b/pyiceberg/expressions/__init__.py
index 8b006a28..2adf898f 100644
--- a/pyiceberg/expressions/__init__.py
+++ b/pyiceberg/expressions/__init__.py
@@ -18,11 +18,13 @@
from __future__ import annotations
from abc import ABC, abstractmethod
-from functools import cached_property, reduce
+from functools import cached_property
from typing import (
Any,
+ Callable,
Generic,
Iterable,
+ Sequence,
Set,
Tuple,
Type,
@@ -79,6 +81,45 @@ class BooleanExpression(ABC):
return Or(self, other)
+def _build_balanced_tree(
+ operator_: Callable[[BooleanExpression, BooleanExpression], BooleanExpression], items: Sequence[BooleanExpression]
+) -> BooleanExpression:
+ """
+ Recursively constructs a balanced binary tree of BooleanExpressions using the provided binary operator.
+
+ This function is a safer and more scalable alternative to:
+ reduce(operator_, items)
+
+ Using `reduce` creates a deeply nested, unbalanced tree (e.g., operator_(a, operator_(b, operator_(c, ...)))),
+ which grows linearly with the number of items. This can lead to RecursionError exceptions in Python
+ when the number of expressions is large (e.g., >1000).
+
+ In contrast, this function builds a balanced binary tree with logarithmic depth (O(log n)),
+ helping avoid recursion issues and ensuring that expression trees remain stable, predictable,
+ and safe to traverse — especially in tools like PyIceberg that operate on large logical trees.
+
+ Parameters:
+ operator_ (Callable): A binary operator function (e.g., pyiceberg.expressions.Or, And) that takes two
+ BooleanExpressions and returns a combined BooleanExpression.
+ items (Sequence[BooleanExpression]): A sequence of BooleanExpression objects to combine.
+
+ Returns:
+ BooleanExpression: The balanced combination of all input BooleanExpressions.
+
+ Raises:
+ ValueError: If the input sequence is empty.
+ """
+ if not items:
+ raise ValueError("No expressions to combine")
+ if len(items) == 1:
+ return items[0]
+ mid = len(items) // 2
+
+ left = _build_balanced_tree(operator_, items[:mid])
+ right = _build_balanced_tree(operator_, items[mid:])
+ return operator_(left, right)
+
+
class Term(Generic[L], ABC):
"""A simple expression that evaluates to a value."""
@@ -214,7 +255,7 @@ class And(BooleanExpression):
def __new__(cls, left: BooleanExpression, right: BooleanExpression, *rest: BooleanExpression) -> BooleanExpression: # type: ignore
if rest:
- return reduce(And, (left, right, *rest))
+ return _build_balanced_tree(And, (left, right, *rest))
if left is AlwaysFalse() or right is AlwaysFalse():
return AlwaysFalse()
elif left is AlwaysTrue():
@@ -257,7 +298,7 @@ class Or(BooleanExpression):
def __new__(cls, left: BooleanExpression, right: BooleanExpression, *rest: BooleanExpression) -> BooleanExpression: # type: ignore
if rest:
- return reduce(Or, (left, right, *rest))
+ return _build_balanced_tree(Or, (left, right, *rest))
if left is AlwaysTrue() or right is AlwaysTrue():
return AlwaysTrue()
elif left is AlwaysFalse():
diff --git a/pyiceberg/table/upsert_util.py b/pyiceberg/table/upsert_util.py
index d2bd48bc..e67f6c02 100644
--- a/pyiceberg/table/upsert_util.py
+++ b/pyiceberg/table/upsert_util.py
@@ -26,6 +26,7 @@ from pyiceberg.expressions import (
BooleanExpression,
EqualTo,
In,
+ Or,
)
@@ -39,7 +40,12 @@ def create_match_filter(df: pyarrow_table, join_cols: list[str]) -> BooleanExpre
functools.reduce(operator.and_, [EqualTo(col, row[col]) for col in join_cols]) for row in unique_keys.to_pylist()
]
- return AlwaysFalse() if len(filters) == 0 else functools.reduce(operator.or_, filters)
+ if len(filters) == 0:
+ return AlwaysFalse()
+ elif len(filters) == 1:
+ return filters[0]
+ else:
+ return Or(*filters)
def has_duplicate_rows(df: pyarrow_table, join_cols: list[str]) -> bool:
diff --git a/tests/expressions/test_expressions.py b/tests/expressions/test_expressions.py
index 4926b701..a9b19353 100644
--- a/tests/expressions/test_expressions.py
+++ b/tests/expressions/test_expressions.py
@@ -574,11 +574,11 @@ def test_negate(lhs: BooleanExpression, rhs: BooleanExpression) -> None:
[
(
And(ExpressionA(), ExpressionB(), ExpressionA()),
- And(And(ExpressionA(), ExpressionB()), ExpressionA()),
+ And(ExpressionA(), And(ExpressionB(), ExpressionA())),
),
(
Or(ExpressionA(), ExpressionB(), ExpressionA()),
- Or(Or(ExpressionA(), ExpressionB()), ExpressionA()),
+ Or(ExpressionA(), Or(ExpressionB(), ExpressionA())),
),
(Not(Not(ExpressionA())), ExpressionA()),
],
diff --git a/tests/expressions/test_visitors.py b/tests/expressions/test_visitors.py
index 94bfcf07..586ba9f5 100644
--- a/tests/expressions/test_visitors.py
+++ b/tests/expressions/test_visitors.py
@@ -230,14 +230,14 @@ def test_boolean_expression_visitor() -> None:
"NOT",
"OR",
"EQUALTO",
- "OR",
"NOTEQUALTO",
"OR",
+ "OR",
"EQUALTO",
"NOT",
- "AND",
"NOTEQUALTO",
"AND",
+ "AND",
]
@@ -335,14 +335,14 @@ def test_always_false_or_always_true_expression_binding(table_schema_simple: Sch
),
),
And(
- And(
- BoundIn(
- BoundReference(
- field=NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
- accessor=Accessor(position=0, inner=None),
- ),
- {literal("bar"), literal("baz")},
+ BoundIn(
+ BoundReference(
+ field=NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
+ accessor=Accessor(position=0, inner=None),
),
+ {literal("bar"), literal("baz")},
+ ),
+ And(
BoundEqualTo[int](
BoundReference(
field=NestedField(field_id=2, name="bar", field_type=IntegerType(), required=True),
@@ -350,13 +350,13 @@ def test_always_false_or_always_true_expression_binding(table_schema_simple: Sch
),
literal(1),
),
- ),
- BoundEqualTo(
- BoundReference(
- field=NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
- accessor=Accessor(position=0, inner=None),
+ BoundEqualTo(
+ BoundReference(
+ field=NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
+ accessor=Accessor(position=0, inner=None),
+ ),
+ literal("baz"),
),
- literal("baz"),
),
),
),
@@ -408,28 +408,28 @@ def test_and_expression_binding(
),
),
Or(
+ BoundIn(
+ BoundReference(
+ field=NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
+ accessor=Accessor(position=0, inner=None),
+ ),
+ {literal("bar"), literal("baz")},
+ ),
Or(
BoundIn(
BoundReference(
field=NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
accessor=Accessor(position=0, inner=None),
),
- {literal("bar"), literal("baz")},
+ {literal("bar")},
),
BoundIn(
BoundReference(
field=NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
accessor=Accessor(position=0, inner=None),
),
- {literal("bar")},
- ),
- ),
- BoundIn(
- BoundReference(
- field=NestedField(field_id=1, name="foo", field_type=StringType(), required=False),
- accessor=Accessor(position=0, inner=None),
+ {literal("baz")},
),
- {literal("baz")},
),
),
),
diff --git a/tests/io/test_pyarrow_visitor.py b/tests/io/test_pyarrow_visitor.py
index d13822f5..a6139901 100644
--- a/tests/io/test_pyarrow_visitor.py
+++ b/tests/io/test_pyarrow_visitor.py
@@ -836,5 +836,5 @@ def test_expression_to_complementary_pyarrow(
# Notice an isNan predicate on a str column is automatically converted to always false and removed from Or and thus will not appear in the pc.expr.
assert (
repr(result)
- == """<pyarrow.compute.Expression (((invert((((((string_field == "hello") and (float_field > 100)) or (is_nan(float_field) and (double_field == 0))) or (float_field > 100)) and invert(is_null(double_field, {nan_is_null=false})))) or is_null(float_field, {nan_is_null=false})) or is_null(string_field, {nan_is_null=false})) or is_nan(double_field))>"""
+ == """<pyarrow.compute.Expression (((invert(((((string_field == "hello") and (float_field > 100)) or ((is_nan(float_field) and (double_field == 0)) or (float_field > 100))) and invert(is_null(double_field, {nan_is_null=false})))) or is_null(float_field, {nan_is_null=false})) or is_null(string_field, {nan_is_null=false})) or is_nan(double_field))>"""
)
(iceberg-python) 05/11: fix `upsert` with null values (#1861)
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.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit cfb4f251a790be7350849ddf167f8960f17f3b36
Author: Kevin Liu <ke...@users.noreply.github.com>
AuthorDate: Mon Mar 31 13:32:56 2025 -0400
fix `upsert` with null values (#1861)
<!--
Thanks for opening a pull request!
-->
<!-- In the case this PR will resolve an issue, please replace
${GITHUB_ISSUE_ID} below with the actual Github issue id. -->
<!-- Closes #${GITHUB_ISSUE_ID} -->
# Rationale for this change
Closes #1835
Original implementation, `!=`
([not_equal](https://arrow.apache.org/docs/python/generated/pyarrow.compute.not_equal.html#pyarrow.compute.not_equal))
does not account for `null` values. It emits `null` when either sides
are `null`
The new implementation, first checks for `not_equal`. And on null
values, returns `true` only if both sides are `null`
Similar to https://github.com/apache/iceberg-rust/pull/1045
# Are these changes tested?
Yes
# Are there any user-facing changes?
No
<!-- In the case of user-facing changes, please add the changelog label.
-->
---
pyiceberg/table/upsert_util.py | 11 ++++++++++-
tests/table/test_upsert.py | 43 ++++++++++++++++++++++++++++++++++++++++++
2 files changed, 53 insertions(+), 1 deletion(-)
diff --git a/pyiceberg/table/upsert_util.py b/pyiceberg/table/upsert_util.py
index e67f6c02..c2d554df 100644
--- a/pyiceberg/table/upsert_util.py
+++ b/pyiceberg/table/upsert_util.py
@@ -71,7 +71,16 @@ def get_rows_to_update(source_table: pa.Table, target_table: pa.Table, join_cols
# When the target table is empty, there is nothing to update :)
return source_table.schema.empty_table()
- diff_expr = functools.reduce(operator.or_, [pc.field(f"{col}-lhs") != pc.field(f"{col}-rhs") for col in non_key_cols])
+ diff_expr = functools.reduce(
+ operator.or_,
+ [
+ pc.or_kleene(
+ pc.not_equal(pc.field(f"{col}-lhs"), pc.field(f"{col}-rhs")),
+ pc.is_null(pc.not_equal(pc.field(f"{col}-lhs"), pc.field(f"{col}-rhs"))),
+ )
+ for col in non_key_cols
+ ],
+ )
return (
source_table
diff --git a/tests/table/test_upsert.py b/tests/table/test_upsert.py
index 7f9e13b5..f9952447 100644
--- a/tests/table/test_upsert.py
+++ b/tests/table/test_upsert.py
@@ -496,3 +496,46 @@ def test_upsert_without_identifier_fields(catalog: Catalog) -> None:
ValueError, match="Join columns could not be found, please set identifier-field-ids or pass in explicitly."
):
tbl.upsert(df)
+
+
+def test_upsert_with_nulls(catalog: Catalog) -> None:
+ identifier = "default.test_upsert_with_nulls"
+ _drop_table(catalog, identifier)
+
+ schema = pa.schema(
+ [
+ ("foo", pa.string()),
+ ("bar", pa.int32()),
+ ("baz", pa.bool_()),
+ ]
+ )
+
+ # create table with null value
+ table = catalog.create_table(identifier, schema)
+ data_with_null = pa.Table.from_pylist(
+ [
+ {"foo": "apple", "bar": None, "baz": False},
+ {"foo": "banana", "bar": None, "baz": False},
+ ],
+ schema=schema,
+ )
+ table.append(data_with_null)
+ assert table.scan().to_arrow()["bar"].is_null()
+
+ # upsert table with non-null value
+ data_without_null = pa.Table.from_pylist(
+ [
+ {"foo": "apple", "bar": 7, "baz": False},
+ ],
+ schema=schema,
+ )
+ upd = table.upsert(data_without_null, join_cols=["foo"])
+ assert upd.rows_updated == 1
+ assert upd.rows_inserted == 0
+ assert table.scan().to_arrow() == pa.Table.from_pylist(
+ [
+ {"foo": "apple", "bar": 7, "baz": False},
+ {"foo": "banana", "bar": None, "baz": False},
+ ],
+ schema=schema,
+ )
(iceberg-python) 10/11: Fix for metadata entries table for MOR tables containing Delete Files. (#1902)
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.9.x
in repository https://gitbox.apache.org/repos/asf/iceberg-python.git
commit ddd72250d254dfe31b38656986a6fcf0794bdd42
Author: Akashdeep Gupta <gu...@users.noreply.github.com>
AuthorDate: Thu Apr 17 02:15:02 2025 +0530
Fix for metadata entries table for MOR tables containing Delete Files. (#1902)
<!--
Thanks for opening a pull request!
-->
<!-- In the case this PR will resolve an issue, please replace
${GITHUB_ISSUE_ID} below with the actual Github issue id. -->
Closes #1884
# Rationale for this change
table.inspect.entries() fails when table is MOR table and has Delete
Files present in it. Iceberg MOR Table is created via Apache Spark 3.5.0
with Iceberg 1.5.0 and it's being read via PyIceberg 0.9.0 using
StaticTable.from_metadata()
# Are these changes tested?
Yes
# Are there any user-facing changes?
No
<!-- In the case of user-facing changes, please add the changelog label.
-->
---
pyiceberg/table/inspect.py | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
diff --git a/pyiceberg/table/inspect.py b/pyiceberg/table/inspect.py
index 91bdb2f2..b850aa10 100644
--- a/pyiceberg/table/inspect.py
+++ b/pyiceberg/table/inspect.py
@@ -205,9 +205,9 @@ class InspectTable:
"record_count": entry.data_file.record_count,
"file_size_in_bytes": entry.data_file.file_size_in_bytes,
"column_sizes": dict(entry.data_file.column_sizes),
- "value_counts": dict(entry.data_file.value_counts),
- "null_value_counts": dict(entry.data_file.null_value_counts),
- "nan_value_counts": dict(entry.data_file.nan_value_counts),
+ "value_counts": dict(entry.data_file.value_counts or {}),
+ "null_value_counts": dict(entry.data_file.null_value_counts or {}),
+ "nan_value_counts": dict(entry.data_file.nan_value_counts or {}),
"lower_bounds": entry.data_file.lower_bounds,
"upper_bounds": entry.data_file.upper_bounds,
"key_metadata": entry.data_file.key_metadata,