feat(flink): make schema arg optional for read_parquet()

ibis/backends/flink/__init__.py

@@ -677,9 +677,20 @@ def _read_file(
             If `schema` is None.
         """
         if schema is None:
-            raise ValueError(
-                f"`schema` must be explicitly provided when calling `read_{file_type}`"
-            )
+            if file_type == "parquet":
+                import pyarrow as pa
+
+                try:
+                    pyarrow_schema = pa.parquet.read_schema(path)
+                except FileNotFoundError:
+                    raise ValueError(f"No file found at {path}")
+
+                schema = sch.Schema.from_pyarrow(pyarrow_schema)
+
+            else:
+                raise ValueError(
+                    f"`schema` must be explicitly provided when calling `read_{file_type}`"
+                )
 
         table_name = table_name or gen_name(f"read_{file_type}")
         tbl_properties = {

ibis/backends/flink/tests/test_ddl.py

@@ -2,7 +2,6 @@
 
 import os
 import tempfile
-from pathlib import Path
 
 import pandas as pd
 import pyarrow as pa
@@ -511,19 +510,20 @@ def test_read_csv(con, awards_players_schema, csv_source_configs, table_name):
 
 
 @pytest.mark.parametrize("table_name", ["new_table", None])
-def test_read_parquet(con, data_dir, tmp_path, table_name):
-    fname = Path("functional_alltypes.parquet")
-    fname = Path(data_dir) / "parquet" / fname.name
+@pytest.mark.parametrize("schema", [_functional_alltypes_schema, None])
+def test_read_parquet(con, data_dir, table_name, schema):
+    path = data_dir.joinpath("parquet", "functional_alltypes.parquet")
     table = con.read_parquet(
-        path=tmp_path / fname.name,
-        schema=_functional_alltypes_schema,
+        path=path,
+        schema=schema,
         table_name=table_name,
     )
 
     if table_name is None:
         table_name = table.get_name()
     assert table_name in con.list_tables()
-    assert table.schema() == _functional_alltypes_schema
+    if schema:
+        assert table.schema() == schema
 
     con.drop_table(table_name)
     assert table_name not in con.list_tables()

ibis/backends/tests/test_register.py

@@ -12,6 +12,7 @@
 
 import ibis
 from ibis.backends.conftest import TEST_TABLES
+from ibis.backends.tests.errors import Py4JJavaError
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -398,8 +399,8 @@ def test_register_garbage(con, monkeypatch):
 @pytest.mark.notyet(["impala", "mssql", "mysql", "postgres", "sqlite", "trino"])
 @pytest.mark.notimpl(
     ["flink"],
-    raises=ValueError,
-    reason="read_parquet() missing required argument: 'schema'",
+    raises=Py4JJavaError,
+    reason="Parquet format jar for Flink is not in the path",
 )
 def test_read_parquet(con, tmp_path, data_dir, fname, in_table_name):
     pq = pytest.importorskip("pyarrow.parquet")
@@ -435,7 +436,7 @@ def ft_data(data_dir):
 @pytest.mark.notimpl(
     ["flink"],
     raises=ValueError,
-    reason="read_parquet() missing required argument: 'schema'",
+    reason="read_parquet() does not support glob",
 )
 def test_read_parquet_glob(con, tmp_path, ft_data):
     pq = pytest.importorskip("pyarrow.parquet")