Merge branch 'main' into 1880-add-expire-snapshots

Author: ForeverAngry

.github/workflows/python-ci.yml

@@ -58,6 +58,8 @@ jobs:
         python-version: ${{ matrix.python }}
         cache: poetry
         cache-dependency-path: ./poetry.lock
+    - name: Install system dependencies
+      run: sudo apt-get update && sudo apt-get install -y libkrb5-dev # for kerberos
     - name: Install
       run: make install-dependencies
     - name: Linters

.github/workflows/python-integration.yml

@@ -50,6 +50,8 @@ jobs:
     - uses: actions/checkout@v4
       with:
         fetch-depth: 2
+    - name: Install system dependencies
+      run: sudo apt-get update && sudo apt-get install -y libkrb5-dev # for kerberos
     - name: Install
       run: make install
     - name: Run integration tests

Makefile

@@ -19,7 +19,7 @@
 help:  ## Display this help
 	@awk 'BEGIN {FS = ":.*##"; printf "\nUsage:\n  make \033[36m\033[0m\n"} /^[a-zA-Z_-]+:.*?##/ { printf "  \033[36m%-20s\033[0m %s\n", $$1, $$2 } /^##@/ { printf "\n\033[1m%s\033[0m\n", substr($$0, 5) } ' $(MAKEFILE_LIST)
 
-POETRY_VERSION = 2.0.1
+POETRY_VERSION = 2.1.1
 install-poetry:  ## Ensure Poetry is installed and the correct version is being used.
 	@if ! command -v poetry &> /dev/null; then \
 		echo "Poetry could not be found. Installing..."; \

poetry.lock

@@ -74,10 +74,17 @@
 
 
 class AvroFileHeader(Record):
-    __slots__ = ("magic", "meta", "sync")
-    magic: bytes
-    meta: Dict[str, str]
-    sync: bytes
+    @property
+    def magic(self) -> bytes:
+        return self._data[0]
+
+    @property
+    def meta(self) -> Dict[str, str]:
+        return self._data[1]
+
+    @property
+    def sync(self) -> bytes:
+        return self._data[2]
 
     def compression_codec(self) -> Optional[Type[Codec]]:
         """Get the file's compression codec algorithm from the file's metadata.
@@ -271,7 +278,7 @@ def __exit__(
     def _write_header(self) -> None:
         json_schema = json.dumps(AvroSchemaConversion().iceberg_to_avro(self.file_schema, schema_name=self.schema_name))
         meta = {**self.metadata, _SCHEMA_KEY: json_schema, _CODEC_KEY: "null"}
-        header = AvroFileHeader(magic=MAGIC, meta=meta, sync=self.sync_bytes)
+        header = AvroFileHeader(MAGIC, meta, self.sync_bytes)
         construct_writer(META_SCHEMA).write(self.encoder, header)
 
     def write_block(self, objects: List[D]) -> None:

pyiceberg/avro/file.py

@@ -312,7 +312,14 @@ def skip(self, decoder: BinaryDecoder) -> None:
 
 
 class StructReader(Reader):
-    __slots__ = ("field_readers", "create_struct", "struct", "_create_with_keyword", "_field_reader_functions", "_hash")
+    __slots__ = (
+        "field_readers",
+        "create_struct",
+        "struct",
+        "_field_reader_functions",
+        "_hash",
+        "_max_pos",
+    )
     field_readers: Tuple[Tuple[Optional[int], Reader], ...]
     create_struct: Callable[..., StructProtocol]
     struct: StructType
@@ -326,34 +333,28 @@ def __init__(
     ) -> None:
         self.field_readers = field_readers
         self.create_struct = create_struct
+        # TODO: Implement struct-reuse
         self.struct = struct
 
-        try:
-            # Try initializing the struct, first with the struct keyword argument
-            created_struct = self.create_struct(struct=self.struct)
-            self._create_with_keyword = True
-        except TypeError as e:
-            if "'struct' is an invalid keyword argument for" in str(e):
-                created_struct = self.create_struct()
-                self._create_with_keyword = False
-            else:
-                raise ValueError(f"Unable to initialize struct: {self.create_struct}") from e
-
-        if not isinstance(created_struct, StructProtocol):
+        if not isinstance(self.create_struct(), StructProtocol):
             raise ValueError(f"Incompatible with StructProtocol: {self.create_struct}")
 
         reading_callbacks: List[Tuple[Optional[int], Callable[[BinaryDecoder], Any]]] = []
+        max_pos = -1
         for pos, field in field_readers:
             if pos is not None:
                 reading_callbacks.append((pos, field.read))
+                max_pos = max(max_pos, pos)
             else:
                 reading_callbacks.append((None, field.skip))
 
         self._field_reader_functions = tuple(reading_callbacks)
         self._hash = hash(self._field_reader_functions)
+        self._max_pos = 1 + max_pos
 
     def read(self, decoder: BinaryDecoder) -> StructProtocol:
-        struct = self.create_struct(struct=self.struct) if self._create_with_keyword else self.create_struct()
+        # TODO: Implement struct-reuse
+        struct = self.create_struct(*[None] * self._max_pos)
         for pos, field_reader in self._field_reader_functions:
             if pos is not None:
                 struct[pos] = field_reader(decoder)  # later: pass reuse in here

pyiceberg/avro/reader.py

@@ -290,7 +290,7 @@ def struct(self, file_schema: StructType, record_struct: Optional[IcebergType],
                 # There is a default value
                 if file_field.write_default is not None:
                     # The field is not in the record, but there is a write default value
-                    results.append((None, DefaultWriter(writer=writer, value=file_field.write_default)))  # type: ignore
+                    results.append((None, DefaultWriter(writer=writer, value=file_field.write_default)))
                 elif file_field.required:
                     raise ValueError(f"Field is required, and there is no write default: {file_field}")
             else: