fix(rest): refine S3-compatible scheme fallback

Author: plusplusjiajia

pyiceberg/catalog/rest/__init__.py

@@ -396,7 +396,10 @@ class ListViewsResponse(IcebergBaseModel):
 _PLANNING_RESPONSE_ADAPTER = TypeAdapter(PlanningResponse)
 
 
-def _is_hadoop_only_config(config: Properties) -> bool:
+_S3_COMPATIBLE_SCHEMES = ("s3://", "s3a://", "s3n://", "oss://")
+
+
+def _is_hadoop_only_config(config: dict[str, str]) -> bool:
     """Return True if every key is a Hadoop ``fs.*`` key — pyiceberg has no HadoopFileIO to consume them."""
     return bool(config) and all(k.startswith("fs.") for k in config)
 
@@ -487,10 +490,8 @@ def _resolve_storage_credentials(storage_credentials: list[StorageCredential], l
                 if best_match is None or len(cred.prefix) > len(best_match.prefix):
                     best_match = cred
 
-        # Java S3FileIO falls back to the "s3" ROOT_PREFIX credential; scope it to
-        # schemes pyarrow's S3FileSystem handles so non-S3 schemes (gs://, abfs://,
-        # etc.) don't get handed s3.* keys.
-        if best_match is None and location.startswith(("s3://", "s3a://", "s3n://", "oss://")):
+        # Mirrors Java S3FileIO.clientForStoragePath() ROOT_PREFIX fallback.
+        if best_match is None and location.startswith(_S3_COMPATIBLE_SCHEMES):
             best_match = next((c for c in consumable if c.prefix == "s3"), None)
 
         return best_match.config if best_match else {}

tests/catalog/test_rest.py

@@ -3145,10 +3145,24 @@ def test_resolve_storage_credentials_all_hadoop_only_returns_empty() -> None:
     assert RestCatalog._resolve_storage_credentials(credentials, "custom://bucket/path") == {}
 
 
-def test_resolve_storage_credentials_root_prefix_fallback_for_s3_compatible_scheme() -> None:
+def test_resolve_storage_credentials_s3a_s3n_match_root_prefix_directly() -> None:
     from pyiceberg.catalog.rest.scan_planning import StorageCredential
 
-    # oss:// is routed through pyarrow's S3FileSystem, so ROOT_PREFIX "s3" applies.
+    # s3a:// and s3n:// start with "s3", so they match prefix="s3" via
+    # startswith at the normal matching stage — no fallback needed.
+    credentials = [
+        StorageCredential(prefix="s3", config={"s3.access-key-id": "native-k"}),
+    ]
+    for scheme in ("s3a", "s3n"):
+        result = RestCatalog._resolve_storage_credentials(credentials, f"{scheme}://bucket/path")
+        assert result == {"s3.access-key-id": "native-k"}, f"{scheme}:// should match prefix='s3' directly"
+
+
+def test_resolve_storage_credentials_root_prefix_fallback_for_oss() -> None:
+    from pyiceberg.catalog.rest.scan_planning import StorageCredential
+
+    # oss:// is S3-API-compatible; pyiceberg routes it through pyarrow's
+    # S3FileSystem, so the ROOT_PREFIX "s3" credential applies.
     credentials = [
         StorageCredential(prefix="s3", config={"s3.access-key-id": "native-k"}),
     ]