Merge branch 'main' into ns-timestamp

Author: sungwy

.github/workflows/pypi-build-artifacts.yml

@@ -62,7 +62,7 @@ jobs:
         if: startsWith(matrix.os, 'ubuntu')
 
       - name: Build wheels
-        uses: pypa/cibuildwheel@v2.23.0
+        uses: pypa/cibuildwheel@v2.23.1
         with:
           output-dir: wheelhouse
           config-file: "pyproject.toml"

.github/workflows/svn-build-artifacts.yml

@@ -57,7 +57,7 @@ jobs:
         if: startsWith(matrix.os, 'ubuntu')
 
       - name: Build wheels
-        uses: pypa/cibuildwheel@v2.23.0
+        uses: pypa/cibuildwheel@v2.23.1
         with:
           output-dir: wheelhouse
           config-file: "pyproject.toml"

poetry.lock

@@ -20,6 +20,7 @@
     - Converting partition strings to built-in python objects.
     - Converting a value to a byte buffer.
     - Converting a byte buffer to a value.
+    - Converting a json-single field serialized field
 
 Note:
     Conversion logic varies based on the PrimitiveType implementation. Therefore conversion functions
@@ -28,6 +29,7 @@
     implementations that share the same conversion logic, registrations can be stacked.
 """
 
+import codecs
 import uuid
 from datetime import date, datetime, time
 from decimal import Decimal
@@ -62,7 +64,24 @@
     UUIDType,
     strtobool,
 )
-from pyiceberg.utils.datetime import date_to_days, datetime_to_micros, datetime_to_nanos, time_to_micros
+from pyiceberg.utils.datetime import (
+    date_str_to_days,
+    date_to_days,
+    datetime_to_micros,
+    datetime_to_nanos,
+    days_to_date,
+    micros_to_time,
+    micros_to_timestamp,
+    micros_to_timestamptz,
+    time_str_to_micros,
+    time_to_micros,
+    timestamp_to_micros,
+    timestamptz_to_micros,
+    to_human_day,
+    to_human_time,
+    to_human_timestamp,
+    to_human_timestamptz,
+)
 from pyiceberg.utils.decimal import decimal_to_bytes, unscaled_to_decimal
 
 _BOOL_STRUCT = Struct("<?")
@@ -295,7 +314,7 @@ def from_bytes(primitive_type: PrimitiveType, b: bytes) -> L:  # type: ignore
         primitive_type (PrimitiveType): An implementation of the PrimitiveType base class.
         b (bytes): The bytes to convert.
     """
-    raise TypeError(f"Cannot deserialize bytes, type {primitive_type} not supported: {str(b)}")
+    raise TypeError(f"Cannot deserialize bytes, type {primitive_type} not supported: {b!r}")
 
 
 @from_bytes.register(BooleanType)
@@ -350,3 +369,200 @@ def _(primitive_type: DecimalType, buf: bytes) -> Decimal:
 @from_bytes.register(UnknownType)
 def _(type_: UnknownType, buf: bytes) -> None:
     return None
+
+
+@singledispatch  # type: ignore
+def to_json(primitive_type: PrimitiveType, val: Any) -> L:  # type: ignore
+    """Convert built-in python values into JSON value types.
+
+    https://iceberg.apache.org/spec/#json-single-value-serialization
+
+    Args:
+        primitive_type (PrimitiveType): An implementation of the PrimitiveType base class.
+        val (Any): The arbitrary built-in value to convert into the right form
+    """
+    raise TypeError(f"Cannot deserialize bytes, type {primitive_type} not supported: {val}")
+
+
+@to_json.register(BooleanType)
+def _(_: BooleanType, val: bool) -> bool:
+    """Python bool automatically converts into a JSON bool."""
+    return val
+
+
+@to_json.register(IntegerType)
+@to_json.register(LongType)
+def _(_: Union[IntegerType, LongType], val: int) -> int:
+    """Python int automatically converts to a JSON int."""
+    return val
+
+
+@to_json.register(DateType)
+def _(_: DateType, val: Union[date, int]) -> str:
+    """JSON date is string encoded."""
+    if isinstance(val, date):
+        val = date_to_days(val)
+    return to_human_day(val)
+
+
+@to_json.register(TimeType)
+def _(_: TimeType, val: Union[int, time]) -> str:
+    """Python time or microseconds since epoch serializes into an ISO8601 time."""
+    if isinstance(val, time):
+        val = time_to_micros(val)
+    return to_human_time(val)
+
+
+@to_json.register(TimestampType)
+def _(_: PrimitiveType, val: Union[int, datetime]) -> str:
+    """Python datetime (without timezone) or microseconds since epoch serializes into an ISO8601 timestamp."""
+    if isinstance(val, datetime):
+        val = datetime_to_micros(val)
+
+    return to_human_timestamp(val)
+
+
+@to_json.register(TimestamptzType)
+def _(_: TimestamptzType, val: Union[int, datetime]) -> str:
+    """Python datetime (with timezone) or microseconds since epoch serializes into an ISO8601 timestamp."""
+    if isinstance(val, datetime):
+        val = datetime_to_micros(val)
+    return to_human_timestamptz(val)
+
+
+@to_json.register(FloatType)
+@to_json.register(DoubleType)
+def _(_: Union[FloatType, DoubleType], val: float) -> float:
+    """Float serializes into JSON float."""
+    return val
+
+
+@to_json.register(StringType)
+def _(_: StringType, val: str) -> str:
+    """Python string serializes into JSON string."""
+    return val
+
+
+@to_json.register(FixedType)
+def _(t: FixedType, b: bytes) -> str:
+    """Python bytes serializes into hexadecimal encoded string."""
+    if len(t) != len(b):
+        raise ValueError(f"FixedType has length {len(t)}, which is different from the value: {len(b)}")
+
+    return codecs.encode(b, "hex").decode(UTF8)
+
+
+@to_json.register(BinaryType)
+def _(_: BinaryType, b: bytes) -> str:
+    """Python bytes serializes into hexadecimal encoded string."""
+    return codecs.encode(b, "hex").decode(UTF8)
+
+
+@to_json.register(DecimalType)
+def _(_: DecimalType, val: Decimal) -> str:
+    """Python decimal serializes into string.
+
+    Stores the string representation of the decimal value, specifically, for
+    values with a positive scale, the number of digits to the right of the
+    decimal point is used to indicate scale, for values with a negative scale,
+    the scientific notation is used and the exponent must equal the negated scale.
+    """
+    return str(val)
+
+
+@to_json.register(UUIDType)
+def _(_: UUIDType, val: uuid.UUID) -> str:
+    """Serialize into a JSON string."""
+    return str(val)
+
+
+@singledispatch  # type: ignore
+def from_json(primitive_type: PrimitiveType, val: Any) -> L:  # type: ignore
+    """Convert JSON value types into built-in python values.
+
+    https://iceberg.apache.org/spec/#json-single-value-serialization
+
+    Args:
+        primitive_type (PrimitiveType): An implementation of the PrimitiveType base class.
+        val (Any): The arbitrary JSON value to convert into the right form
+    """
+    raise TypeError(f"Cannot deserialize bytes, type {primitive_type} not supported: {str(val)}")
+
+
+@from_json.register(BooleanType)
+def _(_: BooleanType, val: bool) -> bool:
+    """JSON bool automatically converts into a Python bool."""
+    return val
+
+
+@from_json.register(IntegerType)
+@from_json.register(LongType)
+def _(_: Union[IntegerType, LongType], val: int) -> int:
+    """JSON int automatically converts to a Python int."""
+    return val
+
+
+@from_json.register(DateType)
+def _(_: DateType, val: str) -> date:
+    """JSON date is string encoded."""
+    return days_to_date(date_str_to_days(val))
+
+
+@from_json.register(TimeType)
+def _(_: TimeType, val: str) -> time:
+    """JSON ISO8601 string into Python time."""
+    return micros_to_time(time_str_to_micros(val))
+
+
+@from_json.register(TimestampType)
+def _(_: PrimitiveType, val: str) -> datetime:
+    """JSON ISO8601 string into Python datetime."""
+    return micros_to_timestamp(timestamp_to_micros(val))
+
+
+@from_json.register(TimestamptzType)
+def _(_: TimestamptzType, val: str) -> datetime:
+    """JSON ISO8601 string into Python datetime."""
+    return micros_to_timestamptz(timestamptz_to_micros(val))
+
+
+@from_json.register(FloatType)
+@from_json.register(DoubleType)
+def _(_: Union[FloatType, DoubleType], val: float) -> float:
+    """JSON float deserializes into a Python float."""
+    return val
+
+
+@from_json.register(StringType)
+def _(_: StringType, val: str) -> str:
+    """JSON string serializes into a Python string."""
+    return val
+
+
+@from_json.register(FixedType)
+def _(t: FixedType, val: str) -> bytes:
+    """JSON hexadecimal encoded string into bytes."""
+    b = codecs.decode(val.encode(UTF8), "hex")
+
+    if len(t) != len(b):
+        raise ValueError(f"FixedType has length {len(t)}, which is different from the value: {len(b)}")
+
+    return b
+
+
+@from_json.register(BinaryType)
+def _(_: BinaryType, val: str) -> bytes:
+    """JSON hexadecimal encoded string into bytes."""
+    return codecs.decode(val.encode(UTF8), "hex")
+
+
+@from_json.register(DecimalType)
+def _(_: DecimalType, val: str) -> Decimal:
+    """Convert JSON string into a Python Decimal."""
+    return Decimal(val)
+
+
+@from_json.register(UUIDType)
+def _(_: UUIDType, val: str) -> uuid.UUID:
+    """Convert JSON string into Python UUID."""
+    return uuid.UUID(val)

pyiceberg/conversions.py

@@ -1198,10 +1198,11 @@ def upsert(
 
                 update_row_cnt = len(rows_to_update)
 
-                # build the match predicate filter
-                overwrite_mask_predicate = upsert_util.create_match_filter(rows_to_update, join_cols)
+                if len(rows_to_update) > 0:
+                    # build the match predicate filter
+                    overwrite_mask_predicate = upsert_util.create_match_filter(rows_to_update, join_cols)
 
-                tx.overwrite(rows_to_update, overwrite_filter=overwrite_mask_predicate)
+                    tx.overwrite(rows_to_update, overwrite_filter=overwrite_mask_predicate)
 
             if when_not_matched_insert_all:
                 expr_match = upsert_util.create_match_filter(matched_iceberg_table, join_cols)
@@ -1211,7 +1212,8 @@ def upsert(
 
                 insert_row_cnt = len(rows_to_insert)
 
-                tx.append(rows_to_insert)
+                if insert_row_cnt > 0:
+                    tx.append(rows_to_insert)
 
         return UpsertResult(rows_updated=update_row_cnt, rows_inserted=insert_row_cnt)
 

pyiceberg/table/__init__.py

@@ -161,7 +161,7 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType:
         entries = []
         snapshot = self._get_snapshot(snapshot_id)
         for manifest in snapshot.manifests(self.tbl.io):
-            for entry in manifest.fetch_manifest_entry(io=self.tbl.io):
+            for entry in manifest.fetch_manifest_entry(io=self.tbl.io, discard_deleted=False):
                 column_sizes = entry.data_file.column_sizes or {}
                 value_counts = entry.data_file.value_counts or {}
                 null_value_counts = entry.data_file.null_value_counts or {}

pyiceberg/table/inspect.py

@@ -87,7 +87,7 @@ thrift-sasl = { version = ">=0.4.3", optional = true }
 pytest = "7.4.4"
 pytest-checkdocs = "2.13.0"
 pytest-lazy-fixture = "0.6.3"
-pre-commit = "4.1.0"
+pre-commit = "4.2.0"
 fastavro = "1.10.0"
 coverage = { version = "^7.4.2", extras = ["toml"] }
 requests-mock = "1.12.1"
@@ -103,14 +103,14 @@ docutils = "!=0.21.post1"   # https://github.com/python-poetry/poetry/issues/924
 [tool.poetry.group.docs.dependencies]
 # for mkdocs
 mkdocs = "1.6.1"
-griffe = "1.6.0"
+griffe = "1.6.1"
 jinja2 = "3.1.6"
 mkdocstrings = "0.29.0"
-mkdocstrings-python = "1.16.5"
-mkdocs-literate-nav = "0.6.1"
+mkdocstrings-python = "1.16.6"
+mkdocs-literate-nav = "0.6.2"
 mkdocs-autorefs = "1.4.1"
 mkdocs-gen-files = "0.5.0"
-mkdocs-material = "9.6.8"
+mkdocs-material = "9.6.9"
 mkdocs-material-extensions = "1.3.1"
 mkdocs-section-index = "0.3.9"
 

pyproject.toml

@@ -164,6 +164,8 @@ def test_inspect_entries(
 
     # Write some data
     tbl.append(arrow_table_with_null)
+    # Generate a DELETE entry
+    tbl.overwrite(arrow_table_with_null)
 
     def check_pyiceberg_df_equals_spark_df(df: pa.Table, spark_df: DataFrame) -> None:
         assert df.column_names == [
@@ -185,6 +187,8 @@ def check_pyiceberg_df_equals_spark_df(df: pa.Table, spark_df: DataFrame) -> Non
 
         lhs = df.to_pandas()
         rhs = spark_df.toPandas()
+        assert len(lhs) == len(rhs)
+
         for column in df.column_names:
             for left, right in zip(lhs[column].to_list(), rhs[column].to_list()):
                 if column == "data_file":

tests/integration/test_inspect_table.py

@@ -28,6 +28,7 @@
 from pyiceberg.io.pyarrow import schema_to_pyarrow
 from pyiceberg.schema import Schema
 from pyiceberg.table import UpsertResult
+from pyiceberg.table.snapshots import Operation
 from pyiceberg.table.upsert_util import create_match_filter
 from pyiceberg.types import IntegerType, NestedField, StringType
 from tests.catalog.test_base import InMemoryCatalog, Table
@@ -368,9 +369,21 @@ def test_upsert_with_identifier_fields(catalog: Catalog) -> None:
     )
     upd = tbl.upsert(df)
 
+    expected_operations = [Operation.APPEND, Operation.OVERWRITE, Operation.APPEND, Operation.APPEND]
+
     assert upd.rows_updated == 1
     assert upd.rows_inserted == 1
 
+    assert [snap.summary.operation for snap in tbl.snapshots() if snap.summary is not None] == expected_operations
+
+    # This should be a no-op
+    upd = tbl.upsert(df)
+
+    assert upd.rows_updated == 0
+    assert upd.rows_inserted == 0
+
+    assert [snap.summary.operation for snap in tbl.snapshots() if snap.summary is not None] == expected_operations
+
 
 def test_upsert_into_empty_table(catalog: Catalog) -> None:
     identifier = "default.test_upsert_into_empty_table"