diff --git a/pyiceberg/catalog/__init__.py b/pyiceberg/catalog/__init__.py index 8de113404c..0842f51cc8 100644 --- a/pyiceberg/catalog/__init__.py +++ b/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, ) diff --git a/pyiceberg/catalog/bigquery_metastore.py b/pyiceberg/catalog/bigquery_metastore.py index cc84b0b420..1267233e94 100644 --- a/pyiceberg/catalog/bigquery_metastore.py +++ b/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, ) diff --git a/pyiceberg/catalog/dynamodb.py b/pyiceberg/catalog/dynamodb.py index 10f74fb9a0..b4111a6fbb 100644 --- a/pyiceberg/catalog/dynamodb.py +++ b/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, ) diff --git a/pyiceberg/catalog/glue.py b/pyiceberg/catalog/glue.py index 12b36efc5c..977876918c 100644 --- a/pyiceberg/catalog/glue.py +++ b/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, ) diff --git a/pyiceberg/catalog/hive.py b/pyiceberg/catalog/hive.py index ed6d3c6fd6..a44da81b6c 100644 --- a/pyiceberg/catalog/hive.py +++ b/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, ) diff --git a/pyiceberg/catalog/rest/__init__.py b/pyiceberg/catalog/rest/__init__.py index e7131b1a88..74912b2177 100644 --- a/pyiceberg/catalog/rest/__init__.py +++ b/pyiceberg/catalog/rest/__init__.py @@ -624,8 +624,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) @@ -1138,8 +1143,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, @@ -1155,8 +1161,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, ) diff --git a/pyiceberg/catalog/sql.py b/pyiceberg/catalog/sql.py index abe31194b8..13e67997d8 100644 --- a/pyiceberg/catalog/sql.py +++ b/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, ) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 303b3db135..705081dea7 100644 --- a/pyiceberg/table/__init__.py +++ b/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"), ) diff --git a/tests/catalog/test_rest.py b/tests/catalog/test_rest.py index 0953c5f793..69b798a2d4 100644 --- a/tests/catalog/test_rest.py +++ b/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 @@ -3373,6 +3373,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", diff --git a/tests/catalog/test_sql.py b/tests/catalog/test_sql.py index 6c1767f711..df5e30e25f 100644 --- a/tests/catalog/test_sql.py +++ b/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",