Merge branch 'main' into feat/validation-history

Author: jayceslesar

.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

@@ -303,32 +303,46 @@ def add_glue_catalog_id(params: Dict[str, str], **kwargs: Any) -> None:
 
 
 class GlueCatalog(MetastoreCatalog):
-    def __init__(self, name: str, **properties: Any):
-        super().__init__(name, **properties)
+    glue: GlueClient
 
-        retry_mode_prop_value = get_first_property_value(properties, GLUE_RETRY_MODE)
+    def __init__(self, name: str, client: Optional[GlueClient] = None, **properties: Any):
+        """Glue Catalog.
 
-        session = boto3.Session(
-            profile_name=properties.get(GLUE_PROFILE_NAME),
-            region_name=get_first_property_value(properties, GLUE_REGION, AWS_REGION),
-            botocore_session=properties.get(BOTOCORE_SESSION),
-            aws_access_key_id=get_first_property_value(properties, GLUE_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID),
-            aws_secret_access_key=get_first_property_value(properties, GLUE_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY),
-            aws_session_token=get_first_property_value(properties, GLUE_SESSION_TOKEN, AWS_SESSION_TOKEN),
-        )
-        self.glue: GlueClient = session.client(
-            "glue",
-            endpoint_url=properties.get(GLUE_CATALOG_ENDPOINT),
-            config=Config(
-                retries={
-                    "max_attempts": properties.get(GLUE_MAX_RETRIES, MAX_RETRIES),
-                    "mode": retry_mode_prop_value if retry_mode_prop_value in EXISTING_RETRY_MODES else STANDARD_RETRY_MODE,
-                }
-            ),
-        )
+        You either need to provide a boto3 glue client, or one will be constructed from the properties.
+
+        Args:
+            name: Name to identify the catalog.
+            client: An optional boto3 glue client.
+            properties: Properties for glue client construction and configuration.
+        """
+        super().__init__(name, **properties)
+
+        if client:
+            self.glue = client
+        else:
+            retry_mode_prop_value = get_first_property_value(properties, GLUE_RETRY_MODE)
+
+            session = boto3.Session(
+                profile_name=properties.get(GLUE_PROFILE_NAME),
+                region_name=get_first_property_value(properties, GLUE_REGION, AWS_REGION),
+                botocore_session=properties.get(BOTOCORE_SESSION),
+                aws_access_key_id=get_first_property_value(properties, GLUE_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID),
+                aws_secret_access_key=get_first_property_value(properties, GLUE_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY),
+                aws_session_token=get_first_property_value(properties, GLUE_SESSION_TOKEN, AWS_SESSION_TOKEN),
+            )
+            self.glue: GlueClient = session.client(
+                "glue",
+                endpoint_url=properties.get(GLUE_CATALOG_ENDPOINT),
+                config=Config(
+                    retries={
+                        "max_attempts": properties.get(GLUE_MAX_RETRIES, MAX_RETRIES),
+                        "mode": retry_mode_prop_value if retry_mode_prop_value in EXISTING_RETRY_MODES else STANDARD_RETRY_MODE,
+                    }
+                ),
+            )
 
-        if glue_catalog_id := properties.get(GLUE_ID):
-            _register_glue_catalog_id_with_glue_client(self.glue, glue_catalog_id)
+            if glue_catalog_id := properties.get(GLUE_ID):
+                _register_glue_catalog_id_with_glue_client(self.glue, glue_catalog_id)
 
     def _convert_glue_to_iceberg(self, glue_table: TableTypeDef) -> Table:
         properties: Properties = glue_table["Parameters"]