Merge scan options into FileIO at TableScan.__init__ time

pyiceberg/table/__init__.py

@@ -1713,7 +1713,7 @@ def __init__(
         table_identifier: Identifier | None = None,
     ):
         self.table_metadata = table_metadata
-        self.io = io
+        self.io = load_file_io({**io.properties, **options}, self.table_metadata.location) if options else io
         self.row_filter = _parse_row_filter(row_filter)
         self.selected_fields = selected_fields
         self.case_sensitive = case_sensitive
@@ -1961,13 +1961,11 @@ def _build_residual_evaluator(self, spec_id: int) -> Callable[[DataFile], Residu
         # The lambda created here is run in multiple threads.
         # So we avoid creating _EvaluatorExpression methods bound to a single
         # shared instance across multiple threads.
-        return lambda datafile: (
-            residual_evaluator_of(
-                spec=spec,
-                expr=self.row_filter,
-                case_sensitive=self.case_sensitive,
-                schema=self.table_metadata.schema(),
-            )
+        return lambda datafile: residual_evaluator_of(
+            spec=spec,
+            expr=self.row_filter,
+            case_sensitive=self.case_sensitive,
+            schema=self.table_metadata.schema(),
         )
 
     @staticmethod

tests/table/test_init.py

@@ -336,6 +336,25 @@ def test_table_scan_projection_unknown_column(table_v2: Table) -> None:
     assert "Could not find column: 'a'" in str(exc_info.value)
 
 
+def test_table_scan_options_merged_with_existing_io_properties(table_v2: Table) -> None:
+    table = Table(
+        identifier=table_v2._identifier,
+        metadata=table_v2.metadata,
+        metadata_location=table_v2.metadata_location,
+        io=load_file_io({"s3.region": "us-east-1", "s3.connect-timeout": "10"}),
+        catalog=table_v2.catalog,
+    )
+    scan = table.scan(options={"s3.connect-timeout": "60", "s3.endpoint": "https://custom.endpoint"})
+    assert scan.io.properties["s3.region"] == "us-east-1"
+    assert scan.io.properties["s3.connect-timeout"] == "60"
+    assert scan.io.properties["s3.endpoint"] == "https://custom.endpoint"
+
+
+def test_table_scan_empty_options_reuses_file_io(table_v2: Table) -> None:
+    scan = table_v2.scan()
+    assert scan.io is table_v2.io
+
+
 def test_static_table_same_as_table(table_v2: Table, metadata_location: str) -> None:
     static_table = StaticTable.from_metadata(metadata_location)
     assert isinstance(static_table, Table)