FsspecFileIO: Resolve the ADLS account per location instead of caching it in properties

_adls wrote the account name it inferred back into the properties dict it was
given, and the caller passes the FileIO's own properties. The first location
therefore pinned the account for every later one, so a second storage account
was served a filesystem built for the first. Resolve the account and the SAS
token into locals and leave the caller's dict alone. Precedence is unchanged:
an explicit adls.account-name still wins, then a per account SAS token key,
then the hostname.

Author: krishnakaanchan-png

pyiceberg/io/fsspec.py

@@ -265,17 +265,23 @@ def _adls(properties: Properties, hostname: str | None = None) -> AbstractFileSy
     from azure.core.credentials import AccessToken
     from azure.core.credentials_async import AsyncTokenCredential
 
-    for key, sas_token in {
+    # These are resolved into locals rather than written back into `properties`. The same
+    # properties dict is shared by every location this FileIO serves, so storing an account
+    # name derived from one location would apply it to all the later ones as well.
+    account_name = properties.get(ADLS_ACCOUNT_NAME)
+    sas_token = properties.get(ADLS_SAS_TOKEN)
+
+    for key, per_account_sas_token in {
         key.replace(f"{ADLS_SAS_TOKEN}.", ""): value for key, value in properties.items() if key.startswith(f"{ADLS_SAS_TOKEN}.")
     }.items():
-        if ADLS_ACCOUNT_NAME not in properties:
-            properties[ADLS_ACCOUNT_NAME] = key.split(".")[0]
-        if ADLS_SAS_TOKEN not in properties:
-            properties[ADLS_SAS_TOKEN] = sas_token
+        if account_name is None:
+            account_name = key.split(".")[0]
+        if sas_token is None:
+            sas_token = per_account_sas_token
 
     # Fallback: extract account_name from URI hostname (e.g. "account.dfs.core.windows.net" -> "account")
-    if hostname and ADLS_ACCOUNT_NAME not in properties:
-        properties[ADLS_ACCOUNT_NAME] = hostname.split(".")[0]
+    if hostname and account_name is None:
+        account_name = hostname.split(".")[0]
 
     class StaticTokenCredential(AsyncTokenCredential):
         _DEFAULT_EXPIRY_SECONDS = 3600
@@ -298,9 +304,9 @@ async def get_token(self, *scopes: str, **kwargs: Any) -> AccessToken:
     return AzureBlobFileSystem(
         connection_string=properties.get(ADLS_CONNECTION_STRING),
         credential=credential,
-        account_name=properties.get(ADLS_ACCOUNT_NAME),
+        account_name=account_name,
         account_key=properties.get(ADLS_ACCOUNT_KEY),
-        sas_token=properties.get(ADLS_SAS_TOKEN),
+        sas_token=sas_token,
         tenant_id=properties.get(ADLS_TENANT_ID),
         client_id=properties.get(ADLS_CLIENT_ID),
         client_secret=properties.get(ADLS_CLIENT_SECRET),

tests/io/test_fsspec.py

@@ -697,6 +697,32 @@ def test_adls_account_name_extracted_from_uri_hostname() -> None:
         )
 
 
+def test_adls_account_name_resolved_per_location() -> None:
+    """Two locations in different accounts must each get a filesystem for their own account."""
+    session_properties: Properties = {"adls.tenant-id": "test-tenant-id"}
+
+    with mock.patch("adlfs.AzureBlobFileSystem") as mock_adlfs:
+        adls_fileio = FsspecFileIO(properties=session_properties)
+
+        adls_fileio.new_input(location="abfss://data@accountone.dfs.core.windows.net/wh/t/a.parquet")
+        adls_fileio.new_input(location="abfss://data@accounttwo.dfs.core.windows.net/wh/t/b.parquet")
+
+    account_names = [call.kwargs["account_name"] for call in mock_adlfs.call_args_list]
+    assert account_names == ["accountone", "accounttwo"]
+
+
+def test_adls_does_not_mutate_properties() -> None:
+    """Resolving an account from a location must not write it back into the shared properties."""
+    session_properties: Properties = {"adls.tenant-id": "test-tenant-id"}
+
+    with mock.patch("adlfs.AzureBlobFileSystem"):
+        adls_fileio = FsspecFileIO(properties=session_properties)
+        adls_fileio.new_input(location="abfss://data@accountone.dfs.core.windows.net/wh/t/a.parquet")
+
+    assert "adls.account-name" not in adls_fileio.properties
+    assert session_properties == {"adls.tenant-id": "test-tenant-id"}
+
+
 def test_adls_account_name_not_overridden_when_in_properties() -> None:
     """Test that explicit adls.account-name in properties is not overridden by URI hostname."""
     session_properties: Properties = {