Fix file:// URL handling in _register_object_store_for_path

- Allow empty netloc for file:// scheme (e.g. file:///tmp/data.parquet)
- Require netloc (host/bucket) only for non-file schemes (s3, gs, az, https)
- Add doctest +SKIP to register_parquet example referencing nonexistent file
- Use Path.as_uri() in end-to-end tests for portable file:// URL construction
- Add tests for file:// acceptance and scheme-without-host rejection

Author: qzyu999

python/datafusion/context.py

@@ -624,21 +624,32 @@ def _register_object_store_for_path(
         parameter is provided.
 
         Args:
-            path: A URL-style path (e.g. ``"s3://bucket/key.parquet"``).
+            path: A URL-style path (e.g. ``"s3://bucket/key.parquet"`` or
+                ``"file:///tmp/data.parquet"``).
             store: An object store instance to register.
 
         Raises:
-            ValueError: If the path does not contain a recognized URL scheme.
+            ValueError: If the path does not contain a URL scheme, or if
+                a non-file scheme is missing a host/bucket component.
         """
         parsed = urlparse(str(path))
-        if not parsed.scheme or not parsed.netloc:
+        if not parsed.scheme:
             msg = (
                 f"Cannot determine object store URL from path {path!r}. "
                 "The path must use a URL scheme (e.g. 's3://bucket/key')."
             )
             raise ValueError(msg)
+        # file:// URLs typically have an empty netloc (e.g. file:///tmp/a.parquet)
+        # For other schemes (s3, gs, az, https) the netloc (bucket/host) is required.
+        if parsed.scheme != "file" and not parsed.netloc:
+            msg = (
+                f"Cannot determine object store URL from path {path!r}. "
+                "The path must include a host or bucket "
+                "(e.g. 's3://bucket/key')."
+            )
+            raise ValueError(msg)
         scheme = f"{parsed.scheme}://"
-        host = parsed.netloc
+        host = parsed.netloc or None
         self.register_object_store(scheme, store, host=host)
 
     def register_listing_table(
@@ -1096,7 +1107,7 @@ def register_parquet(
 
             >>> import datafusion
             >>> ctx = datafusion.SessionContext()
-            >>> ctx.register_parquet("my_table", "data.parquet")
+            >>> ctx.register_parquet("my_table", "data.parquet")  # doctest: +SKIP
 
             Register from S3 with inline credentials (thread-safe):
 

python/tests/test_object_store_param.py

@@ -84,11 +84,26 @@ def test_raises_on_relative_path(self, ctx):
 
     def test_raises_on_windows_path(self, ctx):
         mock_store = MagicMock()
-        with pytest.raises(ValueError, match="Cannot determine object store URL"):
+        with pytest.raises(ValueError, match="must include a host or bucket"):
             ctx._register_object_store_for_path(
                 "C:\\Users\\data\\file.parquet", mock_store
             )
 
+    def test_raises_on_scheme_without_host_for_non_file(self, ctx):
+        """Non-file schemes (s3, gs, etc.) require a host/bucket."""
+        mock_store = MagicMock()
+        with pytest.raises(ValueError, match="must include a host or bucket"):
+            ctx._register_object_store_for_path("s3:///key.parquet", mock_store)
+
+    def test_parses_file_url(self, ctx):
+        """file:// URLs with empty netloc should be accepted."""
+        mock_store = MagicMock()
+        with patch.object(ctx, "register_object_store") as mock_register:
+            ctx._register_object_store_for_path(
+                "file:///tmp/path/to/file.parquet", mock_store
+            )
+            mock_register.assert_called_once_with("file://", mock_store, host=None)
+
     def test_accepts_pathlib_path_raises(self, ctx):
         """pathlib.Path cannot represent URLs, so this should raise."""
         mock_store = MagicMock()
@@ -258,7 +273,7 @@ def test_register_parquet_with_local_object_store(self, ctx, tmp_path):
 
         # Use file:// URL with LocalFileSystem object store
         store = LocalFileSystem()
-        file_url = f"file://{tmp_path}/test.parquet"
+        file_url = parquet_path.as_uri()
 
         ctx.register_parquet("test_tbl", file_url, object_store=store)
         result = ctx.sql("SELECT * FROM test_tbl").collect()
@@ -277,7 +292,7 @@ def test_read_parquet_with_local_object_store(self, ctx, tmp_path):
         pq.write_table(table, str(parquet_path))
 
         store = LocalFileSystem()
-        file_url = f"file://{tmp_path}/read_test.parquet"
+        file_url = parquet_path.as_uri()
 
         df = ctx.read_parquet(file_url, object_store=store)
         result = df.collect()