fix: preserve original path for non-S3 schemes in access point resolution

The _resolve_s3_access_point method was incorrectly constructing paths
for non-S3 schemes (like local files) by concatenating netloc and path_suffix.
This caused issues when local paths had double slashes (e.g., //tmp/...) because
urlparse interprets these as network paths with netloc.

Now the method takes the original path from parse_location as a parameter
and returns it unchanged for non-S3 schemes, ensuring local file operations
work correctly.

Author: JannicCutura

pyiceberg/io/pyarrow.py

@@ -589,7 +589,7 @@ def _initialize_gcs_fs(self) -> FileSystem:
     def _initialize_local_fs(self) -> FileSystem:
         return PyArrowLocalFileSystem()
 
-    def _resolve_s3_access_point(self, scheme: str, netloc: str, path_suffix: str) -> tuple[str, str]:
+    def _resolve_s3_access_point(self, scheme: str, netloc: str, path_suffix: str, original_path: str) -> tuple[str, str]:
         """Resolve S3 access point alias for a bucket if configured.
 
         For cross-account access, S3 paths need to use access point aliases instead of bucket names.
@@ -600,12 +600,13 @@ def _resolve_s3_access_point(self, scheme: str, netloc: str, path_suffix: str) -
             scheme: The URI scheme (s3, s3a, s3n)
             netloc: The bucket name from the original URI
             path_suffix: The path within the bucket (without bucket name)
+            original_path: The original path from parse_location (fallback for non-S3)
 
         Returns:
             Tuple of (resolved_netloc, resolved_path) where netloc may be replaced with access point alias
         """
         if scheme not in {"s3", "s3a", "s3n"}:
-            return netloc, f"{netloc}{path_suffix}"
+            return netloc, original_path
 
         # Check for access point alias configuration for this bucket
         access_point_key = f"{S3_ACCESS_POINT_PREFIX}{netloc}"
@@ -614,7 +615,7 @@ def _resolve_s3_access_point(self, scheme: str, netloc: str, path_suffix: str) -
             # Replace bucket with access point alias in the path
             return access_point_alias, f"{access_point_alias}{path_suffix}"
 
-        return netloc, f"{netloc}{path_suffix}"
+        return netloc, original_path
 
     def new_input(self, location: str) -> PyArrowFile:
         """Get a PyArrowFile instance to read bytes from the file at the given location.
@@ -625,10 +626,10 @@ def new_input(self, location: str) -> PyArrowFile:
         Returns:
             PyArrowFile: A PyArrowFile instance for the given location.
         """
-        scheme, netloc, _ = self.parse_location(location, self.properties)
-        # For S3, resolve access point ARN if configured
+        scheme, netloc, path = self.parse_location(location, self.properties)
+        # For S3, resolve access point if configured
         uri = urlparse(location)
-        resolved_netloc, resolved_path = self._resolve_s3_access_point(scheme, netloc, uri.path)
+        resolved_netloc, resolved_path = self._resolve_s3_access_point(scheme, netloc, uri.path, path)
         return PyArrowFile(
             fs=self.fs_by_scheme(scheme, resolved_netloc),
             location=location,
@@ -645,10 +646,10 @@ def new_output(self, location: str) -> PyArrowFile:
         Returns:
             PyArrowFile: A PyArrowFile instance for the given location.
         """
-        scheme, netloc, _ = self.parse_location(location, self.properties)
-        # For S3, resolve access point ARN if configured
+        scheme, netloc, path = self.parse_location(location, self.properties)
+        # For S3, resolve access point if configured
         uri = urlparse(location)
-        resolved_netloc, resolved_path = self._resolve_s3_access_point(scheme, netloc, uri.path)
+        resolved_netloc, resolved_path = self._resolve_s3_access_point(scheme, netloc, uri.path, path)
         return PyArrowFile(
             fs=self.fs_by_scheme(scheme, resolved_netloc),
             location=location,
@@ -670,10 +671,10 @@ def delete(self, location: str | InputFile | OutputFile) -> None:
                 an AWS error code 15.
         """
         str_location = location.location if isinstance(location, (InputFile, OutputFile)) else location
-        scheme, netloc, _ = self.parse_location(str_location, self.properties)
-        # For S3, resolve access point ARN if configured
+        scheme, netloc, path = self.parse_location(str_location, self.properties)
+        # For S3, resolve access point if configured
         uri = urlparse(str_location)
-        resolved_netloc, resolved_path = self._resolve_s3_access_point(scheme, netloc, uri.path)
+        resolved_netloc, resolved_path = self._resolve_s3_access_point(scheme, netloc, uri.path, path)
         fs = self.fs_by_scheme(scheme, resolved_netloc)
 
         try:

tests/io/test_pyarrow.py

@@ -458,7 +458,8 @@ def test_s3_access_point_resolution_with_config() -> None:
     fileio = PyArrowFileIO(properties=properties)
 
     # Test _resolve_s3_access_point directly
-    resolved_netloc, resolved_path = fileio._resolve_s3_access_point("s3", bucket_name, "/path/to/file.parquet")
+    original_path = f"{bucket_name}/path/to/file.parquet"
+    resolved_netloc, resolved_path = fileio._resolve_s3_access_point("s3", bucket_name, "/path/to/file.parquet", original_path)
 
     assert resolved_netloc == access_point_alias
     assert resolved_path == f"{access_point_alias}/path/to/file.parquet"
@@ -469,7 +470,8 @@ def test_s3_access_point_resolution_without_config() -> None:
     bucket_name = "my-bucket"
     fileio = PyArrowFileIO(properties={})
 
-    resolved_netloc, resolved_path = fileio._resolve_s3_access_point("s3", bucket_name, "/path/to/file.parquet")
+    original_path = f"{bucket_name}/path/to/file.parquet"
+    resolved_netloc, resolved_path = fileio._resolve_s3_access_point("s3", bucket_name, "/path/to/file.parquet", original_path)
 
     assert resolved_netloc == bucket_name
     assert resolved_path == f"{bucket_name}/path/to/file.parquet"
@@ -486,10 +488,11 @@ def test_s3_access_point_resolution_non_s3_scheme() -> None:
     fileio = PyArrowFileIO(properties=properties)
 
     # Test with non-S3 scheme (should not resolve)
-    resolved_netloc, resolved_path = fileio._resolve_s3_access_point("gs", bucket_name, "/path/to/file.parquet")
+    original_path = f"{bucket_name}/path/to/file.parquet"
+    resolved_netloc, resolved_path = fileio._resolve_s3_access_point("gs", bucket_name, "/path/to/file.parquet", original_path)
 
     assert resolved_netloc == bucket_name
-    assert resolved_path == f"{bucket_name}/path/to/file.parquet"
+    assert resolved_path == original_path
 
 
 def test_s3_access_point_resolution_s3a_scheme() -> None:
@@ -502,7 +505,8 @@ def test_s3_access_point_resolution_s3a_scheme() -> None:
 
     fileio = PyArrowFileIO(properties=properties)
 
-    resolved_netloc, resolved_path = fileio._resolve_s3_access_point("s3a", bucket_name, "/path/to/file.parquet")
+    original_path = f"{bucket_name}/path/to/file.parquet"
+    resolved_netloc, resolved_path = fileio._resolve_s3_access_point("s3a", bucket_name, "/path/to/file.parquet", original_path)
 
     assert resolved_netloc == access_point_alias
     assert resolved_path == f"{access_point_alias}/path/to/file.parquet"