feat: support read_parquet for backend with no native support

ibis/backends/__init__.py

@@ -3,10 +3,13 @@
 import abc
 import collections.abc
 import functools
+import glob
 import importlib.metadata
 import keyword
 import re
 import urllib.parse
+import urllib.request
+from io import BytesIO
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar
 
@@ -1199,6 +1202,61 @@ def has_operation(cls, operation: type[ops.Value]) -> bool:
             f"{cls.name} backend has not implemented `has_operation` API"
         )
 
+    def read_parquet(
+        self, path: str | Path, table_name: str | None = None, **kwargs: Any
+    ) -> ir.Table:
+        """Register a parquet file as a table in the current backend.
+
+        Parameters
+        ----------
+        path
+            The data source. May be a path to a file, an iterable of files,
+            or directory of parquet files.
+        table_name
+            An optional name to use for the created table. This defaults to
+            a sequentially generated name.
+        **kwargs
+            Additional keyword arguments passed to the pyarrow loading function.
+            See https://arrow.apache.org/docs/python/generated/pyarrow.parquet.read_table.html
+            for more information.
+
+        Returns
+        -------
+        ir.Table
+            The just-registered table
+
+        """
+
+        table = self._get_pyarrow_table_from_path(path, **kwargs)
+        table_name = table_name or util.gen_name("read_parquet")
+        self.create_table(table_name, table)
+        return self.table(table_name)
+
+    def _get_pyarrow_table_from_path(self, path: str | Path, **kwargs) -> pa.Table:
+        pq = util.import_object("pyarrow.parquet")
+
+        path = str(path)
+        # handle url
+        if util.is_url(path):
+            headers = kwargs.pop("headers", {})
+            req_info = urllib.request.Request(path, headers=headers)  # noqa: S310
+            with urllib.request.urlopen(req_info) as req:  # noqa: S310
+                with BytesIO(req.read()) as reader:
+                    return pq.read_table(reader)
+
+        # handle fsspec compatible url
+        if util.is_fsspec_url(path):
+            return pq.read_table(path, **kwargs)
+
+        # Handle local file paths or patterns
+        paths = glob.glob(path)
+        if not paths:
+            raise ValueError(f"No files found matching pattern: {path!r}")
+        elif len(paths) == 1:
+            paths = paths[0]
+
+        return pq.read_table(paths, **kwargs)
+
     def _cached(self, expr: ir.Table):
         """Cache the provided expression.
 

ibis/backends/tests/test_register.py

@@ -4,8 +4,10 @@
 import csv
 import gzip
 import os
+import urllib
 from pathlib import Path
 from typing import TYPE_CHECKING
+from unittest import mock
 
 import pytest
 from pytest import param
@@ -18,8 +20,6 @@
 
     import pyarrow as pa
 
-pytestmark = pytest.mark.notimpl(["druid", "exasol", "oracle"])
-
 
 @contextlib.contextmanager
 def pushd(new_dir):
@@ -98,6 +98,7 @@ def gzip_csv(data_dir, tmp_path):
         "trino",
     ]
 )
+@pytest.mark.notimpl(["druid", "exasol", "oracle"])
 def test_register_csv(con, data_dir, fname, in_table_name, out_table_name):
     with pushd(data_dir / "csv"):
         with pytest.warns(FutureWarning, match="v9.1"):
@@ -109,7 +110,7 @@ def test_register_csv(con, data_dir, fname, in_table_name, out_table_name):
 
 
 # TODO: rewrite or delete test when register api is removed
-@pytest.mark.notimpl(["datafusion"])
+@pytest.mark.notimpl(["datafusion", "druid", "exasol", "oracle"])
 @pytest.mark.notyet(
     [
         "bigquery",
@@ -153,6 +154,7 @@ def test_register_csv_gz(con, data_dir, gzip_csv):
         "trino",
     ]
 )
+@pytest.mark.notimpl(["druid", "exasol", "oracle"])
 def test_register_with_dotted_name(con, data_dir, tmp_path):
     basename = "foo.bar.baz/diamonds.csv"
     f = tmp_path.joinpath(basename)
@@ -212,6 +214,7 @@ def read_table(path: Path) -> Iterator[tuple[str, pa.Table]]:
         "trino",
     ]
 )
+@pytest.mark.notimpl(["druid", "exasol", "oracle"])
 def test_register_parquet(
     con, tmp_path, data_dir, fname, in_table_name, out_table_name
 ):
@@ -252,6 +255,7 @@ def test_register_parquet(
         "trino",
     ]
 )
+@pytest.mark.notimpl(["druid", "exasol", "oracle"])
 def test_register_iterator_parquet(
     con,
     tmp_path,
@@ -280,7 +284,7 @@ def test_register_iterator_parquet(
 # TODO: remove entirely when `register` is removed
 # This same functionality is implemented across all backends
 # via `create_table` and tested in `test_client.py`
-@pytest.mark.notimpl(["datafusion"])
+@pytest.mark.notimpl(["datafusion", "druid", "exasol", "oracle"])
 @pytest.mark.notyet(
     [
         "bigquery",
@@ -316,7 +320,7 @@ def test_register_pandas(con):
 # TODO: remove entirely when `register` is removed
 # This same functionality is implemented across all backends
 # via `create_table` and tested in `test_client.py`
-@pytest.mark.notimpl(["datafusion", "polars"])
+@pytest.mark.notimpl(["datafusion", "polars", "druid", "exasol", "oracle"])
 @pytest.mark.notyet(
     [
         "bigquery",
@@ -361,6 +365,7 @@ def test_register_pyarrow_tables(con):
         "trino",
     ]
 )
+@pytest.mark.notimpl(["druid", "exasol", "oracle"])
 def test_csv_reregister_schema(con, tmp_path):
     foo = tmp_path.joinpath("foo.csv")
     with foo.open("w", newline="") as csvfile:
@@ -390,10 +395,13 @@ def test_csv_reregister_schema(con, tmp_path):
         "clickhouse",
         "dask",
         "datafusion",
+        "druid",
+        "exasol",
         "flink",
         "impala",
         "mysql",
         "mssql",
+        "oracle",
         "pandas",
         "polars",
         "postgres",
@@ -425,9 +433,8 @@ def test_register_garbage(con, monkeypatch):
         ("functional_alltypes.parquet", "funk_all"),
     ],
 )
-@pytest.mark.notyet(
-    ["flink", "impala", "mssql", "mysql", "postgres", "risingwave", "sqlite", "trino"]
-)
+@pytest.mark.notyet(["flink"])
+@pytest.mark.notimpl(["druid"])
 def test_read_parquet(con, tmp_path, data_dir, fname, in_table_name):
     pq = pytest.importorskip("pyarrow.parquet")
 
@@ -448,6 +455,61 @@ def test_read_parquet(con, tmp_path, data_dir, fname, in_table_name):
     assert table.count().execute()
 
 
+# test reading a Parquet file from a URL request for backends using pyarrow
+# that do not have their own implementation
+@pytest.mark.parametrize(
+    ("url", "in_table_name"),
+    [
+        ("http://example.com/functional_alltypes.parquet", "http_table"),
+        ("sftp://example.com/path/to/functional_alltypes.parquet", "sftp_table"),
+    ],
+)
+@pytest.mark.notimpl(
+    [
+        "druid",
+        "flink",
+        "duckdb",
+        "pandas",
+        "polars",
+        "bigquery",
+        "dask",
+        "clickhouse",
+        "datafusion",
+        "snowflake",
+        "pyspark",
+    ]
+)
+def test_read_parquet_url_request(con, url, data_dir, in_table_name, monkeypatch):
+    pytest.importorskip("pyarrow.parquet")
+
+    headers = {"User-Agent": "test-agent"}
+    fname = Path("functional_alltypes.parquet")
+    fname = Path(data_dir) / "parquet" / fname.name
+    mock_calls = []
+
+    mock_request = mock.create_autospec(urllib.request.Request)
+
+    def mock_urlopen(request, *args, **kwargs):
+        mock_calls.append((request, args, kwargs))
+        return open(fname, "rb")  # noqa: SIM115
+
+    monkeypatch.setattr("urllib.request.Request", mock_request)
+    monkeypatch.setattr("urllib.request.urlopen", mock_urlopen)
+
+    table = con.read_parquet(url, in_table_name, headers=headers)
+
+    mock_request.assert_called_once_with(url, headers=headers)
+    called_url = mock_request.call_args[0][0]
+    called_headers = mock_request.call_args[1]
+    assert url == called_url
+    assert called_headers["headers"] == headers
+    assert len(mock_calls) == 1
+    assert table.count().execute()
+
+    if in_table_name is not None:
+        assert table.op().name == in_table_name
+
+
 @pytest.fixture(scope="module")
 def ft_data(data_dir):
     pq = pytest.importorskip("pyarrow.parquet")
@@ -456,19 +518,8 @@ def ft_data(data_dir):
     return table.slice(0, nrows)
 
 
-@pytest.mark.notyet(
-    [
-        "flink",
-        "impala",
-        "mssql",
-        "mysql",
-        "pandas",
-        "postgres",
-        "risingwave",
-        "sqlite",
-        "trino",
-    ]
-)
+@pytest.mark.notyet(["flink", "pandas"])
+@pytest.mark.notimpl(["druid"])
 def test_read_parquet_glob(con, tmp_path, ft_data):
     pq = pytest.importorskip("pyarrow.parquet")
 
@@ -498,6 +549,7 @@ def test_read_parquet_glob(con, tmp_path, ft_data):
         "trino",
     ]
 )
+@pytest.mark.notimpl(["druid", "exasol", "oracle"])
 def test_read_csv_glob(con, tmp_path, ft_data):
     pc = pytest.importorskip("pyarrow.csv")
 
@@ -534,6 +586,7 @@ def test_read_csv_glob(con, tmp_path, ft_data):
     raises=ValueError,
     reason="read_json() missing required argument: 'schema'",
 )
+@pytest.mark.notimpl(["druid", "exasol", "oracle"])
 def test_read_json_glob(con, tmp_path, ft_data):
     nrows = len(ft_data)
     ntables = 2
@@ -580,6 +633,7 @@ def num_diamonds(data_dir):
 @pytest.mark.notyet(
     ["flink", "impala", "mssql", "mysql", "postgres", "risingwave", "sqlite", "trino"]
 )
+@pytest.mark.notimpl(["druid", "exasol", "oracle"])
 def test_read_csv(con, data_dir, in_table_name, num_diamonds):
     fname = "diamonds.csv"
     with pushd(data_dir / "csv"):

ibis/tests/test_util.py

@@ -4,7 +4,13 @@
 
 import pytest
 
-from ibis.util import PseudoHashable, flatten_iterable, import_object
+from ibis.util import (
+    PseudoHashable,
+    flatten_iterable,
+    import_object,
+    is_fsspec_url,
+    is_url,
+)
 
 
 @pytest.mark.parametrize(
@@ -51,6 +57,65 @@ def test_import_object():
         import_object("collections.this_attribute_doesnt_exist")
 
 
+@pytest.mark.parametrize(
+    ("url", "expected"),
+    [
+        ("http://example.com", True),  # Valid http URL
+        ("https://example.com", True),  # Valid https URL
+        ("ftp://example.com", True),  # Valid ftp URL
+        ("sftp://example.com", True),  # Valid sftp URL
+        ("ws://example.com", True),  # Valid WebSocket URL
+        ("wss://example.com", True),  # Valid WebSocket Secure URL
+        ("file:///home/user/file.txt", True),  # Valid file URL
+        ("mailto:example@example.com", False),  # Invalid URL with non-supported scheme
+        ("http://localhost:8000", True),  # Valid URL with port
+        ("ftp://192.168.1.1", True),  # Valid URL with IP address
+        ("https://example.com/path/to/resource", True),  # Valid URL with path
+        ("http://user:pass@example.com", True),  # Valid URL with credentials
+        ("ftp://example.com/resource", True),  # Valid FTP URL with resource
+        ("telnet://example.com", True),  # Valid Telnet URL
+        ("git://example.com/repo.git", True),  # Valid Git URL
+        ("sip://example.com", True),  # Valid SIP URL
+        ("sips://example.com", True),  # Valid SIPS URL
+        ("invalid://example.com", False),  # Invalid URL with unknown scheme
+    ],
+)
+def test_is_url(url, expected):
+    assert is_url(url) == expected
+
+
+@pytest.mark.parametrize(
+    ("url", "expected"),
+    [
+        ("s3://bucket/path/to/file", True),  # Valid fsspec URL
+        ("ftp://example.com/file.txt", True),  # Valid fsspec URL
+        ("gs://bucket/path/to/file", True),  # Valid fsspec URL
+        ("http://example.com/file.txt", False),  # Invalid URL (HTTP)
+        ("https://example.com/file.txt", False),  # Invalid URL (HTTPS)
+        ("file://localhost/path/to/file", True),  # Valid fsspec URL
+        ("mailto:user@example.com", False),  # Invalid URL
+        (
+            "ftp://user:pass@example.com/path/to/file",
+            True,
+        ),  # Valid fsspec URL with credentials
+        ("ftp://example.com", True),  # Valid fsspec URL without file
+        ("", False),  # Empty string (invalid URL)
+        ("invalid://path/to/file", True),  # Invalid scheme but valid format
+        ("http://localhost:8000", False),  # Invalid URL (HTTP with port)
+        (
+            "https://192.168.1.1/path/to/file",
+            False,
+        ),  # Invalid URL (HTTPS with IP address)
+        (
+            "file:/path/to/file",
+            False,
+        ),  # Invalid URL (missing double slashes after file:)
+    ],
+)
+def test_is_fsspec_url(url, expected):
+    assert is_fsspec_url(url) == expected
+
+
 # TODO(kszucs): add tests for promote_list and promote_tuple