GH-41365: [Python] Fix S3 URI with whitespace silently falling back to LocalFileSystem

cpp/src/arrow/filesystem/azurefs.cc

@@ -500,7 +500,7 @@ struct AzureLocation {
     // container = testcontainer
     // path = testdir/testfile.txt
     // path_parts = [testdir, testfile.txt]
-    if (internal::IsLikelyUri(string)) {
+    if (IsLikelyUri(string)) {
       return Status::Invalid(
           "Expected an Azure object location of the form 'container/path...',"
           " got a URI: '",

cpp/src/arrow/filesystem/filesystem.cc

@@ -278,7 +278,7 @@ Result<std::string> FileSystem::MakeUri(std::string path) const {
 namespace {
 
 Status ValidateSubPath(std::string_view s) {
-  if (internal::IsLikelyUri(s)) {
+  if (IsLikelyUri(s)) {
     return Status::Invalid("Expected a filesystem path, got a URI: '", s, "'");
   }
   return Status::OK();

cpp/src/arrow/filesystem/filesystem.h

@@ -24,6 +24,7 @@
 #include <iosfwd>
 #include <memory>
 #include <string>
+#include <string_view>
 #include <utility>
 #include <vector>
 
@@ -562,6 +563,12 @@ class ARROW_EXPORT SlowFileSystem : public FileSystem {
 /// The user is responsible for synchronization of calls to this function.
 void EnsureFinalized();
 
+/// \brief Return whether a path string is likely a URI.
+///
+/// This heuristic is conservative and may return false for malformed URIs.
+ARROW_EXPORT
+bool IsLikelyUri(std::string_view path);
+
 /// \defgroup filesystem-factories Functions for creating FileSystem instances
 ///
 /// @{

cpp/src/arrow/filesystem/gcsfs.cc

@@ -59,7 +59,7 @@ struct GcsPath {
   std::string object;
 
   static Result<GcsPath> FromString(const std::string& s) {
-    if (internal::IsLikelyUri(s)) {
+    if (IsLikelyUri(s)) {
       return Status::Invalid(
           "Expected a GCS object path of the form 'bucket/key...', got a URI: '", s, "'");
     }

cpp/src/arrow/filesystem/localfs.cc

@@ -55,7 +55,7 @@ using ::arrow::internal::PlatformFilename;
 namespace {
 
 Status ValidatePath(std::string_view s) {
-  if (internal::IsLikelyUri(s)) {
+  if (IsLikelyUri(s)) {
     return Status::Invalid("Expected a local filesystem path, got a URI: '", s, "'");
   }
   return Status::OK();

cpp/src/arrow/filesystem/mockfs.cc

@@ -45,7 +45,7 @@ namespace internal {
 namespace {
 
 Status ValidatePath(std::string_view s) {
-  if (internal::IsLikelyUri(s)) {
+  if (IsLikelyUri(s)) {
     return Status::Invalid("Expected a filesystem path, got a URI: '", s, "'");
   }
   return Status::OK();

cpp/src/arrow/filesystem/path_util.cc

@@ -337,6 +337,8 @@ bool IsEmptyPath(std::string_view v) {
   return true;
 }
 
+}  // namespace internal
+
 bool IsLikelyUri(std::string_view v) {
   if (v.empty() || v[0] == '/') {
     return false;
@@ -357,6 +359,8 @@ bool IsLikelyUri(std::string_view v) {
   return ::arrow::util::IsValidUriScheme(v.substr(0, pos));
 }
 
+namespace internal {
+
 struct Globber::Impl {
   std::regex pattern_;
 

cpp/src/arrow/filesystem/path_util.h

@@ -159,9 +159,6 @@ std::string ToSlashes(std::string_view s);
 ARROW_EXPORT
 bool IsEmptyPath(std::string_view s);
 
-ARROW_EXPORT
-bool IsLikelyUri(std::string_view s);
-
 class ARROW_EXPORT Globber {
  public:
   ~Globber();

cpp/src/arrow/filesystem/s3fs.cc

@@ -531,7 +531,7 @@ struct S3Path {
   std::vector<std::string> key_parts;
 
   static Result<S3Path> FromString(const std::string& s) {
-    if (internal::IsLikelyUri(s)) {
+    if (IsLikelyUri(s)) {
       return Status::Invalid(
           "Expected an S3 object path of the form 'bucket/key...', got a URI: '", s, "'");
     }
@@ -3660,7 +3660,7 @@ Result<std::string> ResolveS3BucketRegion(const std::string& bucket) {
   RETURN_NOT_OK(CheckS3Initialized());
 
   if (bucket.empty() || bucket.find_first_of(kSep) != bucket.npos ||
-      internal::IsLikelyUri(bucket)) {
+      IsLikelyUri(bucket)) {
     return Status::Invalid("Not a valid bucket name: '", bucket, "'");
   }
 

python/pyarrow/_fs.pyx

@@ -84,6 +84,14 @@ def _file_type_to_string(ty):
     return f"{ty.__class__.__name__}.{ty._name_}"
 
 
+def _is_likely_uri(path):
+    cdef c_string c_path
+    if not isinstance(path, str):
+        raise TypeError("Path must be a string")
+    c_path = tobytes(path)
+    return CIsLikelyUri(c_path)
+
+
 cdef class FileInfo(_Weakrefable):
     """
     FileSystem entry info.

python/pyarrow/fs.py

@@ -33,6 +33,7 @@
     PyFileSystem,
     _copy_files,
     _copy_files_selector,
+    _is_likely_uri,
 )
 
 # For backward compatibility.
@@ -174,13 +175,17 @@ def _resolve_filesystem_and_path(path, filesystem=None, *, memory_map=False):
             filesystem, path = FileSystem.from_uri(path)
         except ValueError as e:
             msg = str(e)
-            if "empty scheme" in msg or "Cannot parse URI" in msg:
-                # neither an URI nor a locally existing path, so assume that
-                # local path was given and propagate a nicer file not found
-                # error instead of a more confusing scheme parsing error
+            if "empty scheme" in msg:
+                # No scheme at all — treat as a local path and propagate
+                # a nicer "file not found" error later.
+                pass
+            elif "Cannot parse URI" in msg and not _is_likely_uri(path):
+                # Path doesn't look like a URI (no valid scheme prefix),
+                # so treat it as a local path rather than surfacing a
+                # confusing URI-parsing error.
                 pass
             else:
-                raise e
+                raise
     else:
         path = filesystem.normalize_path(path)
 

python/pyarrow/includes/libarrow_fs.pxd

@@ -93,6 +93,8 @@ cdef extern from "arrow/filesystem/api.h" namespace "arrow::fs" nogil:
         "arrow::fs::FileSystemFromUriOrPath"(const c_string& uri,
                                              c_string* out_path)
 
+    c_bool CIsLikelyUri "arrow::fs::IsLikelyUri"(const c_string& path)
+
     cdef cppclass CFileSystemGlobalOptions \
             "arrow::fs::FileSystemGlobalOptions":
         c_string tls_ca_file_path

python/pyarrow/tests/test_fs.py

@@ -1725,6 +1725,89 @@ def test_filesystem_from_path_object(path):
     assert path == p.resolve().absolute().as_posix()
 
 
+def test_is_likely_uri():
+    """Unit tests for the _is_likely_uri() heuristic."""
+    from pyarrow.fs import _is_likely_uri
+
+    # Valid URI schemes
+    assert _is_likely_uri("s3://bucket/key")
+    assert _is_likely_uri("gs://bucket/key")
+    assert _is_likely_uri("hdfs://host/path")
+    assert _is_likely_uri("file:///local/path")
+    assert _is_likely_uri("abfss://container@account/path")
+    assert _is_likely_uri("grpc+https://host:443")
+
+    # Only the scheme (everything before the first ':') is inspected, so
+    # non-ASCII characters in the *path* don't change the verdict.
+    assert _is_likely_uri("s3://asdf/äöü")
+    assert _is_likely_uri("s3://bucket/über/daten.parquet")
+    assert _is_likely_uri("s3://bucket/数据/file.parquet")
+
+    # Not URIs — local paths, Windows drives, empty, etc.
+    assert not _is_likely_uri("")
+    assert not _is_likely_uri("/absolute/path")
+    assert not _is_likely_uri("relative/path")
+    assert not _is_likely_uri("C:\\Users\\foo")      # single-letter → drive
+    assert not _is_likely_uri("C:/Users/foo")
+    assert not _is_likely_uri("3bucket://key")       # scheme starts with digit
+    assert not _is_likely_uri("-scheme://key")        # scheme starts with dash
+    assert not _is_likely_uri("schéme://bucket/key")  # non-ASCII in scheme
+    assert not _is_likely_uri("漢字://bucket/key")     # non-ASCII in scheme
+    assert not _is_likely_uri("/tmp/äöü/data")        # non-ASCII local path
+    assert not _is_likely_uri("dätä/file.parquet")    # non-ASCII, no scheme
+
+
+@pytest.mark.parametrize('uri', [
+    # Un-encoded spaces
+    "s3://bucket/path with space/file.parquet",
+    "gs://bucket/path with space/file.csv",
+    "abfss://container@account/dir with space/file",
+    # Un-encoded non-ASCII
+    "s3://asdf/äöü",
+    "s3://bucket/über/daten.parquet",
+    "s3://bucket/数据/file.parquet",
+    "gs://bucket/äöü/x.csv",
+    "abfss://container@account/äöü",
+])
+def test_resolve_filesystem_and_path_uri_unencoded(uri):
+    """
+    A URI with a recognised scheme but un-encoded characters must raise
+    ValueError — NOT silently fall back to LocalFileSystem.  The URI parser
+    rejects any byte outside the RFC 3986 set, so such paths have to be
+    percent-encoded (e.g. "s3://asdf/%C3%A4%C3%B6%C3%BC").
+    (GH-41365)
+    """
+    from pyarrow.fs import _resolve_filesystem_and_path
+
+    with pytest.raises(ValueError, match="Cannot parse URI"):
+        _resolve_filesystem_and_path(uri)
+
+
+@pytest.mark.parametrize('path', [
+    # Spaces
+    "/tmp/path with spaces/data",
+    "/nonexistent/path",
+    # Non-ASCII
+    "/tmp/äöü/data",
+    "/tmp/数据/data",
+    # Relative non-ASCII paths do reach the URI parser and fail with
+    # "Cannot parse URI", but _is_likely_uri() rejects them as URIs, so
+    # they must still fall back to LocalFileSystem.
+    "dätä/file.parquet",
+    "dätä/fi:le.parquet",
+])
+def test_resolve_filesystem_and_path_local_unencoded(path):
+    """
+    Local paths (no scheme) containing spaces or non-ASCII characters should
+    still resolve to LocalFileSystem — they must NOT be confused with
+    malformed URIs.
+    """
+    from pyarrow.fs import _resolve_filesystem_and_path
+
+    fs, _ = _resolve_filesystem_and_path(path)
+    assert isinstance(fs, LocalFileSystem)
+
+
 @pytest.mark.s3
 def test_filesystem_from_uri_s3(s3_server):
     from pyarrow.fs import S3FileSystem