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

Author: Fokko

.github/ISSUE_TEMPLATE/iceberg_bug_report.yml

@@ -28,7 +28,8 @@ body:
       description: What Apache Iceberg version are you using?
       multiple: false
       options:
-        - "0.8.1 (latest release)"
+        - "0.9.0 (latest release)"
+        - "0.8.1"
         - "0.8.0"
         - "0.7.1"
         - "0.7.0"

.github/pull_request_template.md

@@ -0,0 +1,14 @@
+<!--
+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. -->

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

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

dev/.rat-excludes

@@ -1,3 +1,4 @@
+.github/*
 .rat-excludes
 build
 .git

dev/Dockerfile

@@ -40,7 +40,7 @@ WORKDIR ${SPARK_HOME}
 ENV SPARK_VERSION=3.5.4
 ENV ICEBERG_SPARK_RUNTIME_VERSION=3.5_2.12
 ENV ICEBERG_VERSION=1.8.0
-ENV PYICEBERG_VERSION=0.8.1
+ENV PYICEBERG_VERSION=0.9.0
 
 RUN curl --retry 5 -s -C - https://archive.apache.org/dist/spark/spark-${SPARK_VERSION}/spark-${SPARK_VERSION}-bin-hadoop3.tgz -o spark-${SPARK_VERSION}-bin-hadoop3.tgz \
  && tar xzf spark-${SPARK_VERSION}-bin-hadoop3.tgz --directory /opt/spark --strip-components 1 \

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)))
     """
     )
 

mkdocs/docs/how-to-release.md

@@ -379,6 +379,8 @@ Then, select the previous release version as the **Previous tag** to use the dif
 
 **Set as the latest release** and **Publish**.
 
+Make sure to check the `changelog` label on GitHub to see if anything needs to be highlighted.
+
 ### Release the docs
 
 Run the [`Release Docs` Github Action](https://github.com/apache/iceberg-python/actions/workflows/python-release-docs.yml).

poetry.lock

@@ -15,4 +15,4 @@
 # specific language governing permissions and limitations
 # under the License.
 
-__version__ = "0.9.0"
+__version__ = "0.10.0"

pyiceberg/__init__.py

@@ -14,6 +14,7 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
+from typing import Any
 from uuid import UUID
 
 from pyiceberg.avro import STRUCT_DOUBLE, STRUCT_FLOAT
@@ -74,3 +75,6 @@ def write_uuid(self, uuid: UUID) -> None:
         if len(uuid.bytes) != 16:
             raise ValueError(f"Expected UUID to have 16 bytes, got: len({uuid.bytes!r})")
         return self.write(uuid.bytes)
+
+    def write_unknown(self, _: Any) -> None:
+        """Nulls are written as 0 bytes in avro, so we do nothing."""

pyiceberg/avro/encoder.py

@@ -175,6 +175,14 @@ class TimestampReader(IntegerReader):
     """
 
 
+class TimestampNanoReader(IntegerReader):
+    """Reads a nanosecond granularity timestamp from the stream.
+
+    Long is decoded as python integer which represents
+    the number of nanoseconds from the unix epoch, 1 January 1970.
+    """
+
+
 class TimestamptzReader(IntegerReader):
     """Reads a microsecond granularity timestamptz from the stream.
 
@@ -185,6 +193,16 @@ class TimestamptzReader(IntegerReader):
     """
 
 
+class TimestamptzNanoReader(IntegerReader):
+    """Reads a microsecond granularity timestamptz from the stream.
+
+    Long is decoded as python integer which represents
+    the number of nanoseconds from the unix epoch, 1 January 1970.
+
+    Adjusted to UTC.
+    """
+
+
 class StringReader(Reader):
     def read(self, decoder: BinaryDecoder) -> str:
         return decoder.read_utf8()
@@ -201,6 +219,14 @@ def skip(self, decoder: BinaryDecoder) -> None:
         decoder.skip(16)
 
 
+class UnknownReader(Reader):
+    def read(self, decoder: BinaryDecoder) -> None:
+        return None
+
+    def skip(self, decoder: BinaryDecoder) -> None:
+        pass
+
+
 @dataclass(frozen=True)
 class FixedReader(Reader):
     _len: int = dataclassfield()

pyiceberg/avro/reader.py

@@ -44,8 +44,11 @@
     StringReader,
     StructReader,
     TimeReader,
+    TimestampNanoReader,
     TimestampReader,
+    TimestamptzNanoReader,
     TimestamptzReader,
+    UnknownReader,
     UUIDReader,
 )
 from pyiceberg.avro.writer import (
@@ -63,9 +66,12 @@
     OptionWriter,
     StringWriter,
     StructWriter,
+    TimestampNanoWriter,
+    TimestamptzNanoWriter,
     TimestamptzWriter,
     TimestampWriter,
     TimeWriter,
+    UnknownWriter,
     UUIDWriter,
     Writer,
 )
@@ -97,9 +103,12 @@
     PrimitiveType,
     StringType,
     StructType,
+    TimestampNanoType,
     TimestampType,
+    TimestamptzNanoType,
     TimestamptzType,
     TimeType,
+    UnknownType,
     UUIDType,
 )
 
@@ -181,9 +190,15 @@ def visit_time(self, time_type: TimeType) -> Writer:
     def visit_timestamp(self, timestamp_type: TimestampType) -> Writer:
         return TimestampWriter()
 
+    def visit_timestamp_ns(self, timestamp_ns_type: TimestampNanoType) -> Writer:
+        return TimestampNanoWriter()
+
     def visit_timestamptz(self, timestamptz_type: TimestamptzType) -> Writer:
         return TimestamptzWriter()
 
+    def visit_timestamptz_ns(self, timestamptz_ns_type: TimestamptzNanoType) -> Writer:
+        return TimestamptzNanoWriter()
+
     def visit_string(self, string_type: StringType) -> Writer:
         return StringWriter()
 
@@ -193,6 +208,9 @@ def visit_uuid(self, uuid_type: UUIDType) -> Writer:
     def visit_binary(self, binary_type: BinaryType) -> Writer:
         return BinaryWriter()
 
+    def visit_unknown(self, unknown_type: UnknownType) -> Writer:
+        return UnknownWriter()
+
 
 CONSTRUCT_WRITER_VISITOR = ConstructWriter()
 
@@ -326,9 +344,15 @@ def visit_time(self, time_type: TimeType, partner: Optional[IcebergType]) -> Wri
     def visit_timestamp(self, timestamp_type: TimestampType, partner: Optional[IcebergType]) -> Writer:
         return TimestampWriter()
 
+    def visit_timestamp_ns(self, timestamp_ns_type: TimestampNanoType, partner: Optional[IcebergType]) -> Writer:
+        return TimestampNanoWriter()
+
     def visit_timestamptz(self, timestamptz_type: TimestamptzType, partner: Optional[IcebergType]) -> Writer:
         return TimestamptzWriter()
 
+    def visit_timestamptz_ns(self, timestamptz_ns_type: TimestamptzNanoType, partner: Optional[IcebergType]) -> Writer:
+        return TimestamptzNanoWriter()
+
     def visit_string(self, string_type: StringType, partner: Optional[IcebergType]) -> Writer:
         return StringWriter()
 
@@ -341,6 +365,9 @@ def visit_fixed(self, fixed_type: FixedType, partner: Optional[IcebergType]) ->
     def visit_binary(self, binary_type: BinaryType, partner: Optional[IcebergType]) -> Writer:
         return BinaryWriter()
 
+    def visit_unknown(self, unknown_type: UnknownType, partner: Optional[IcebergType]) -> Writer:
+        return UnknownWriter()
+
 
 class ReadSchemaResolver(PrimitiveWithPartnerVisitor[IcebergType, Reader]):
     __slots__ = ("read_types", "read_enums", "context")
@@ -456,9 +483,15 @@ def visit_time(self, time_type: TimeType, partner: Optional[IcebergType]) -> Rea
     def visit_timestamp(self, timestamp_type: TimestampType, partner: Optional[IcebergType]) -> Reader:
         return TimestampReader()
 
+    def visit_timestamp_ns(self, timestamp_ns_type: TimestampNanoType, partner: Optional[IcebergType]) -> Reader:
+        return TimestampNanoReader()
+
     def visit_timestamptz(self, timestamptz_type: TimestamptzType, partner: Optional[IcebergType]) -> Reader:
         return TimestamptzReader()
 
+    def visit_timestamptz_ns(self, timestamptz_ns_type: TimestamptzNanoType, partner: Optional[IcebergType]) -> Reader:
+        return TimestamptzNanoReader()
+
     def visit_string(self, string_type: StringType, partner: Optional[IcebergType]) -> Reader:
         return StringReader()
 
@@ -471,6 +504,9 @@ def visit_fixed(self, fixed_type: FixedType, partner: Optional[IcebergType]) ->
     def visit_binary(self, binary_type: BinaryType, partner: Optional[IcebergType]) -> Reader:
         return BinaryReader()
 
+    def visit_unknown(self, unknown_type: UnknownType, partner: Optional[IcebergType]) -> Reader:
+        return UnknownReader()
+
 
 class SchemaPartnerAccessor(PartnerAccessor[IcebergType]):
     def schema_partner(self, partner: Optional[IcebergType]) -> Optional[IcebergType]:

pyiceberg/avro/resolver.py

@@ -95,12 +95,24 @@ def write(self, encoder: BinaryEncoder, val: int) -> None:
         encoder.write_int(val)
 
 
+@dataclass(frozen=True)
+class TimestampNanoWriter(Writer):
+    def write(self, encoder: BinaryEncoder, val: int) -> None:
+        encoder.write_int(val)
+
+
 @dataclass(frozen=True)
 class TimestamptzWriter(Writer):
     def write(self, encoder: BinaryEncoder, val: int) -> None:
         encoder.write_int(val)
 
 
+@dataclass(frozen=True)
+class TimestamptzNanoWriter(Writer):
+    def write(self, encoder: BinaryEncoder, val: int) -> None:
+        encoder.write_int(val)
+
+
 @dataclass(frozen=True)
 class StringWriter(Writer):
     def write(self, encoder: BinaryEncoder, val: Any) -> None:
@@ -113,6 +125,12 @@ def write(self, encoder: BinaryEncoder, val: UUID) -> None:
         encoder.write(val.bytes)
 
 
+@dataclass(frozen=True)
+class UnknownWriter(Writer):
+    def write(self, encoder: BinaryEncoder, val: Any) -> None:
+        encoder.write_unknown(val)
+
+
 @dataclass(frozen=True)
 class FixedWriter(Writer):
     _len: int = dataclassfield()

pyiceberg/avro/writer.py

@@ -110,6 +110,7 @@
     TimestampType,
     TimestamptzType,
     TimeType,
+    UnknownType,
     UUIDType,
 )
 from pyiceberg.utils.properties import property_as_bool, property_as_float
@@ -236,6 +237,7 @@ def _annotate_namespace(database: HiveDatabase, properties: Properties) -> HiveD
     UUIDType: "string",
     BinaryType: "binary",
     FixedType: "binary",
+    UnknownType: "void",
 }
 
 

pyiceberg/catalog/hive.py

@@ -148,7 +148,7 @@ def _retry_hook(retry_state: RetryCallState) -> None:
 
 
 _RETRY_ARGS = {
-    "retry": retry_if_exception_type(AuthorizationExpiredError),
+    "retry": retry_if_exception_type((AuthorizationExpiredError, UnauthorizedError)),
     "stop": stop_after_attempt(2),
     "before_sleep": _retry_hook,
     "reraise": True,

pyiceberg/catalog/rest.py

@@ -619,15 +619,28 @@ def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identi
         table_stmt = select(IcebergTables.table_namespace).where(IcebergTables.catalog_name == self.name)
         namespace_stmt = select(IcebergNamespaceProperties.namespace).where(IcebergNamespaceProperties.catalog_name == self.name)
         if namespace:
-            namespace_str = Catalog.namespace_to_string(namespace, NoSuchNamespaceError)
-            table_stmt = table_stmt.where(IcebergTables.table_namespace.like(namespace_str))
-            namespace_stmt = namespace_stmt.where(IcebergNamespaceProperties.namespace.like(namespace_str))
+            namespace_like = Catalog.namespace_to_string(namespace, NoSuchNamespaceError) + "%"
+            table_stmt = table_stmt.where(IcebergTables.table_namespace.like(namespace_like))
+            namespace_stmt = namespace_stmt.where(IcebergNamespaceProperties.namespace.like(namespace_like))
         stmt = union(
             table_stmt,
             namespace_stmt,
         )
         with Session(self.engine) as session:
-            return [Catalog.identifier_to_tuple(namespace_col) for namespace_col in session.execute(stmt).scalars()]
+            namespace_tuple = Catalog.identifier_to_tuple(namespace)
+            sub_namespaces_level_length = len(namespace_tuple) + 1
+
+            namespaces = list(
+                {  # only get distinct namespaces
+                    ns[:sub_namespaces_level_length]  # truncate to the required level
+                    for ns in {Catalog.identifier_to_tuple(ns) for ns in session.execute(stmt).scalars()}
+                    if len(ns) >= sub_namespaces_level_length  # only get sub namespaces/children
+                    and ns[: sub_namespaces_level_length - 1] == namespace_tuple
+                    # exclude fuzzy matches when `namespace` contains `%` or `_`
+                }
+            )
+
+            return namespaces
 
     def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Properties:
         """Get properties for a namespace.