unify piping with fallback

Author: JohnMount

Examples/Arrow/Arrow.ipynb

@@ -1079,7 +1079,7 @@
      "text": [
       "[\n",
       " 'd2':\n",
-      "  [ ngroup, x, g, v ]\n",
+      "  [ g, ngroup, x, v ]\n",
       "   ->\n",
       "  [ g, ngroup, row_number, shift_v, v, x ]\n",
       "]\n",

Examples/Arrow/ArrowSetSystem.ipynb

@@ -81,7 +81,7 @@
     {
      "data": {
       "text/plain": [
-       "(TableDescription(table_name=\"table_a\", column_names=[\"a\", \"b\"]).extend({\"c\": \"a + b\"}))"
+       "(TableDescription(table_name=\"table_a\", column_names=[\"a\", \"b\"]).extend({\"c\": \"a + b\"}))\n"
       ]
      },
      "execution_count": 2,
@@ -158,7 +158,7 @@
        "    TableDescription(table_name=\"table_b\", column_names=[\"a\", \"b\", \"c\"]).rename_columns(\n",
        "        {\"x\": \"a\"}\n",
        "    )\n",
-       ")"
+       ")\n"
       ]
      },
      "execution_count": 4,
@@ -233,7 +233,7 @@
        "    TableDescription(table_name=\"table_a\", column_names=[\"a\", \"b\"])\n",
        "    .extend({\"c\": \"a + b\"})\n",
        "    .rename_columns({\"x\": \"a\"})\n",
-       ")"
+       ")\n"
       ]
      },
      "execution_count": 6,
@@ -742,7 +742,7 @@
        "    TableDescription(table_name=\"table_c\", column_names=[\"a\", \"b\", \"c\"])\n",
        "    .extend({\"x\": \"a + b\"})\n",
        "    .select_columns([\"x\"])\n",
-       ")"
+       ")\n"
       ]
      },
      "execution_count": 14,
@@ -1036,7 +1036,7 @@
        "    TableDescription(\n",
        "        table_name=\"data_frame\", column_names=[\"a\", \"b\", \"d\", \"x\"]\n",
        "    ).drop_columns([\"d\", \"x\"])\n",
-       ")"
+       ")\n"
       ]
      },
      "execution_count": 19,
@@ -1073,7 +1073,7 @@
        "    .drop_columns([\"d\", \"x\"])\n",
        "    .extend({\"c\": \"a + b\"})\n",
        "    .rename_columns({\"x\": \"a\"})\n",
-       ")"
+       ")\n"
       ]
      },
      "execution_count": 20,

Examples/Arrow/CDesign.ipynb

@@ -350,7 +350,7 @@
        "    TableDescription(table_name=\"data_frame\", column_names=[\"x\", \"y\"]).extend(\n",
        "        {\"z\": \"x.mean()\"}, partition_by=[\"y\"]\n",
        "    )\n",
-       ")"
+       ")\n"
       ]
      },
      "execution_count": 2,
@@ -481,7 +481,7 @@
        "    TableDescription(table_name=\"data_frame\", column_names=[\"x\", \"y\"])\n",
        "    .extend({\"z\": \"x.mean()\"}, partition_by=[\"y\"])\n",
        "    .extend({\"ratio\": \"y / x\"})\n",
-       ")"
+       ")\n"
       ]
      },
      "execution_count": 4,
@@ -773,7 +773,7 @@
        "    TableDescription(table_name=\"data_frame\", column_names=[\"x\", \"y\"])\n",
        "    .extend({\"z\": \"x.mean()\"}, partition_by=[\"y\"])\n",
        "    .extend({\"ratio\": \"y / x\"})\n",
-       ")"
+       ")\n"
       ]
      },
      "execution_count": 11,
@@ -915,7 +915,7 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "Caught: extra incoming columns: {'z', 'ratio'}\n"
+      "Caught: extra incoming columns: {'ratio', 'z'}\n"
      ]
     }
    ],
@@ -1065,7 +1065,7 @@
        "    TableDescription(table_name=\"data_frame\", column_names=[\"x\", \"y\"]).extend(\n",
        "        {\"x\": \"x + 1\", \"y\": \"9\"}\n",
        "    )\n",
-       ")"
+       ")\n"
       ]
      },
      "execution_count": 19,

data_algebra/arrow.py

@@ -2,9 +2,10 @@
 
 import data_algebra.data_ops
 import data_algebra.flow_text
+from data_algebra.shift_pipe_action import ShiftPipeAction
 
 
-class Arrow(abc.ABC):
+class Arrow(ShiftPipeAction):
     """
     Arrow from category theory: see Steve Awody,
     "Category Theory, 2nd Edition", Oxford Univ. Press, 2010 pg. 4.
@@ -13,7 +14,7 @@ class Arrow(abc.ABC):
     """
 
     def __init__(self):
-        pass
+        ShiftPipeAction.__init__(self)
 
     @abc.abstractmethod
     def dom(self):
@@ -23,28 +24,16 @@ def dom(self):
     def cod(self):
         """return co-domain, object at head of arrow"""
 
-    @abc.abstractmethod
-    def apply_to(self, b):
-        """apply_to b, compose arrows (right to left)"""
-
     # noinspection PyPep8Naming
     @abc.abstractmethod
-    def act_on(self, X):
-        """act on X, must associate with composition"""
+    def act_on(self, b):
+        """act on b, must associate with composition"""
 
     # noinspection PyPep8Naming
     def transform(self, X):
         """transform X, may or may not associate with composition"""
         return self.act_on(X)
 
-    def __rshift__(self, other):  # override self >> other
-        return other.apply_to(self)
-
-    def __rrshift__(self, other):  # override other >> self
-        if isinstance(other, Arrow):
-            return self.apply_to(other)
-        return self.act_on(other)
-
 
 class DataOpArrow(Arrow):
     """
@@ -79,39 +68,38 @@ def get_feature_names(self):
         cp = self.outgoing_columns.copy()
         return cp
 
-    def apply_to(self, b):
+    def act_on(self, b):
         """replace self input table with b"""
         if isinstance(b, data_algebra.data_ops.ViewRepresentation):
             b = DataOpArrow(b)
-        assert isinstance(b, DataOpArrow)
-        # check categorical arrow composition conditions
-        missing = set(self.incoming_columns) - set(b.outgoing_columns)
-        if len(missing) > 0:
-            raise ValueError("missing required columns: " + str(missing))
-        excess = set(b.outgoing_columns) - set(self.incoming_columns)
-        if len(excess) > 0:
-            raise ValueError("extra incoming columns: " + str(excess))
-        new_pipeline = self.pipeline.replace_leaves({self.free_table_key: b.pipeline})
-        new_pipeline.get_tables()  # check tables are compatible
-        res = DataOpArrow(
-            pipeline=new_pipeline,
-            free_table_key=b.free_table_key,
-        )
-        return res
-
-    # noinspection PyPep8Naming
-    def act_on(self, X):
+        if isinstance(b, DataOpArrow):
+            # check categorical arrow composition conditions
+            missing = set(self.incoming_columns) - set(b.outgoing_columns)
+            if len(missing) > 0:
+                raise ValueError("missing required columns: " + str(missing))
+            excess = set(b.outgoing_columns) - set(self.incoming_columns)
+            if len(excess) > 0:
+                raise ValueError("extra incoming columns: " + str(excess))
+            new_pipeline = self.pipeline.replace_leaves({self.free_table_key: b.pipeline})
+            new_pipeline.get_tables()  # check tables are compatible
+            res = DataOpArrow(
+                pipeline=new_pipeline,
+                free_table_key=b.free_table_key,
+            )
+            return res
+        if isinstance(b, ShiftPipeAction):
+            return b.act_on(self)  # fall back
         # assume a pandas.DataFrame compatible object
         # noinspection PyUnresolvedReferences
-        cols = set(X.columns)
+        cols = set(b.columns)
         missing = set(self.incoming_columns) - cols
         if len(missing) > 0:
             raise ValueError("missing required columns: " + str(missing))
         excess = cols - set(self.incoming_columns)
         assert len(excess) == 0
         if len(excess) > 0:
-            X = X[self.incoming_columns]
-        return self.pipeline.act_on(X)
+            b = b[self.incoming_columns]
+        return self.pipeline.act_on(b)
 
     def dom(self):
         return DataOpArrow(

data_algebra/cdata.py

@@ -10,6 +10,7 @@
 import data_algebra.data_model
 import data_algebra.util
 import data_algebra.data_ops
+from data_algebra.shift_pipe_action import ShiftPipeAction
 
 
 def _str_list_to_html(lst: Iterable[str]) -> str:
@@ -257,7 +258,7 @@ def map_from_rows(self):
         return RecordMap(blocks_out=self, strict=self.strict)
 
 
-class RecordMap:
+class RecordMap(ShiftPipeAction):
     """
     Class for specifying general record to record transforms.
     """
@@ -280,6 +281,7 @@ def __init__(
         :param blocks_out: outgoing record specification, None for row-records.
         :param strict: if True insist block be strict, and in and out blocks agree on row-form columns.∂
         """
+        ShiftPipeAction.__init__(self)
         assert isinstance(strict, bool)
         self.strict = strict
         if blocks_in is not None:
@@ -426,10 +428,6 @@ def transform(
                 blocks_out=self.blocks_out,
             )
         return X
-    
-    def act_on(self, X):
-        assert isinstance(X, data_algebra.data_ops.ViewRepresentation)
-        return X.convert_records(self)
 
     def compose(self, other):
         """
@@ -534,12 +532,16 @@ def inverse(self):
         """
         assert self.strict
         return RecordMap(blocks_in=self.blocks_out, blocks_out=self.blocks_in, strict=True)
-
-    def __rshift__(self, other):  # override self >> other
-        return self.act_on(other)
-
-    def __rrshift__(self, other):  # override other >> self
-        return self.transform(other)
+    
+    def act_on(self, b):
+        if isinstance(b, RecordMap):
+            self.compose(b)
+        if isinstance(b, data_algebra.data_ops.ViewRepresentation):
+            return b.convert_records(self)
+        if isinstance(b, ShiftPipeAction):
+            return b.act_on(self)  # fall back to peer's action
+        # assume table like
+        return self.transform(b)
 
     def fmt(self) -> str:
         """Format for informal presentation."""

data_algebra/data_ops.py

@@ -24,6 +24,7 @@
     ordered_union,
 )
 import data_algebra.util
+from data_algebra.shift_pipe_action import ShiftPipeAction
 
 
 _have_black = False
@@ -411,36 +412,36 @@ def __ne__(self, other):
 
     # composition
 
-    # noinspection PyPep8Naming
-    def act_on(self, X, *, data_model=None):
+    def act_on(self, b):
         """
-        apply self to X, must associate with composition
+        apply self to b, must associate with composition
         Operator is strict about column names.
 
-        :param X: input data frame
-        :param data_model implementation to use
-        :return: transformed result
+        :param b: input data frame
+        :return: transformed or composed result
         """
         tables = self.get_tables()
-        if isinstance(X, ViewRepresentation):
+        if isinstance(b, ViewRepresentation):
             # insert to only table or if more than one, table with matching key
             if len(tables) == 1:
                 key = list(tables.keys())[0]
             else:
-                key = X.key
+                key = b.key
             assert isinstance(key, str)
             old = tables[key]
-            assert set(X.column_names) == set(old.column_names)  # this is defending associativity of composition against table narrowing
-            return self.replace_leaves({key: X})
+            assert set(b.column_names) == set(old.column_names)  # this is defending associativity of composition against table narrowing
+            return self.replace_leaves({key: b})
+        # see if b is ShiftPipeAction, so it can handle the mapping (using fact data is not a ShiftPipeAction instance)
+        if isinstance(b, ShiftPipeAction):
+            return b.act_on(self)
         # assume a table
         assert len(tables) == 1
         key = list(tables.keys())[0]
         assert isinstance(key, str)
         old = tables[key]
-        assert set(X.columns) == set(old.column_names)
+        assert set(b.columns) == set(old.column_names)
         return self.transform(
-            X=X,
-            data_model=data_model,
+            b,
             strict=True,
         )
 

data_algebra/data_ops_types.py

@@ -8,6 +8,7 @@
 import data_algebra.expr_rep
 import data_algebra.cdata
 import data_algebra.OrderedSet
+from data_algebra.shift_pipe_action import ShiftPipeAction
 
 
 class MethodUse(NamedTuple):
@@ -19,12 +20,13 @@ class MethodUse(NamedTuple):
     is_ordered: bool = False
 
 
-class OperatorPlatform(abc.ABC):
+class OperatorPlatform(ShiftPipeAction):
     """Abstract class representing ability to apply data_algebra operations."""
 
     node_name: str
 
     def __init__(self, *, node_name: str):
+        ShiftPipeAction.__init__(self)
         assert isinstance(node_name, str)
         self.node_name = node_name
 
@@ -73,24 +75,6 @@ def replace_leaves(self, replacement_map: Dict[str, Any]):
         :return: new operator DAG
         """
 
-    # noinspection PyPep8Naming
-    @abc.abstractmethod
-    def act_on(self, X, *, data_model=None):
-        """
-        apply self to X, must associate with composition
-        Operator is strict about column names.
-
-        :param X: input data frame
-        :param data_model implementation to use
-        :return: transformed result
-        """
-
-    def __rshift__(self, other):  # override self >> other
-        return other.act_on(self)
-
-    def __rrshift__(self, other):  # override other >> self
-        return self.act_on(other)
-
     # imitate a method
     def use(self, user_function, *args, **kwargs):
         """