Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 11 additions & 4 deletions pyiceberg/catalog/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
)

Expand Down
2 changes: 1 addition & 1 deletion pyiceberg/catalog/bigquery_metastore.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)

Expand Down
2 changes: 1 addition & 1 deletion pyiceberg/catalog/dynamodb.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)

Expand Down
6 changes: 3 additions & 3 deletions pyiceberg/catalog/glue.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)

Expand Down Expand Up @@ -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,
)

Expand Down Expand Up @@ -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,
)

Expand Down
2 changes: 1 addition & 1 deletion pyiceberg/catalog/hive.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)

Expand Down
15 changes: 11 additions & 4 deletions pyiceberg/catalog/rest/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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,
Expand All @@ -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,
)
Expand Down
2 changes: 1 addition & 1 deletion pyiceberg/catalog/sql.py
Original file line number Diff line number Diff line change
Expand Up @@ -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,
)

Expand Down
2 changes: 1 addition & 1 deletion pyiceberg/table/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"),
)

Expand Down
28 changes: 27 additions & 1 deletion tests/catalog/test_rest.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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",
Expand Down
20 changes: 20 additions & 0 deletions tests/catalog/test_sql.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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",
Expand Down
Loading