fix(rest): refine S3-compatible scheme fallback

Author: plusplusjiajia

pyiceberg/catalog/rest/__init__.py

@@ -396,7 +396,7 @@ class ListViewsResponse(IcebergBaseModel):
 _PLANNING_RESPONSE_ADAPTER = TypeAdapter(PlanningResponse)
 
 
-def _is_hadoop_only_config(config: Properties) -> bool:
+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 +487,9 @@ 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://")):
+        # Java S3FileIO ROOT_PREFIX fallback. Only oss:// needs it — s3/s3a/s3n
+        # already match the "s3" credential via startswith in the loop above.
+        if best_match is None and location.startswith("oss://"):
             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

@@ -3112,7 +3112,6 @@ def test_resolve_storage_credentials_empty() -> None:
 def test_resolve_storage_credentials_skips_hadoop_only() -> None:
     from pyiceberg.catalog.rest.scan_planning import StorageCredential
 
-    # The longer fs.* prefix would win a blind longest-match; the filter drops it.
     credentials = [
         StorageCredential(prefix="s3://warehouse/jindo", config={"fs.s3.access-key": "hadoop-k"}),
         StorageCredential(prefix="s3://warehouse", config={"s3.access-key-id": "native-k"}),
@@ -3141,10 +3140,20 @@ 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
+
+    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 routed through pyarrow's S3FileSystem, so ROOT_PREFIX "s3" applies.
     credentials = [
         StorageCredential(prefix="s3", config={"s3.access-key-id": "native-k"}),
     ]