API: pre_execute now accepts (Node, Client) as a dispatching type

to allow backends to operate on the physical tables

ibis/pandas/core.py

@@ -10,8 +10,11 @@
 
 import toolz
 
+from ibis.common import IbisError
 import ibis.expr.types as ir
 import ibis.expr.datatypes as dt
+from ibis.client import find_backend
+
 
 import ibis.pandas.aggcontext as agg_ctx
 from ibis.pandas.dispatch import (
@@ -85,8 +88,22 @@ def execute_with_scope(expr, scope, context=None, **kwargs):
     result : scalar, pd.Series, pd.DataFrame
     """
     op = expr.op()
-    pre_loaded_scope = pre_execute(op, scope=scope, context=context, **kwargs)
-    scope = toolz.merge(scope, pre_loaded_scope)
+
+    # TODO(jreback), might be better to pass this thru as a kwargs
+    # or cache it on the op?
+    try:
+        backends = find_backend(expr)
+        if not isinstance(backends, list):
+            backends = [backends]
+        for client in backends:
+            pre_loaded_scope = pre_execute(op,
+                                           client,
+                                           scope=scope,
+                                           context=context,
+                                           **kwargs)
+            scope = toolz.merge(scope, pre_loaded_scope)
+    except IbisError:
+        pass
 
     # base case: our op has been computed (or is a leaf data node), so
     # return the corresponding value

ibis/pandas/dispatch.py

@@ -29,6 +29,6 @@ def data_preload_default(node, data, **kwargs):
 
 
 # Default returns an empty scope
-@pre_execute.register(object)
-def pre_execute_default(node, **kwargs):
+@pre_execute.register(object, object)
+def pre_execute_default(node, client, **kwargs):
     return {}

ibis/pandas/tests/test_core.py

@@ -12,7 +12,7 @@
 from ibis.pandas.execution import (
     execute, execute_node, execute_first
 )  # noqa: E402
-from ibis.pandas.client import PandasTable  # noqa: E402
+from ibis.pandas.client import PandasTable, PandasClient  # noqa: E402
 from ibis.pandas.core import data_preload, pre_execute  # noqa: E402
 from multipledispatch.conflict import ambiguities  # noqa: E402
 
@@ -72,20 +72,20 @@ def data_preload_check_a_thing(_, df, **kwargs):
     data_preload._cache.clear()
 
 
-def test_pre_execute(ibis_table, dataframe):
+def test_pre_execute_basic(ibis_table, dataframe):
     """
     Test that pre_execute has intercepted execution and provided its own
     scope dict
     """
-    @pre_execute.register(ir.Node)
-    def pre_execute_test(op, **kwargs):
+    @pre_execute.register(ir.Node, PandasClient)
+    def pre_execute_test(op, client, **kwargs):
         df = dataframe.assign(plain_int64=dataframe['plain_int64'] + 1)
         return {op: df}
 
     result = ibis_table.execute()
     tm.assert_frame_equal(
         result, dataframe.assign(plain_int64=dataframe['plain_int64'] + 1))
 
-    del pre_execute.funcs[ir.Node, ]
+    del pre_execute.funcs[(ir.Node, PandasClient)]
     pre_execute.reorder()
     pre_execute._cache.clear()