Make Record purely position based (#1768)

This aligns the implementation with Java.

We had the keywords there mostly for the tests, but they should not be
used, and it seems like that's already the case :'( I was undecided if
the costs of this PR (all the changes), are worth it, but I see more PRs
using the Record in a bad way (example
#1743) that might lead to
very subtle bugs where the position might sometime change based on the
ordering of the dict.

Blocked by Eventual-Inc/Daft#3917

Author: Fokko

pyiceberg/avro/file.py

@@ -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/reader.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/io/pyarrow.py

@@ -2271,7 +2271,7 @@ def _partition_value(self, partition_field: PartitionField, schema: Schema) -> A
         return lower_value
 
     def partition(self, partition_spec: PartitionSpec, schema: Schema) -> Record:
-        return Record(**{field.name: self._partition_value(field, schema) for field in partition_spec.fields})
+        return Record(*[self._partition_value(field, schema) for field in partition_spec.fields])
 
     def to_serialized_dict(self) -> Dict[str, Any]:
         lower_bounds = {}
@@ -2449,7 +2449,7 @@ def write_parquet(task: WriteTask) -> DataFile:
             stats_columns=compute_statistics_plan(file_schema, table_metadata.properties),
             parquet_column_mapping=parquet_path_to_id_mapping(file_schema),
         )
-        data_file = DataFile(
+        data_file = DataFile.from_args(
             content=DataFileContent.DATA,
             file_path=file_path,
             file_format=FileFormat.PARQUET,
@@ -2540,7 +2540,7 @@ def parquet_file_to_data_file(io: FileIO, table_metadata: TableMetadata, file_pa
         stats_columns=compute_statistics_plan(schema, table_metadata.properties),
         parquet_column_mapping=parquet_path_to_id_mapping(schema),
     )
-    data_file = DataFile(
+    data_file = DataFile.from_args(
         content=DataFileContent.DATA,
         file_path=file_path,
         file_format=FileFormat.PARQUET,