diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py
index 53e7a2e1e3..88ca051015 100644
--- a/pyiceberg/manifest.py
+++ b/pyiceberg/manifest.py
@@ -471,9 +471,14 @@ def data_file_with_partition(partition_type: StructType, format_version: TableVe
 
 class DataFile(Record):
     @classmethod
-    def from_args(cls, _table_format_version: TableVersion = DEFAULT_READ_VERSION, **arguments: Any) -> DataFile:
+    def from_args(
+        cls, _table_format_version: TableVersion = DEFAULT_READ_VERSION, *, spec_id: int | None = None, **arguments: Any
+    ) -> DataFile:
         struct = DATA_FILE_TYPE[_table_format_version]
-        return super()._bind(struct, **arguments)
+        data_file = super()._bind(struct, **arguments)
+        if spec_id is not None:
+            data_file.spec_id = spec_id
+        return data_file
 
     @property
     def content(self) -> DataFileContent:
diff --git a/pyiceberg/typedef.py b/pyiceberg/typedef.py
index 6989144ef9..489a44d5f7 100644
--- a/pyiceberg/typedef.py
+++ b/pyiceberg/typedef.py
@@ -179,6 +179,9 @@ class Record(StructProtocol):
 
     @classmethod
     def _bind(cls, struct: StructType, **arguments: Any) -> Self:
+        field_names = {field.name for field in struct.fields}
+        if unknown_fields := arguments.keys() - field_names:
+            raise TypeError(f"Unexpected {cls.__name__} fields: {', '.join(sorted(unknown_fields))}")
         return cls(*[arguments[field.name] if field.name in arguments else field.initial_default for field in struct.fields])
 
     def __init__(self, *data: Any) -> None:
diff --git a/tests/avro/test_file.py b/tests/avro/test_file.py
index 28c7436eac..2d3ddeefab 100644
--- a/tests/avro/test_file.py
+++ b/tests/avro/test_file.py
@@ -266,8 +266,7 @@ def test_write_v2_referenced_data_file_with_fastavro() -> None:
 
 @pytest.mark.parametrize("format_version", [1, 2])
 def test_write_manifest_entry_with_fastavro_read_with_iceberg(format_version: TableVersion) -> None:
-    data_file_dict = {
-        "content": DataFileContent.DATA,
+    common_data_file_args = {
         "file_path": "s3://some-path/some-file.parquet",
         "file_format": FileFormat.PARQUET,
         "partition": Record(),
@@ -281,16 +280,16 @@ def test_write_manifest_entry_with_fastavro_read_with_iceberg(format_version: Ta
         "upper_bounds": {1: b"zzzzzzzzzzzzzzzz"},
         "key_metadata": b"\xde\xad\xbe\xef",
         "split_offsets": [4, 133697593],
-        "equality_ids": [],
         "sort_order_id": 4,
         "spec_id": 3,
     }
-    data_file_v2 = DataFile.from_args(**data_file_dict)  # type: ignore
+    data_file = DataFile.from_args(content=DataFileContent.DATA, **common_data_file_args)  # type: ignore
+    assert data_file.spec_id == 3
 
     entry = ManifestEntry.from_args(
         status=ManifestEntryStatus.ADDED,
         snapshot_id=8638475580105682862,
-        data_file=data_file_v2,
+        data_file=data_file,
     )
 
     with TemporaryDirectory() as tmpdir:
@@ -322,7 +321,11 @@ def test_write_manifest_entry_with_fastavro_read_with_iceberg(format_version: Ta
             avro_entry = next(it)
 
             if format_version == 1:
-                data_file_v1 = DataFile.from_args(**data_file_dict, _table_format_version=format_version)
+                data_file_v1 = DataFile.from_args(
+                    _table_format_version=format_version,
+                    block_size_in_bytes=DEFAULT_BLOCK_SIZE,
+                    **common_data_file_args,  # type: ignore
+                )
 
                 assert avro_entry == ManifestEntry.from_args(
                     status=1,
diff --git a/tests/integration/test_rest_manifest.py b/tests/integration/test_rest_manifest.py
index 21832116b5..849d1ac2b9 100644
--- a/tests/integration/test_rest_manifest.py
+++ b/tests/integration/test_rest_manifest.py
@@ -90,7 +90,6 @@ def test_write_sample_manifest(table_test_all_types: Table, compression: AvroCom
     test_schema = table_test_all_types.schema()
     test_spec = table_test_all_types.spec()
     wrapped_data_file_v2_debug = DataFile.from_args(
-        format_version=2,
         content=entry.data_file.content,
         file_path=entry.data_file.file_path,
         file_format=entry.data_file.file_format,
@@ -112,7 +111,7 @@ def test_write_sample_manifest(table_test_all_types: Table, compression: AvroCom
     wrapped_entry_v2 = copy(entry)
     wrapped_entry_v2.data_file = wrapped_data_file_v2_debug
     wrapped_entry_v2_dict = todict(wrapped_entry_v2, [field.name for field in test_spec.fields])
-    for field in ("first_row_id", "content_offset", "content_size_in_bytes"):
+    for field in ("first_row_id", "content_offset", "content_size_in_bytes", "spec_id"):
         del wrapped_entry_v2_dict["data_file"][field]
 
     with TemporaryDirectory() as tmpdir:
diff --git a/tests/test_typedef.py b/tests/test_typedef.py
index fbbb619968..f13349a42f 100644
--- a/tests/test_typedef.py
+++ b/tests/test_typedef.py
@@ -17,6 +17,7 @@
 import pytest
 
 from pyiceberg.typedef import FrozenDict, KeyDefaultDict, Record
+from pyiceberg.types import IntegerType, NestedField, StructType
 
 
 def test_setitem_frozendict() -> None:
@@ -47,3 +48,20 @@ def test_record_named_args() -> None:
     assert r[2] is True
 
     assert repr(r) == "Record[1, a, True]"
+
+
+def test_record_bind_rejects_unknown_arguments() -> None:
+    struct = StructType(NestedField(1, "known", IntegerType()))
+
+    with pytest.raises(TypeError, match="Unexpected Record fields: unknown"):
+        Record._bind(struct, known=1, unknown=2)
+
+
+def test_record_bind_rejects_non_schema_property() -> None:
+    class RecordWithProperty(Record):
+        @property
+        def computed(self) -> int:
+            return 1
+
+    with pytest.raises(TypeError, match="Unexpected RecordWithProperty fields: computed"):
+        RecordWithProperty._bind(StructType(), computed=1)