From c334954791cfdb294588e256c3afc7c32b5f9140 Mon Sep 17 00:00:00 2001
From: Anas Khan <83116240+anxkhn@users.noreply.github.com>
Date: Sat, 4 Jul 2026 22:30:41 +0530
Subject: [PATCH] fix: return flat namespace properties from
 BigQueryMetastoreCatalog.load_namespace_properties

BigQueryMetastoreCatalog.create_namespace stores user properties in the
ExternalCatalogDatasetOptions.parameters map, but load_namespace_properties
returned external_catalog_dataset_options.to_api_repr(). That API
representation is a nested dict that buries the properties under a
"parameters" key and adds "defaultStorageLocationUri", so a round-trip of
create_namespace(ns, {"owner": "..."}) followed by
load_namespace_properties(ns)["owner"] raised KeyError.

Return the flat parameters map instead, matching the base MetastoreCatalog
contract and the Glue and DynamoDB catalogs. Guard the None case (parameters
is unset by default) so a namespace with no properties yields an empty dict.

Add regression tests covering both the populated round-trip and the
no-parameters case.

Signed-off-by: Anas Khan <83116240+anxkhn@users.noreply.github.com>
---
 pyiceberg/catalog/bigquery_metastore.py  |  2 +-
 tests/catalog/test_bigquery_metastore.py | 41 ++++++++++++++++++++++++
 2 files changed, 42 insertions(+), 1 deletion(-)

diff --git a/pyiceberg/catalog/bigquery_metastore.py b/pyiceberg/catalog/bigquery_metastore.py
index 938ac6992f..dcb53bd445 100644
--- a/pyiceberg/catalog/bigquery_metastore.py
+++ b/pyiceberg/catalog/bigquery_metastore.py
@@ -341,7 +341,7 @@ def load_namespace_properties(self, namespace: str | Identifier) -> Properties:
             dataset = self.client.get_dataset(DatasetReference(project=self.project_id, dataset_id=dataset_name))
 
             if dataset and dataset.external_catalog_dataset_options:
-                return dataset.external_catalog_dataset_options.to_api_repr()
+                return dict(dataset.external_catalog_dataset_options.parameters or {})
         except NotFound as e:
             raise NoSuchNamespaceError(f"Namespace {namespace} not found") from e
         return {}
diff --git a/tests/catalog/test_bigquery_metastore.py b/tests/catalog/test_bigquery_metastore.py
index c8c7584262..ba78755321 100644
--- a/tests/catalog/test_bigquery_metastore.py
+++ b/tests/catalog/test_bigquery_metastore.py
@@ -123,6 +123,47 @@ def test_drop_namespace(mocker: MockFixture, gcp_dataset_name: str) -> None:
     assert args[0].dataset_id == gcp_dataset_name
 
 
+def test_load_namespace_properties_returns_flat_parameters(mocker: MockFixture, gcp_dataset_name: str) -> None:
+    client_mock = MagicMock()
+
+    # Round-trip the Dataset that create_namespace persists back through get_dataset,
+    # mimicking how BigQuery stores and returns the namespace.
+    def create_dataset(dataset: Dataset) -> Dataset:
+        client_mock.get_dataset.return_value = dataset
+        return dataset
+
+    client_mock.create_dataset.side_effect = create_dataset
+
+    mocker.patch("pyiceberg.catalog.bigquery_metastore.Client", return_value=client_mock)
+    mocker.patch.dict(os.environ, values={"PYICEBERG_LEGACY_CURRENT_SNAPSHOT_ID": "True"})
+
+    test_catalog = BigQueryMetastoreCatalog(
+        "test_catalog", **{"gcp.bigquery.project-id": "my-project", "warehouse": "gs://test-bucket/"}
+    )
+
+    properties = {"owner": "anas", "comment": "namespace comment"}
+    test_catalog.create_namespace(namespace=gcp_dataset_name, properties=properties)
+
+    # The stored, flat property map must be returned as-is, not the nested BigQuery API
+    # representation (which would bury the values under a "parameters" key and add
+    # "defaultStorageLocationUri").
+    assert test_catalog.load_namespace_properties(gcp_dataset_name) == properties
+
+
+def test_load_namespace_properties_without_parameters(mocker: MockFixture, gcp_dataset_name: str) -> None:
+    client_mock = MagicMock()
+    client_mock.get_dataset.return_value = dataset_mock()
+
+    mocker.patch("pyiceberg.catalog.bigquery_metastore.Client", return_value=client_mock)
+    mocker.patch.dict(os.environ, values={"PYICEBERG_LEGACY_CURRENT_SNAPSHOT_ID": "True"})
+
+    test_catalog = BigQueryMetastoreCatalog("test_catalog", **{"gcp.bigquery.project-id": "my-project"})
+
+    # A dataset whose external_catalog_dataset_options carries no parameters yields an
+    # empty property map, not None or a nested dict.
+    assert test_catalog.load_namespace_properties(gcp_dataset_name) == {}
+
+
 def test_list_tables(mocker: MockFixture, gcp_dataset_name: str) -> None:
     client_mock = MagicMock()