work on replacement test

rebuild and retest

Author: JohnMount

build/lib/data_algebra/arrow.py

@@ -137,9 +137,7 @@ def apply_to(self, b):
                 raise ValueError("forbidden incoming columns: " + str(excess))
             if self.strict:
                 raise ValueError("extra incoming columns: " + str(excess))
-        new_pipeline = self.pipeline.apply_to(
-            b.pipeline, target_table_key=self.free_table_key
-        )
+        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,

build/lib/data_algebra/data_ops.py

@@ -1013,17 +1013,19 @@ def forbidden_columns(
             forbidden = set()
         return {self.key: set(forbidden)}
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
-        if (target_table_key is None) or (target_table_key == self.key):
-            # replace table with a
-            return a
+        assert isinstance(replacement_map, dict)
+        try:
+            return replacement_map[self.key]
+        except KeyError:
+            pass
         # copy self
         r = TableDescription(
             table_name=self.table_name,
@@ -1374,16 +1376,17 @@ def __init__(
             self, column_names=column_names, sources=[source], node_name="ExtendNode"
         )
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
+        assert isinstance(replacement_map, dict)
         new_sources = [
-            s.apply_to(a, target_table_key=target_table_key) for s in self.sources
+            s.replace_leaves(replacement_map) for s in self.sources
         ]
         return new_sources[0].extend_parsed_(
             parsed_ops=self.ops,
@@ -1635,16 +1638,17 @@ def get_method_uses_(self, methods_seen: Set[MethodUse]) -> None:
                 MethodUse(k, is_project=True, is_windowed=False, is_ordered=False)
             )
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
+        assert isinstance(replacement_map, dict)
         new_sources = [
-            s.apply_to(a, target_table_key=target_table_key) for s in self.sources
+            s.replace_leaves(replacement_map) for s in self.sources
         ]
         return new_sources[0].project_parsed_(
             parsed_ops=self.ops, group_by=self.group_by
@@ -1773,16 +1777,17 @@ def get_method_uses_(self, methods_seen: Set[MethodUse]) -> None:
                 MethodUse(k, is_project=False, is_windowed=False, is_ordered=False)
             )
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
+        assert isinstance(replacement_map, dict)
         new_sources = [
-            s.apply_to(a, target_table_key=target_table_key) for s in self.sources
+            s.replace_leaves(replacement_map) for s in self.sources
         ]
         return new_sources[0].select_rows_parsed_(parsed_ops=self.ops)
 
@@ -1888,16 +1893,17 @@ def forbidden_columns(
         forbidden = set(forbidden).intersection(self.column_selection)
         return self.sources[0].forbidden_columns(forbidden=forbidden)
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
+        assert isinstance(replacement_map, dict)
         new_sources = [
-            s.apply_to(a, target_table_key=target_table_key) for s in self.sources
+            s.replace_leaves(replacement_map) for s in self.sources
         ]
         return new_sources[0].select_columns(columns=self.column_selection)
 
@@ -2000,16 +2006,17 @@ def forbidden_columns(
         forbidden = set(forbidden) - set(self.column_deletions)
         return self.sources[0].forbidden_columns(forbidden=forbidden)
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
+        assert isinstance(replacement_map, dict)
         new_sources = [
-            s.apply_to(a, target_table_key=target_table_key) for s in self.sources
+            s.replace_leaves(replacement_map) for s in self.sources
         ]
         return new_sources[0].drop_columns(column_deletions=self.column_deletions)
 
@@ -2101,16 +2108,17 @@ def __init__(self, source, columns, *, reverse=None, limit=None):
             node_name="OrderRowsNode",
         )
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
+        assert isinstance(replacement_map, dict)
         new_sources = [
-            s.apply_to(a, target_table_key=target_table_key) for s in self.sources
+            s.replace_leaves(replacement_map) for s in self.sources
         ]
         return new_sources[0].order_rows(
             columns=self.order_columns, reverse=self.reverse, limit=self.limit
@@ -2246,16 +2254,17 @@ def forbidden_columns(
         new_forbidden.update(self.new_columns)
         return self.sources[0].forbidden_columns(forbidden=new_forbidden)
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
+        assert isinstance(replacement_map, dict)
         new_sources = [
-            s.apply_to(a, target_table_key=target_table_key) for s in self.sources
+            s.replace_leaves(replacement_map) for s in self.sources
         ]
         return new_sources[0].map_columns(column_remapping=self.column_remapping)
 
@@ -2383,16 +2392,17 @@ def forbidden_columns(
         new_forbidden.update(self.new_columns)
         return self.sources[0].forbidden_columns(forbidden=new_forbidden)
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
+        assert isinstance(replacement_map, dict)
         new_sources = [
-            s.apply_to(a, target_table_key=target_table_key) for s in self.sources
+            s.replace_leaves(replacement_map) for s in self.sources
         ]
         return new_sources[0].rename_columns(column_remapping=self.column_remapping)
 
@@ -2536,16 +2546,17 @@ def __init__(
         if (self.jointype == "CROSS") and (len(self.on_a) != 0):
             raise ValueError("CROSS joins must have an empty 'on' list")
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
+        assert isinstance(replacement_map, dict)
         new_sources = [
-            s.apply_to(a, target_table_key=target_table_key) for s in self.sources
+            s.replace_leaves(replacement_map) for s in self.sources
         ]
         return new_sources[0].natural_join(
             b=new_sources[1], on=[(va, vb) for (va, vb) in zip(self.on_a, self.on_b)], jointype=self.jointype
@@ -2663,16 +2674,17 @@ def __init__(self, a, b, *, id_column="table_name", a_name="a", b_name="b"):
         self.a_name = a_name
         self.b_name = b_name
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
+        assert isinstance(replacement_map, dict)
         new_sources = [
-            s.apply_to(a, target_table_key=target_table_key) for s in self.sources
+            s.replace_leaves(replacement_map) for s in self.sources
         ]
         return new_sources[0].concat_rows(
             b=new_sources[1],
@@ -2778,16 +2790,17 @@ def __init__(self, *, source, record_map):
             node_name="ConvertRecordsNode",
         )
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
+        assert isinstance(replacement_map, dict)
         new_sources = [
-            s.apply_to(a, target_table_key=target_table_key) for s in self.sources
+            s.replace_leaves(replacement_map) for s in self.sources
         ]
         return new_sources[0].convert_records(record_map=self.record_map)
 
@@ -2910,15 +2923,26 @@ def __init__(
             node_name="SQLNode",
         )
 
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        Apply self to operator DAG a. Basic OperatorPlatform, composable API.
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+         :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
-        return self
+        assert isinstance(replacement_map, dict)
+        try:
+            return replacement_map[self.view_name]
+        except KeyError:
+            pass
+        # copy self
+        r = SQLNode(
+            sql=self.sql.copy(),
+            column_names=self.column_names.copy(),
+            view_name=self.view_name
+        )
+        return r
 
     def _equiv_nodes(self, other):
         if not isinstance(other, SQLNode):

build/lib/data_algebra/data_ops_types.py

@@ -79,50 +79,15 @@ def act_on(self, X, *, data_model=None):
         )
 
     @abc.abstractmethod
-    def apply_to(self, a, *, target_table_key=None):
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
         """
-        apply self to operator DAG a
+        Replace leaves of DAG
 
         :param a: operators to apply to
-        :param target_table_key: table key to replace with self, None counts as "match all"
+        :param replacement_map, table/sqlkeys mapped to replacement Operator platforms
         :return: new operator DAG
         """
 
-    def __rrshift__(self, other):  # override other >> self
-        """
-        override other >> self
-        self.apply_to/act_on(other)
-
-        :param other:
-        :return:
-        """
-        if isinstance(other, OperatorPlatform):
-            return self.apply_to(other)
-        return self.act_on(other)
-
-    def __rshift__(self, other):  # override self >> other
-        """
-        override self >> other
-        other.apply_to(self)
-
-        :param other:
-        :return:
-        """
-        # can't use type >> type if only __rrshift__ is defined (must have __rshift__ in this case)
-        if isinstance(other, OperatorPlatform):
-            return other.apply_to(self)
-        raise TypeError("unexpected type: " + str(type(other)))
-
-    # composition
-    def add(self, other):
-        """
-        other.apply_to(self)
-
-        :param other:
-        :return:
-        """
-        return other.apply_to(self)
-
     # imitate a method
     def use(self, user_function, *args, **kwargs):
         """
@@ -383,7 +348,7 @@ def get_feature_names(self, input_features=None):
             cp = cp + [f for f in input_features if f not in cp_set]
         return cp
 
-    # noinspection PyUnusedLocal,PyMethodMayBeStatic
+    # noinspection PyMethodMayBeStatic
     def get_params(self, deep=False):
         """sklearn interface, noop"""
         return dict()

build/lib/data_algebra/op_container.py

@@ -92,8 +92,8 @@ def act_on(self, X, *, data_model=None):
         self.set(self.ops.act_on(X=X, data_model=data_model))
         return self
 
-    def apply_to(self, a, *, target_table_key=None):
-        self.set(self.ops.apply_to(a=a, target_table_key=target_table_key))
+    def replace_leaves(self, replacement_map: Dict[str, Any]):
+        self.set(self.ops.replace_leaves(replacement_map))
         return self
 
     def __rrshift__(self, other):  # override other >> self
@@ -122,9 +122,9 @@ def columns_produced(self):
 
     # query generation
 
-    def to_near_sql_implementation_(self, db_model, *, using, temp_id_source):
+    def to_near_sql_implementation_(self, db_model, *, using, temp_id_source, sql_format_options=None):
         return self.ops.to_near_sql_implementation_(
-            db_model=db_model, using=using, temp_id_source=temp_id_source
+            db_model=db_model, using=using, temp_id_source=temp_id_source, sql_format_options=sql_format_options
         )
 
     # define builders for all non-initial ops types on base class