Merge branch 'main' of github.com:apache/iceberg-python into fd-add-deletion-vectors

Author: Fokko

.github/dependabot.yml

@@ -22,9 +22,9 @@ updates:
   - package-ecosystem: "pip"
     directory: "/"
     schedule:
-      interval: "daily"
+      interval: "weekly"
     open-pull-requests-limit: 50
   - package-ecosystem: "github-actions"
     directory: "/"
     schedule:
-      interval: "daily"
+      interval: "weekly"

.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

@@ -339,6 +339,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
@@ -351,8 +352,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

@@ -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()

pyiceberg/avro/reader.py

@@ -44,7 +44,9 @@
     StringReader,
     StructReader,
     TimeReader,
+    TimestampNanoReader,
     TimestampReader,
+    TimestamptzNanoReader,
     TimestamptzReader,
     UnknownReader,
     UUIDReader,
@@ -64,6 +66,8 @@
     OptionWriter,
     StringWriter,
     StructWriter,
+    TimestampNanoWriter,
+    TimestamptzNanoWriter,
     TimestamptzWriter,
     TimestampWriter,
     TimeWriter,
@@ -99,7 +103,9 @@
     PrimitiveType,
     StringType,
     StructType,
+    TimestampNanoType,
     TimestampType,
+    TimestamptzNanoType,
     TimestamptzType,
     TimeType,
     UnknownType,
@@ -184,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()
 
@@ -332,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()
 
@@ -465,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()
 

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:

pyiceberg/avro/writer.py

@@ -370,7 +370,7 @@ def _fetch_access_token(self, session: Session, credential: str) -> str:
         except HTTPError as exc:
             self._handle_non_200_response(exc, {400: OAuthError, 401: OAuthError})
 
-        return TokenResponse(**response.json()).access_token
+        return TokenResponse.model_validate_json(response.text).access_token
 
     def _fetch_config(self) -> None:
         params = {}
@@ -383,7 +383,7 @@ def _fetch_config(self) -> None:
             response.raise_for_status()
         except HTTPError as exc:
             self._handle_non_200_response(exc, {})
-        config_response = ConfigResponse(**response.json())
+        config_response = ConfigResponse.model_validate_json(response.text)
 
         config = config_response.defaults
         config.update(self.properties)
@@ -443,14 +443,14 @@ def _handle_non_200_response(self, exc: HTTPError, error_handler: Dict[int, Type
         try:
             if exception == OAuthError:
                 # The OAuthErrorResponse has a different format
-                error = OAuthErrorResponse(**exc.response.json())
+                error = OAuthErrorResponse.model_validate_json(exc.response.text)
                 response = str(error.error)
                 if description := error.error_description:
                     response += f": {description}"
                 if uri := error.error_uri:
                     response += f" ({uri})"
             else:
-                error = ErrorResponse(**exc.response.json()).error
+                error = ErrorResponse.model_validate_json(exc.response.text).error
                 response = f"{error.type}: {error.message}"
         except JSONDecodeError:
             # In the case we don't have a proper response
@@ -588,7 +588,7 @@ def _create_table(
             response.raise_for_status()
         except HTTPError as exc:
             self._handle_non_200_response(exc, {409: TableAlreadyExistsError})
-        return TableResponse(**response.json())
+        return TableResponse.model_validate_json(response.text)
 
     @retry(**_RETRY_ARGS)
     def create_table(
@@ -662,7 +662,7 @@ def register_table(self, identifier: Union[str, Identifier], metadata_location:
         except HTTPError as exc:
             self._handle_non_200_response(exc, {409: TableAlreadyExistsError})
 
-        table_response = TableResponse(**response.json())
+        table_response = TableResponse.model_validate_json(response.text)
         return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
 
     @retry(**_RETRY_ARGS)
@@ -674,7 +674,7 @@ def list_tables(self, namespace: Union[str, Identifier]) -> List[Identifier]:
             response.raise_for_status()
         except HTTPError as exc:
             self._handle_non_200_response(exc, {404: NoSuchNamespaceError})
-        return [(*table.namespace, table.name) for table in ListTablesResponse(**response.json()).identifiers]
+        return [(*table.namespace, table.name) for table in ListTablesResponse.model_validate_json(response.text).identifiers]
 
     @retry(**_RETRY_ARGS)
     def load_table(self, identifier: Union[str, Identifier]) -> Table:
@@ -684,7 +684,7 @@ def load_table(self, identifier: Union[str, Identifier]) -> Table:
         except HTTPError as exc:
             self._handle_non_200_response(exc, {404: NoSuchTableError})
 
-        table_response = TableResponse(**response.json())
+        table_response = TableResponse.model_validate_json(response.text)
         return self._response_to_table(self.identifier_to_tuple(identifier), table_response)
 
     @retry(**_RETRY_ARGS)
@@ -735,7 +735,7 @@ def list_views(self, namespace: Union[str, Identifier]) -> List[Identifier]:
             response.raise_for_status()
         except HTTPError as exc:
             self._handle_non_200_response(exc, {404: NoSuchNamespaceError})
-        return [(*view.namespace, view.name) for view in ListViewsResponse(**response.json()).identifiers]
+        return [(*view.namespace, view.name) for view in ListViewsResponse.model_validate_json(response.text).identifiers]
 
     @retry(**_RETRY_ARGS)
     def commit_table(
@@ -781,7 +781,7 @@ def commit_table(
                     504: CommitStateUnknownException,
                 },
             )
-        return CommitTableResponse(**response.json())
+        return CommitTableResponse.model_validate_json(response.text)
 
     @retry(**_RETRY_ARGS)
     def create_namespace(self, namespace: Union[str, Identifier], properties: Properties = EMPTY_DICT) -> None:
@@ -818,7 +818,7 @@ def list_namespaces(self, namespace: Union[str, Identifier] = ()) -> List[Identi
         except HTTPError as exc:
             self._handle_non_200_response(exc, {})
 
-        return ListNamespaceResponse(**response.json()).namespaces
+        return ListNamespaceResponse.model_validate_json(response.text).namespaces
 
     @retry(**_RETRY_ARGS)
     def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Properties:
@@ -830,7 +830,7 @@ def load_namespace_properties(self, namespace: Union[str, Identifier]) -> Proper
         except HTTPError as exc:
             self._handle_non_200_response(exc, {404: NoSuchNamespaceError})
 
-        return NamespaceResponse(**response.json()).properties
+        return NamespaceResponse.model_validate_json(response.text).properties
 
     @retry(**_RETRY_ARGS)
     def update_namespace_properties(
@@ -844,7 +844,7 @@ def update_namespace_properties(
             response.raise_for_status()
         except HTTPError as exc:
             self._handle_non_200_response(exc, {404: NoSuchNamespaceError})
-        parsed_response = UpdateNamespacePropertiesResponse(**response.json())
+        parsed_response = UpdateNamespacePropertiesResponse.model_validate_json(response.text)
         return PropertiesUpdateSummary(
             removed=parsed_response.removed,
             updated=parsed_response.updated,