fix: rank catalog configuration above table properties when constructing FileIO (#3956)

* Rank catalog configuration above table properties when constructing FileIO

* Cover FileIO property precedence in StaticTable.from_metadata

Author: YuangGao

pyiceberg/catalog/__init__.py

@@ -833,8 +833,15 @@ def identifier_to_database_and_table(
 
         return tuple_identifier[0], tuple_identifier[1]
 
-    def _load_file_io(self, properties: Properties = EMPTY_DICT, location: str | None = None) -> FileIO:
-        return load_file_io({**self.properties, **properties}, location)
+    def _load_file_io(
+        self,
+        properties: Properties = EMPTY_DICT,
+        location: str | None = None,
+        table_properties: Properties = EMPTY_DICT,
+    ) -> FileIO:
+        # table_properties ranks lowest: a principal who can commit to a table must not
+        # be able to redirect its readers.
+        return load_file_io({**table_properties, **self.properties, **properties}, location)
 
     @staticmethod
     def _convert_schema_if_needed(
@@ -1020,7 +1027,7 @@ def _create_staged_table(
         metadata = new_table_metadata(
             location=location, schema=schema, partition_spec=partition_spec, sort_order=sort_order, properties=properties
         )
-        io = self._load_file_io(properties=properties, location=metadata_location)
+        io = self._load_file_io(location=metadata_location, table_properties=properties)
         return StagedTable(
             identifier=(database_name, table_name),
             metadata=metadata,
@@ -1054,7 +1061,7 @@ def _update_and_stage_table(
             identifier=table_identifier,
             metadata=updated_metadata,
             metadata_location=new_metadata_location,
-            io=self._load_file_io(properties=updated_metadata.properties, location=new_metadata_location),
+            io=self._load_file_io(location=new_metadata_location, table_properties=updated_metadata.properties),
             catalog=self,
         )
 

pyiceberg/catalog/bigquery_metastore.py

@@ -404,7 +404,7 @@ def _convert_bigquery_table_to_iceberg_table(self, identifier: str | Identifier,
             identifier=(dataset_name, table_name),
             metadata=metadata,
             metadata_location=metadata_location,
-            io=self._load_file_io(metadata.properties, metadata_location),
+            io=self._load_file_io(location=metadata_location, table_properties=metadata.properties),
             catalog=self,
         )
 

pyiceberg/catalog/dynamodb.py

@@ -697,7 +697,7 @@ def _convert_dynamo_table_item_to_iceberg_table(self, dynamo_table_item: dict[st
             identifier=(database_name, table_name),
             metadata=metadata,
             metadata_location=metadata_location,
-            io=self._load_file_io(metadata.properties, metadata_location),
+            io=self._load_file_io(location=metadata_location, table_properties=metadata.properties),
             catalog=self,
         )
 

pyiceberg/catalog/glue.py

@@ -389,7 +389,7 @@ def _convert_glue_to_iceberg(self, glue_table: "TableTypeDef") -> Table:
             identifier=(database_name, table_name),
             metadata=metadata,
             metadata_location=metadata_location,
-            io=self._load_file_io(metadata.properties, metadata_location),
+            io=self._load_file_io(location=metadata_location, table_properties=metadata.properties),
             catalog=self,
         )
 
@@ -535,7 +535,7 @@ def _create_table_s3tables(
             identifier=self.identifier_to_tuple(identifier),
             metadata=staged_table.metadata,
             metadata_location=staged_table.metadata_location,
-            io=self._load_file_io(staged_table.metadata.properties, staged_table.metadata_location),
+            io=self._load_file_io(location=staged_table.metadata_location, table_properties=staged_table.metadata.properties),
             catalog=self,
         )
 
@@ -599,7 +599,7 @@ def create_table(
             identifier=self.identifier_to_tuple(identifier),
             metadata=staged_table.metadata,
             metadata_location=staged_table.metadata_location,
-            io=self._load_file_io(staged_table.metadata.properties, staged_table.metadata_location),
+            io=self._load_file_io(location=staged_table.metadata_location, table_properties=staged_table.metadata.properties),
             catalog=self,
         )
 

pyiceberg/catalog/hive.py

@@ -353,7 +353,7 @@ def _convert_hive_into_iceberg(self, table: HiveTable) -> Table:
             identifier=(table.dbName, table.tableName),
             metadata=metadata,
             metadata_location=metadata_location,
-            io=self._load_file_io(metadata.properties, metadata_location),
+            io=self._load_file_io(location=metadata_location, table_properties=metadata.properties),
             catalog=self,
         )
 

pyiceberg/catalog/rest/__init__.py

@@ -638,8 +638,13 @@ def _resolve_storage_credentials(storage_credentials: list[StorageCredential], l
 
         return best_match.config if best_match else {}
 
-    def _load_file_io(self, properties: Properties = EMPTY_DICT, location: str | None = None) -> FileIO:
-        merged_properties = {**self.properties, **properties}
+    def _load_file_io(
+        self,
+        properties: Properties = EMPTY_DICT,
+        location: str | None = None,
+        table_properties: Properties = EMPTY_DICT,
+    ) -> FileIO:
+        merged_properties = {**table_properties, **self.properties, **properties}
         if self._auth_manager:
             merged_properties[AUTH_MANAGER] = self._auth_manager
         return load_file_io(merged_properties, location)
@@ -1152,8 +1157,9 @@ def _response_to_table(self, identifier_tuple: tuple[str, ...], table_response:
             metadata_location=table_response.metadata_location,  # type: ignore
             metadata=table_response.metadata,
             io=self._load_file_io(
-                {**table_response.metadata.properties, **table_response.config, **credential_config},
+                {**table_response.config, **credential_config},
                 table_response.metadata_location,
+                table_properties=table_response.metadata.properties,
             ),
             catalog=self,
             config=table_response.config,
@@ -1169,8 +1175,9 @@ def _response_to_staged_table(self, identifier_tuple: tuple[str, ...], table_res
             metadata_location=table_response.metadata_location,  # type: ignore
             metadata=table_response.metadata,
             io=self._load_file_io(
-                {**table_response.metadata.properties, **table_response.config, **credential_config},
+                {**table_response.config, **credential_config},
                 table_response.metadata_location,
+                table_properties=table_response.metadata.properties,
             ),
             catalog=self,
         )

pyiceberg/catalog/sql.py

@@ -237,7 +237,7 @@ def _convert_orm_to_iceberg(self, orm_table: IcebergTables) -> Table:
             identifier=Catalog.identifier_to_tuple(table_namespace) + (table_name,),
             metadata=metadata,
             metadata_location=metadata_location,
-            io=self._load_file_io(metadata.properties, metadata_location),
+            io=self._load_file_io(location=metadata_location, table_properties=metadata.properties),
             catalog=self,
         )
 

pyiceberg/table/__init__.py

@@ -1995,7 +1995,7 @@ def from_metadata(cls, metadata_location: str, properties: Properties = EMPTY_DI
             identifier=("static-table", metadata_location),
             metadata_location=metadata_location,
             metadata=metadata,
-            io=load_file_io({**properties, **metadata.properties}, location=metadata_location),
+            io=load_file_io({**metadata.properties, **properties}, location=metadata_location),
             catalog=NoopCatalog("static-table"),
         )
 

tests/catalog/test_rest.py

@@ -65,7 +65,7 @@
     TableAlreadyExistsError,
     ViewAlreadyExistsError,
 )
-from pyiceberg.io import load_file_io
+from pyiceberg.io import ARROW_FILE_IO, FSSPEC_FILE_IO, PY_IO_IMPL, load_file_io
 from pyiceberg.partitioning import PartitionField, PartitionSpec
 from pyiceberg.schema import Schema
 from pyiceberg.table import Table
@@ -3415,6 +3415,32 @@ def test_load_table_with_storage_credentials(rest_mock: Mocker, example_table_me
     assert table.io.properties["s3.session-token"] == "vended-token"
 
 
+def test_load_table_catalog_config_outranks_table_properties(
+    rest_mock: Mocker, example_table_metadata_with_snapshot_v1: dict[str, Any]
+) -> None:
+    metadata_location = "s3://warehouse/database/table/metadata/00001.metadata.json"
+    rest_mock.get(
+        f"{TEST_URI}v1/namespaces/fokko/tables/table",
+        json={
+            "metadata-location": metadata_location,
+            "metadata": {
+                **example_table_metadata_with_snapshot_v1,
+                "properties": {PY_IO_IMPL: FSSPEC_FILE_IO, "s3.proxy-uri": "http://table-only-proxy"},
+            },
+            "config": {"s3.region": "from-config"},
+        },
+        status_code=200,
+        request_headers=TEST_HEADERS,
+    )
+    catalog = RestCatalog("rest", uri=TEST_URI, token=TEST_TOKEN, **{PY_IO_IMPL: ARROW_FILE_IO, "s3.region": "from-catalog"})
+    table = catalog.load_table(("fokko", "table"))
+
+    assert table.io.properties[PY_IO_IMPL] == ARROW_FILE_IO
+    # Server config and a key the catalog leaves unset keep working.
+    assert table.io.properties["s3.region"] == "from-config"
+    assert table.io.properties["s3.proxy-uri"] == "http://table-only-proxy"
+
+
 def test_load_credentials_with_longest_prefix(rest_mock: Mocker) -> None:
     rest_mock.get(
         f"{TEST_URI}v1/namespaces/fokko/tables/table/credentials",

tests/catalog/test_sql.py

@@ -36,6 +36,7 @@
     NoSuchTableError,
     TableAlreadyExistsError,
 )
+from pyiceberg.io import ARROW_FILE_IO, FSSPEC_FILE_IO, PY_IO_IMPL
 from pyiceberg.schema import Schema
 from pyiceberg.types import NestedField, StringType, strtobool
 
@@ -299,6 +300,25 @@ def test_idempotent_when_column_already_exists(warehouse: Path) -> None:
     assert "iceberg_type" in get_columns(catalog.engine)
 
 
+def test_load_table_ranks_catalog_config_above_table_properties(warehouse: Path) -> None:
+    catalog = SqlCatalog(
+        name="test",
+        uri="sqlite:///:memory:",
+        warehouse=f"file://{warehouse}",
+        **{PY_IO_IMPL: ARROW_FILE_IO},
+    )
+    catalog.create_namespace("ns")
+    catalog.create_table(
+        ("ns", "tbl"),
+        Schema(NestedField(1, "id", StringType(), required=True)),
+        properties={PY_IO_IMPL: FSSPEC_FILE_IO, "s3.proxy-uri": "http://table-only-proxy"},
+    )
+
+    io = catalog.load_table(("ns", "tbl")).io
+    assert io.properties[PY_IO_IMPL] == ARROW_FILE_IO
+    assert io.properties["s3.proxy-uri"] == "http://table-only-proxy"
+
+
 def test_list_tables_filters_by_iceberg_type(warehouse: Path) -> None:
     catalog = SqlCatalog(
         name="test",