Make records purely position based

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.

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

@@ -286,7 +286,15 @@ 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",
+        "_create_with_keyword",
+        "_field_reader_functions",
+        "_hash",
+        "_max_pos",
+    )
     field_readers: Tuple[Tuple[Optional[int], Reader], ...]
     create_struct: Callable[..., StructProtocol]
     struct: StructType
@@ -300,34 +308,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

@@ -2225,7 +2225,7 @@ def _partition_value(self, partition_field: PartitionField, schema: Schema) -> A
         return transform(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 = {}
@@ -2398,7 +2398,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,
@@ -2489,7 +2489,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,