|
14 | 14 | # KIND, either express or implied. See the License for the |
15 | 15 | # specific language governing permissions and limitations |
16 | 16 | # under the License. |
| 17 | +import datetime |
17 | 18 | from pathlib import PosixPath |
18 | 19 |
|
19 | 20 | import pyarrow as pa |
|
26 | 27 | from pyiceberg.expressions import AlwaysTrue, And, EqualTo, Reference |
27 | 28 | from pyiceberg.expressions.literals import LongLiteral |
28 | 29 | from pyiceberg.io.pyarrow import schema_to_pyarrow |
| 30 | +from pyiceberg.partitioning import PartitionField, PartitionSpec |
29 | 31 | from pyiceberg.schema import Schema |
30 | 32 | from pyiceberg.table import Table, UpsertResult |
31 | 33 | from pyiceberg.table.snapshots import Operation |
32 | 34 | from pyiceberg.table.upsert_util import create_match_filter |
33 | | -from pyiceberg.types import IntegerType, NestedField, StringType, StructType |
| 35 | +from pyiceberg.transforms import IdentityTransform |
| 36 | +from pyiceberg.types import DateType, IntegerType, NestedField, StringType, StructType |
34 | 37 | from tests.catalog.test_base import InMemoryCatalog |
35 | 38 |
|
36 | 39 |
|
@@ -714,6 +717,48 @@ def test_upsert_with_nulls(catalog: Catalog) -> None: |
714 | 717 | ) |
715 | 718 |
|
716 | 719 |
|
| 720 | +def test_upsert_updates_existing_row_when_non_join_partition_value_changes(catalog: Catalog) -> None: |
| 721 | + identifier = "default.test_upsert_non_join_partition_value_change" |
| 722 | + _drop_table(catalog, identifier) |
| 723 | + |
| 724 | + schema = Schema( |
| 725 | + NestedField(1, "order_id", IntegerType(), required=True), |
| 726 | + NestedField(2, "order_date", DateType(), required=True), |
| 727 | + NestedField(3, "order_type", StringType(), required=True), |
| 728 | + ) |
| 729 | + spec = PartitionSpec(PartitionField(source_id=2, field_id=1000, transform=IdentityTransform(), name="order_date")) |
| 730 | + table = catalog.create_table(identifier, schema=schema, partition_spec=spec) |
| 731 | + |
| 732 | + arrow_schema = pa.schema( |
| 733 | + [ |
| 734 | + pa.field("order_id", pa.int32(), nullable=False), |
| 735 | + pa.field("order_date", pa.date32(), nullable=False), |
| 736 | + pa.field("order_type", pa.string(), nullable=False), |
| 737 | + ] |
| 738 | + ) |
| 739 | + table.append( |
| 740 | + pa.Table.from_pylist( |
| 741 | + [{"order_id": 1, "order_date": datetime.date(2026, 1, 1), "order_type": "A"}], |
| 742 | + schema=arrow_schema, |
| 743 | + ) |
| 744 | + ) |
| 745 | + |
| 746 | + result = table.upsert( |
| 747 | + pa.Table.from_pylist( |
| 748 | + [{"order_id": 1, "order_date": datetime.date(2026, 1, 2), "order_type": "B"}], |
| 749 | + schema=arrow_schema, |
| 750 | + ), |
| 751 | + join_cols=["order_id"], |
| 752 | + ) |
| 753 | + |
| 754 | + assert result.rows_updated == 1 |
| 755 | + assert result.rows_inserted == 0 |
| 756 | + assert table.scan().to_arrow() == pa.Table.from_pylist( |
| 757 | + [{"order_id": 1, "order_date": datetime.date(2026, 1, 2), "order_type": "B"}], |
| 758 | + schema=arrow_schema, |
| 759 | + ) |
| 760 | + |
| 761 | + |
717 | 762 | def test_transaction(catalog: Catalog) -> None: |
718 | 763 | """Test the upsert within a Transaction. Make sure that if something fails the entire Transaction is |
719 | 764 | rolled back.""" |
|
0 commit comments