fix ray tests and coverage

fugue-project · Dec 28, 2022 · 49d3724 · 49d3724
1 parent 3412839
commit 49d3724
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 6 deletions.
diff --git a/fugue/column/sql.py b/fugue/column/sql.py
@@ -261,7 +261,7 @@ def where(self, condition: ColumnExpr, table: str) -> Iterable[Tuple[bool, str]]
             lambda: ValueError(f"{condition} has aggregation functions"),
         )
         cond = self.generate(condition.alias(""))
-        yield (False, "SELECT * FROM ")
+        yield (False, "SELECT * FROM")
         yield (True, table)
         yield (False, f"WHERE {cond}")
 
@@ -312,14 +312,14 @@ def _having(as_where: bool = False) -> str:
         if len(columns.literals) == 0:
             expr = ", ".join(self.generate(x) for x in columns.all_cols)
             if len(columns.group_keys) == 0:
-                yield (False, f"SELECT {distinct}{expr} FROM ")
+                yield (False, f"SELECT {distinct}{expr} FROM")
                 yield (True, table)
                 yield (False, _where())
                 yield (False, _having())
                 return
             else:
                 keys = ", ".join(self.generate(x) for x in columns.group_keys)
-                yield (False, f"SELECT {distinct}{expr} FROM ")
+                yield (False, f"SELECT {distinct}{expr} FROM")
                 yield (True, table)
                 yield (False, _where())
                 yield (False, f"GROUP BY {keys}")

diff --git a/fugue/workflow/input.py b/fugue/workflow/input.py
@@ -1,7 +1,7 @@
 from typing import Type
 
 
-def register_raw_df_type(df_type: Type) -> None:
+def register_raw_df_type(df_type: Type) -> None:  # pragma: no cover
     """TODO: This function is to be removed before 0.9.0
 
     .. deprecated:: 3.1

diff --git a/fugue_ray/dataframe.py b/fugue_ray/dataframe.py
@@ -90,7 +90,7 @@ def __init__(  # noqa: C901
                 schema = df.schema
             metadata = None if not df.has_metadata else df.metadata
         else:
-            raise ValueError(f"{df} is incompatible with DaskDataFrame")
+            raise ValueError(f"{df} is incompatible with RayDataFrame")
         rdf, schema = self._apply_schema(rdf, schema, internal_schema)
         super().__init__(schema)
         self._native = rdf

diff --git a/fugue_ray/execution_engine.py b/fugue_ray/execution_engine.py
@@ -1,7 +1,7 @@
 from typing import Any, Callable, Dict, List, Optional, Union
 
 import pyarrow as pa
-from duckdb import DuckDBPyConnection
+from duckdb import DuckDBPyConnection, DuckDBPyRelation
 from triad import Schema, assert_or_throw, to_uuid
 from triad.utils.threading import RunOnce
 
@@ -269,6 +269,12 @@ def _to_auto_df(self, df: Any, schema: Any = None) -> DataFrame:
                 ValueError("schema must be None when df is a DataFrame"),
             )
             return df
+        if isinstance(df, DuckDBPyRelation):
+            assert_or_throw(
+                schema is None,
+                ValueError("schema must be None when df is a DuckDBPyRelation"),
+            )
+            return DuckDataFrame(df)
         return RayDataFrame(df, schema)
 
     def _get_remote_args(self) -> Dict[str, Any]: