update

fugue/test/plugins.py

@@ -29,10 +29,6 @@ def _get_all_ini_conf() -> Dict[str, Any]:
     return _FUGUE_TEST_ALL_INI_CONF
 
 
-def _get_all_backends() -> Dict[str, Type["FugueTestBackend"]]:
-    return _FUGUE_TEST_BACKENDS
-
-
 @run_once
 def _load_all_backends() -> None:
     from fugue.constants import FUGUE_ENTRYPOINT
@@ -173,11 +169,7 @@ def extract_conf(
 @contextmanager
 def _make_backend_context(obj: Any, session: Any) -> Iterator[Any]:
     _load_all_backends()
-    if isinstance(obj, str):
-        key = obj
-        extra_conf: Dict[str, Any] = {}
-    else:
-        key, extra_conf = obj
+    key, extra_conf = _parse_backend(obj)
     assert_or_throw(
         key in _FUGUE_TEST_BACKENDS,
         lambda: ValueError(

fugue_ibis/dataframe.py

@@ -78,7 +78,7 @@ def num_partitions(self) -> int:
 
     @property
     def columns(self) -> List[str]:
-        return self._table.columns
+        return _get_ibis_columns(self._table)
 
     def peek_array(self) -> List[Any]:
         res = self._to_local_df(self._table.head(1)).as_array()
@@ -202,7 +202,7 @@ def _drop_ibis_columns(df: IbisTable, columns: List[str]) -> IbisTable:
 
 
 @rename.candidate(lambda df, *args, **kwargs: isinstance(df, IbisTable))
-def _rename_dask_dataframe(df: IbisTable, columns: Dict[str, Any]) -> IbisTable:
+def _rename_ibis_table(df: IbisTable, columns: Dict[str, Any]) -> IbisTable:
     _assert_no_missing(df, columns.keys())
     old_names = df.columns
     new_names = [columns.get(name, name) for name in old_names]

fugue_spark/execution_engine.py

@@ -492,11 +492,6 @@ def persist(
         res.reset_metadata(df.metadata)
         return res
 
-    def register(self, df: DataFrame, name: str) -> SparkDataFrame:
-        sdf = self._to_spark_df(df)
-        sdf.native.createOrReplaceTempView(name)
-        return sdf
-
     def join(
         self,
         df1: DataFrame,

setup.py

@@ -35,18 +35,18 @@ def get_version() -> str:
         "adagio>=0.2.4",
         # sql dependencies
         "qpd>=0.4.4",
-        "fugue-sql-antlr>=0.1.6",
+        "fugue-sql-antlr>=0.2.0",
         "sqlglot",
         "jinja2",
     ],
     extras_require={
         "sql": [
             "qpd>=0.4.4",
-            "fugue-sql-antlr>=0.1.6",
+            "fugue-sql-antlr>=0.2.0",
             "sqlglot",
             "jinja2",
         ],
-        "cpp_sql_parser": ["fugue-sql-antlr[cpp]>=0.1.6"],
+        "cpp_sql_parser": ["fugue-sql-antlr[cpp]>=0.2.0"],
         "spark": ["pyspark>=3.1.1"],
         "dask": [
             "dask[distributed,dataframe]>=2023.5.0",
@@ -64,7 +64,7 @@ def get_version() -> str:
         "all": [
             "sqlglot",
             "jinja2",
-            "fugue-sql-antlr[cpp]>=0.1.6",
+            "fugue-sql-antlr[cpp]>=0.2.0",
             "pyspark>=3.1.1",
             "dask[distributed,dataframe]>=2023.5.0",
             "dask-sql",

tests/fugue/workflow/test_workflow.py

@@ -39,6 +39,7 @@ def test_worflow_dataframes():
     dfs1 = WorkflowDataFrames(a=df1, b=df2)
     assert dfs1["a"] is df1
     assert dfs1["b"] is df2
+    assert dfs1.workflow is dag1
 
     dfs2 = WorkflowDataFrames(dfs1, aa=df1, bb=df2)
     assert 4 == len(dfs2)

tests/fugue_ibis/test_dataframe.py

@@ -3,11 +3,12 @@
 from typing import Any
 
 import pandas as pd
+import pytest
 
 import fugue.api as fe
 import fugue.test as ft
 from fugue import ArrowDataFrame
-from fugue_duckdb.dataframe import DuckDataFrame
+from fugue.exceptions import FugueDataFrameOperationError
 from fugue_test.dataframe_suite import DataFrameTests
 
 from .mock.dataframe import MockDuckDataFrame
@@ -56,3 +57,23 @@ def test_deep_nested_types(self):
 
     def test_list_type(self):
         pass
+
+    def test_native_table(self):
+        df = self.df([["x", 1]], "a:str,b:int").native
+        assert fe.get_schema(fe.rename(df, dict())) == "a:str,b:int"
+        assert fe.get_schema(fe.rename(df, dict(a="c"))) == "c:str,b:int"
+
+        with pytest.raises(Exception):
+            fe.rename(df, dict(a="b"))
+
+        with pytest.raises(FugueDataFrameOperationError):
+            fe.rename(df, dict(x="y"))
+
+        assert fe.get_schema(fe.drop_columns(df, [])) == "a:str,b:int"
+        assert fe.get_schema(fe.drop_columns(df, ["a"])) == "b:int"
+
+        with pytest.raises(FugueDataFrameOperationError):
+            fe.get_schema(fe.drop_columns(df, ["a", "b"]))
+
+        with pytest.raises(FugueDataFrameOperationError):
+            fe.get_schema(fe.drop_columns(df, ["a", "c"]))