Merge branch 'main' of github.com:apache/iceberg-python into fd-use-iceberg-rust

Author: Fokko

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

@@ -62,7 +62,7 @@ jobs:
         if: startsWith(matrix.os, 'ubuntu')
 
       - name: Build wheels
-        uses: pypa/cibuildwheel@v2.23.1
+        uses: pypa/cibuildwheel@v2.23.2
         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.1
+        uses: pypa/cibuildwheel@v2.23.2
         with:
           output-dir: wheelhouse
           config-file: "pyproject.toml"

dev/provision.py

@@ -328,6 +328,7 @@
     CREATE TABLE {catalog_name}.default.test_table_empty_list_and_map (
         col_list             array<int>,
         col_map              map<int, int>,
+        col_struct           struct<test:int>,
         col_list_with_struct array<struct<test:int>>
     )
     USING iceberg
@@ -340,8 +341,8 @@
     spark.sql(
         f"""
     INSERT INTO {catalog_name}.default.test_table_empty_list_and_map
-    VALUES (null, null, null),
-           (array(), map(), array(struct(1)))
+    VALUES (null, null, null, null),
+           (array(), map(), struct(1), array(struct(1)))
     """
     )
 

poetry.lock

@@ -63,6 +63,7 @@
 S3_PROXY_URI = "s3.proxy-uri"
 S3_CONNECT_TIMEOUT = "s3.connect-timeout"
 S3_REQUEST_TIMEOUT = "s3.request-timeout"
+S3_SIGNER = "s3.signer"
 S3_SIGNER_URI = "s3.signer.uri"
 S3_SIGNER_ENDPOINT = "s3.signer.endpoint"
 S3_SIGNER_ENDPOINT_DEFAULT = "v1/aws/s3/sign"

pyiceberg/io/__init__.py

@@ -67,6 +67,7 @@
     S3_REQUEST_TIMEOUT,
     S3_SECRET_ACCESS_KEY,
     S3_SESSION_TOKEN,
+    S3_SIGNER,
     S3_SIGNER_ENDPOINT,
     S3_SIGNER_ENDPOINT_DEFAULT,
     S3_SIGNER_URI,
@@ -137,7 +138,7 @@ def _s3(properties: Properties) -> AbstractFileSystem:
     config_kwargs = {}
     register_events: Dict[str, Callable[[Properties], None]] = {}
 
-    if signer := properties.get("s3.signer"):
+    if signer := properties.get(S3_SIGNER):
         logger.info("Loading signer %s", signer)
         if signer_func := SIGNERS.get(signer):
             signer_func_with_properties = partial(signer_func, properties)

pyiceberg/io/fsspec.py

@@ -1240,6 +1240,7 @@ class _BuildPositionAccessors(SchemaVisitor[Dict[Position, Accessor]]):
         ...     1: Accessor(position=1, inner=None),
         ...     5: Accessor(position=2, inner=Accessor(position=0, inner=None)),
         ...     6: Accessor(position=2, inner=Accessor(position=1, inner=None))
+        ...     3: Accessor(position=2, inner=None),
         ... }
         >>> result == expected
         True
@@ -1255,8 +1256,7 @@ def struct(self, struct: StructType, field_results: List[Dict[Position, Accessor
             if field_results[position]:
                 for inner_field_id, acc in field_results[position].items():
                     result[inner_field_id] = Accessor(position, inner=acc)
-            else:
-                result[field.field_id] = Accessor(position)
+            result[field.field_id] = Accessor(position)
 
         return result
 

pyiceberg/schema.py

@@ -106,7 +106,7 @@ mkdocs = "1.6.1"
 griffe = "1.6.2"
 jinja2 = "3.1.6"
 mkdocstrings = "0.29.0"
-mkdocstrings-python = "1.16.7"
+mkdocstrings-python = "1.16.8"
 mkdocs-literate-nav = "0.6.2"
 mkdocs-autorefs = "1.4.1"
 mkdocs-gen-files = "0.5.0"

pyproject.toml

@@ -168,6 +168,23 @@ def test_notnull_bind_required() -> None:
     assert NotNull(Reference("a")).bind(schema) == AlwaysTrue()
 
 
+def test_notnull_bind_top_struct() -> None:
+    schema = Schema(
+        NestedField(
+            3,
+            "struct_col",
+            required=False,
+            field_type=StructType(
+                NestedField(1, "id", IntegerType(), required=True),
+                NestedField(2, "cost", DecimalType(38, 18), required=False),
+            ),
+        ),
+        schema_id=1,
+    )
+    bound = BoundNotNull(BoundReference(schema.find_field(3), schema.accessor_for_field(3)))
+    assert NotNull(Reference("struct_col")).bind(schema) == bound
+
+
 def test_isnan_inverse() -> None:
     assert ~IsNaN(Reference("f")) == NotNaN(Reference("f"))
 

tests/expressions/test_expressions.py

@@ -41,6 +41,7 @@
     LessThan,
     NotEqualTo,
     NotNaN,
+    NotNull,
 )
 from pyiceberg.io import PYARROW_USE_LARGE_TYPES_ON_READ
 from pyiceberg.io.pyarrow import (
@@ -667,6 +668,24 @@ def test_filter_case_insensitive(catalog: Catalog) -> None:
     assert arrow_table["b"].to_pylist() == ["2"]
 
 
+@pytest.mark.integration
+@pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog_hive"), pytest.lazy_fixture("session_catalog")])
+def test_filters_on_top_level_struct(catalog: Catalog) -> None:
+    test_empty_struct = catalog.load_table("default.test_table_empty_list_and_map")
+
+    arrow_table = test_empty_struct.scan().to_arrow()
+    assert None in arrow_table["col_struct"].to_pylist()
+
+    arrow_table = test_empty_struct.scan(row_filter=NotNull("col_struct")).to_arrow()
+    assert arrow_table["col_struct"].to_pylist() == [{"test": 1}]
+
+    arrow_table = test_empty_struct.scan(row_filter="col_struct is not null", case_sensitive=False).to_arrow()
+    assert arrow_table["col_struct"].to_pylist() == [{"test": 1}]
+
+    arrow_table = test_empty_struct.scan(row_filter="COL_STRUCT is null", case_sensitive=False).to_arrow()
+    assert arrow_table["col_struct"].to_pylist() == [None]
+
+
 @pytest.mark.integration
 @pytest.mark.parametrize("catalog", [pytest.lazy_fixture("session_catalog_hive"), pytest.lazy_fixture("session_catalog")])
 def test_upgrade_table_version(catalog: Catalog) -> None: