We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[FEA] apply joint TE on multiple columns after another op
**Is your feature request related to a problem? Please describe.**
We'd like to the the following pipeline with LambdaOp and Target Encoding. 

```
unixReviewTime = ColumnGroup(['unixReviewTime'])
unixReviewTime_toDatetime = unixReviewTime >> (lambda col: cudf.to_datetime(col, unit='s')) >> Rename(postfix="_toDatetime")
unixReviewTime_toDatetime_year = unixReviewTime_toDatetime >> (lambda col: col.dt.year) >> Rename(postfix="_year")

columns_1 = [[['userID'] + unixReviewTime_toDatetime_year]]
te_features = columns_1 >> nvt.ops.TargetEncoding("label", kfold=1, p_smooth=20)
```
However this does not work. 

We also tried `columns_1 = ColumnGroup(['user_id']) +  unixReviewTime_toDatetime_year` but this is not providing us with the joint multi-column TE as we want.

**Describe the solution you'd like**
We want to be able to do  joint multi-column TE on `userID` (from the input dataset) and `unixReviewTime_toDatetime_year` that is calculated from a lambda op. So our final output column should be `TE_userID_unixReviewTime_toDatetime_year_label`.

**Additional context**
We are using new api and conda environment with the latest NVT main branch.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `nvtabular/column_group.py`
Content:
```
1 #
2 # Copyright (c) 2020, NVIDIA CORPORATION.
3 #
4 # Licensed under the Apache License, Version 2.0 (the "License");
5 # you may not use this file except in compliance with the License.
6 # You may obtain a copy of the License at
7 #
8 #     http://www.apache.org/licenses/LICENSE-2.0
9 #
10 # Unless required by applicable law or agreed to in writing, software
11 # distributed under the License is distributed on an "AS IS" BASIS,
12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 # See the License for the specific language governing permissions and
14 # limitations under the License.
15 #
16 import collections.abc
17 
18 from dask.core import flatten
19 
20 from nvtabular.ops import LambdaOp, Operator
21 
22 
23 class ColumnGroup:
24     """A ColumnGroup is a group of columns that you want to apply the same transformations to.
25     ColumnGroup's can be transformed by shifting operators on to them, which returns a new
26     ColumnGroup with the transformations applied. This lets you define a graph of operations
27     that makes up your workflow
28 
29     Parameters
30     ----------
31     columns: list of (str or tuple of str)
32         The columns to select from the input Dataset. The elements of this list are strings
33         indicating the column names in most cases, but can also be tuples of strings
34         for feature crosses.
35     """
36 
37     def __init__(self, columns):
38         if isinstance(columns, str):
39             self.columns = [columns]
40         else:
41             self.columns = [_convert_col(col) for col in columns]
42         self.parents = []
43         self.children = []
44         self.op = None
45         self.kind = None
46         self.dependencies = None
47 
48     def __rshift__(self, operator):
49         """Transforms this ColumnGroup by applying an Operator
50 
51         Parameters
52         -----------
53         operators: Operator or callable
54 
55         Returns
56         -------
57         ColumnGroup
58         """
59         if isinstance(operator, type) and issubclass(operator, Operator):
60             # handle case where an operator class is passed
61             operator = operator()
62         elif callable(operator):
63             # implicit lambdaop conversion.
64             operator = LambdaOp(operator)
65 
66         if not isinstance(operator, Operator):
67             raise ValueError(f"Expected operator or callable, got {operator.__class__}")
68 
69         child = ColumnGroup(operator.output_column_names(self.columns))
70         child.parents = [self]
71         self.children.append(child)
72         child.op = operator
73 
74         dependencies = operator.dependencies()
75         if dependencies:
76             child.dependencies = set()
77             if not isinstance(dependencies, collections.abc.Sequence):
78                 dependencies = [dependencies]
79 
80             for dependency in dependencies:
81                 if not isinstance(dependency, ColumnGroup):
82                     dependency = ColumnGroup(dependency)
83                 dependency.children.append(child)
84                 child.parents.append(dependency)
85                 child.dependencies.add(dependency)
86 
87         return child
88 
89     def __add__(self, other):
90         """Adds columns from this ColumnGroup with another to return a new ColumnGroup
91 
92         Parameters
93         -----------
94         other: ColumnGroup or str or list of str
95 
96         Returns
97         -------
98         ColumnGroup
99         """
100         if isinstance(other, str):
101             other = ColumnGroup([other])
102         elif isinstance(other, collections.abc.Sequence):
103             other = ColumnGroup(other)
104 
105         # check if there are any columns with the same name in both column groups
106         overlap = set(self.columns).intersection(other.columns)
107         if overlap:
108             raise ValueError(f"duplicate column names found: {overlap}")
109 
110         child = ColumnGroup(self.columns + other.columns)
111         child.parents = [self, other]
112         child.kind = "+"
113         self.children.append(child)
114         other.children.append(child)
115         return child
116 
117     # handle the "column_name" + ColumnGroup case
118     __radd__ = __add__
119 
120     def __sub__(self, other):
121         """Adds columns from this ColumnGroup with another to return a new ColumnGroup
122 
123         Parameters
124         -----------
125         other: ColumnGroup or str or list of str
126 
127         Returns
128         -------
129         ColumnGroup
130         """
131         if isinstance(other, ColumnGroup):
132             to_remove = set(other.columns)
133         elif isinstance(other, str):
134             to_remove = {other}
135         elif isinstance(other, collections.abc.Sequence):
136             to_remove = set(other)
137         else:
138             raise ValueError(f"Expected ColumnGroup, str, or list of str. Got {other.__class__}")
139         new_columns = [c for c in self.columns if c not in to_remove]
140         child = ColumnGroup(new_columns)
141         child.parents = [self]
142         self.children.append(child)
143         child.kind = f"- {list(to_remove)}"
144         return child
145 
146     def __repr__(self):
147         output = " output" if not self.children else ""
148         return f"<ColumnGroup {self.label}{output}>"
149 
150     @property
151     def flattened_columns(self):
152         return list(flatten(self.columns, container=tuple))
153 
154     @property
155     def input_column_names(self):
156         """ returns the names of columns in the main chain """
157         dependencies = self.dependencies or set()
158         return [
159             col for parent in self.parents for col in parent.columns if parent not in dependencies
160         ]
161 
162     @property
163     def label(self):
164         if self.op:
165             return str(self.op.__class__.__name__)
166         elif self.kind:
167             return self.kind
168         elif not self.parents:
169             return f"input cols=[{self._cols_repr}]"
170         else:
171             return "??"
172 
173     @property
174     def _cols_repr(self):
175         cols = ", ".join(map(str, self.columns[:3]))
176         if len(self.columns) > 3:
177             cols += "..."
178         return cols
179 
180     @property
181     def graph(self):
182         return _to_graphviz(self)
183 
184 
185 def iter_nodes(nodes):
186     queue = nodes[:]
187     while queue:
188         current = queue.pop()
189         yield current
190         # TODO: deduplicate nodes?
191         for parent in current.parents:
192             queue.append(parent)
193 
194 
195 def _to_graphviz(column_group):
196     """ converts a columngroup to a GraphViz DiGraph object useful for display in notebooks """
197     from graphviz import Digraph
198 
199     column_group = _merge_add_nodes(column_group)
200     graph = Digraph()
201 
202     # get all the nodes from parents of this columngroup
203     # and add edges between each of them
204     allnodes = list(set(iter_nodes([column_group])))
205     node_ids = {v: str(k) for k, v in enumerate(allnodes)}
206     for node, nodeid in node_ids.items():
207         graph.node(nodeid, node.label)
208         for parent in node.parents:
209             graph.edge(node_ids[parent], nodeid)
210 
211     # add a single 'output' node representing the final state
212     output_node_id = str(len(allnodes))
213     graph.node(output_node_id, f"output cols=[{column_group._cols_repr}]")
214     graph.edge(node_ids[column_group], output_node_id)
215     return graph
216 
217 
218 def _merge_add_nodes(graph):
219     """ merges repeat '+' nodes, leading to nicer looking outputs """
220     # lets take a copy to avoid mutating the input
221     import copy
222 
223     graph = copy.deepcopy(graph)
224 
225     queue = [graph]
226     while queue:
227         current = queue.pop()
228         if current.kind == "+":
229             changed = True
230             while changed:
231                 changed = False
232                 parents = []
233                 for i, parent in enumerate(current.parents):
234                     if parent.kind == "+" and len(parent.children) == 1:
235                         changed = True
236                         # disconnect parent, point all the grandparents at current instead
237                         parents.extend(parent.parents)
238                         for grandparent in parent.parents:
239                             grandparent.children = [
240                                 current if child == parent else child
241                                 for child in grandparent.children
242                             ]
243                     else:
244                         parents.append(parent)
245                 current.parents = parents
246 
247         queue.extend(current.parents)
248 
249     return graph
250 
251 
252 def _convert_col(col):
253     if isinstance(col, (str, tuple)):
254         return col
255     elif isinstance(col, list):
256         return tuple(col)
257     else:
258         raise ValueError("Invalid column value for ColumnGroup: %s", col)
259
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/nvtabular/column_group.py b/nvtabular/column_group.py
--- a/nvtabular/column_group.py
+++ b/nvtabular/column_group.py
@@ -35,16 +35,36 @@
     """
 
     def __init__(self, columns):
-        if isinstance(columns, str):
-            self.columns = [columns]
-        else:
-            self.columns = [_convert_col(col) for col in columns]
         self.parents = []
         self.children = []
         self.op = None
         self.kind = None
         self.dependencies = None
 
+        if isinstance(columns, str):
+            columns = [columns]
+
+        # if any of the values we're passed are a columngroup
+        # we have to ourselves as a childnode in the graph.
+        if any(isinstance(col, ColumnGroup) for col in columns):
+            self.columns = []
+            self.kind = "[...]"
+            for col in columns:
+                if not isinstance(col, ColumnGroup):
+                    col = ColumnGroup(col)
+                else:
+                    # we can't handle nesting arbitrarily deep here
+                    # only accept non-nested (str) columns here
+                    if any(not isinstance(c, str) for c in col.columns):
+                        raise ValueError("Can't handle more than 1 level of nested columns")
+
+                col.children.append(self)
+                self.parents.append(col)
+                self.columns.append(tuple(col.columns))
+
+        else:
+            self.columns = [_convert_col(col) for col in columns]
+
     def __rshift__(self, operator):
         """Transforms this ColumnGroup by applying an Operator

{"golden_diff": "diff --git a/nvtabular/column_group.py b/nvtabular/column_group.py\n--- a/nvtabular/column_group.py\n+++ b/nvtabular/column_group.py\n@@ -35,16 +35,36 @@\n     \"\"\"\n \n     def __init__(self, columns):\n-        if isinstance(columns, str):\n-            self.columns = [columns]\n-        else:\n-            self.columns = [_convert_col(col) for col in columns]\n         self.parents = []\n         self.children = []\n         self.op = None\n         self.kind = None\n         self.dependencies = None\n \n+        if isinstance(columns, str):\n+            columns = [columns]\n+\n+        # if any of the values we're passed are a columngroup\n+        # we have to ourselves as a childnode in the graph.\n+        if any(isinstance(col, ColumnGroup) for col in columns):\n+            self.columns = []\n+            self.kind = \"[...]\"\n+            for col in columns:\n+                if not isinstance(col, ColumnGroup):\n+                    col = ColumnGroup(col)\n+                else:\n+                    # we can't handle nesting arbitrarily deep here\n+                    # only accept non-nested (str) columns here\n+                    if any(not isinstance(c, str) for c in col.columns):\n+                        raise ValueError(\"Can't handle more than 1 level of nested columns\")\n+\n+                col.children.append(self)\n+                self.parents.append(col)\n+                self.columns.append(tuple(col.columns))\n+\n+        else:\n+            self.columns = [_convert_col(col) for col in columns]\n+\n     def __rshift__(self, operator):\n         \"\"\"Transforms this ColumnGroup by applying an Operator\n", "issue": "[FEA] apply joint TE on multiple columns after another op\n**Is your feature request related to a problem? Please describe.**\r\nWe'd like to the the following pipeline with LambdaOp and Target Encoding. \r\n\r\n```\r\nunixReviewTime = ColumnGroup(['unixReviewTime'])\r\nunixReviewTime_toDatetime = unixReviewTime >> (lambda col: cudf.to_datetime(col, unit='s')) >> Rename(postfix=\"_toDatetime\")\r\nunixReviewTime_toDatetime_year = unixReviewTime_toDatetime >> (lambda col: col.dt.year) >> Rename(postfix=\"_year\")\r\n\r\ncolumns_1 = [[['userID'] + unixReviewTime_toDatetime_year]]\r\nte_features = columns_1 >> nvt.ops.TargetEncoding(\"label\", kfold=1, p_smooth=20)\r\n```\r\nHowever this does not work. \r\n\r\nWe also tried `columns_1 = ColumnGroup(['user_id']) +  unixReviewTime_toDatetime_year` but this is not providing us with the joint multi-column TE as we want.\r\n\r\n**Describe the solution you'd like**\r\nWe want to be able to do  joint multi-column TE on `userID` (from the input dataset) and `unixReviewTime_toDatetime_year` that is calculated from a lambda op. So our final output column should be `TE_userID_unixReviewTime_toDatetime_year_label`.\r\n\r\n**Additional context**\r\nWe are using new api and conda environment with the latest NVT main branch.\r\n\n", "before_files": [{"content": "#\n# Copyright (c) 2020, NVIDIA CORPORATION.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\nimport collections.abc\n\nfrom dask.core import flatten\n\nfrom nvtabular.ops import LambdaOp, Operator\n\n\nclass ColumnGroup:\n    \"\"\"A ColumnGroup is a group of columns that you want to apply the same transformations to.\n    ColumnGroup's can be transformed by shifting operators on to them, which returns a new\n    ColumnGroup with the transformations applied. This lets you define a graph of operations\n    that makes up your workflow\n\n    Parameters\n    ----------\n    columns: list of (str or tuple of str)\n        The columns to select from the input Dataset. The elements of this list are strings\n        indicating the column names in most cases, but can also be tuples of strings\n        for feature crosses.\n    \"\"\"\n\n    def __init__(self, columns):\n        if isinstance(columns, str):\n            self.columns = [columns]\n        else:\n            self.columns = [_convert_col(col) for col in columns]\n        self.parents = []\n        self.children = []\n        self.op = None\n        self.kind = None\n        self.dependencies = None\n\n    def __rshift__(self, operator):\n        \"\"\"Transforms this ColumnGroup by applying an Operator\n\n        Parameters\n        -----------\n        operators: Operator or callable\n\n        Returns\n        -------\n        ColumnGroup\n        \"\"\"\n        if isinstance(operator, type) and issubclass(operator, Operator):\n            # handle case where an operator class is passed\n            operator = operator()\n        elif callable(operator):\n            # implicit lambdaop conversion.\n            operator = LambdaOp(operator)\n\n        if not isinstance(operator, Operator):\n            raise ValueError(f\"Expected operator or callable, got {operator.__class__}\")\n\n        child = ColumnGroup(operator.output_column_names(self.columns))\n        child.parents = [self]\n        self.children.append(child)\n        child.op = operator\n\n        dependencies = operator.dependencies()\n        if dependencies:\n            child.dependencies = set()\n            if not isinstance(dependencies, collections.abc.Sequence):\n                dependencies = [dependencies]\n\n            for dependency in dependencies:\n                if not isinstance(dependency, ColumnGroup):\n                    dependency = ColumnGroup(dependency)\n                dependency.children.append(child)\n                child.parents.append(dependency)\n                child.dependencies.add(dependency)\n\n        return child\n\n    def __add__(self, other):\n        \"\"\"Adds columns from this ColumnGroup with another to return a new ColumnGroup\n\n        Parameters\n        -----------\n        other: ColumnGroup or str or list of str\n\n        Returns\n        -------\n        ColumnGroup\n        \"\"\"\n        if isinstance(other, str):\n            other = ColumnGroup([other])\n        elif isinstance(other, collections.abc.Sequence):\n            other = ColumnGroup(other)\n\n        # check if there are any columns with the same name in both column groups\n        overlap = set(self.columns).intersection(other.columns)\n        if overlap:\n            raise ValueError(f\"duplicate column names found: {overlap}\")\n\n        child = ColumnGroup(self.columns + other.columns)\n        child.parents = [self, other]\n        child.kind = \"+\"\n        self.children.append(child)\n        other.children.append(child)\n        return child\n\n    # handle the \"column_name\" + ColumnGroup case\n    __radd__ = __add__\n\n    def __sub__(self, other):\n        \"\"\"Adds columns from this ColumnGroup with another to return a new ColumnGroup\n\n        Parameters\n        -----------\n        other: ColumnGroup or str or list of str\n\n        Returns\n        -------\n        ColumnGroup\n        \"\"\"\n        if isinstance(other, ColumnGroup):\n            to_remove = set(other.columns)\n        elif isinstance(other, str):\n            to_remove = {other}\n        elif isinstance(other, collections.abc.Sequence):\n            to_remove = set(other)\n        else:\n            raise ValueError(f\"Expected ColumnGroup, str, or list of str. Got {other.__class__}\")\n        new_columns = [c for c in self.columns if c not in to_remove]\n        child = ColumnGroup(new_columns)\n        child.parents = [self]\n        self.children.append(child)\n        child.kind = f\"- {list(to_remove)}\"\n        return child\n\n    def __repr__(self):\n        output = \" output\" if not self.children else \"\"\n        return f\"<ColumnGroup {self.label}{output}>\"\n\n    @property\n    def flattened_columns(self):\n        return list(flatten(self.columns, container=tuple))\n\n    @property\n    def input_column_names(self):\n        \"\"\" returns the names of columns in the main chain \"\"\"\n        dependencies = self.dependencies or set()\n        return [\n            col for parent in self.parents for col in parent.columns if parent not in dependencies\n        ]\n\n    @property\n    def label(self):\n        if self.op:\n            return str(self.op.__class__.__name__)\n        elif self.kind:\n            return self.kind\n        elif not self.parents:\n            return f\"input cols=[{self._cols_repr}]\"\n        else:\n            return \"??\"\n\n    @property\n    def _cols_repr(self):\n        cols = \", \".join(map(str, self.columns[:3]))\n        if len(self.columns) > 3:\n            cols += \"...\"\n        return cols\n\n    @property\n    def graph(self):\n        return _to_graphviz(self)\n\n\ndef iter_nodes(nodes):\n    queue = nodes[:]\n    while queue:\n        current = queue.pop()\n        yield current\n        # TODO: deduplicate nodes?\n        for parent in current.parents:\n            queue.append(parent)\n\n\ndef _to_graphviz(column_group):\n    \"\"\" converts a columngroup to a GraphViz DiGraph object useful for display in notebooks \"\"\"\n    from graphviz import Digraph\n\n    column_group = _merge_add_nodes(column_group)\n    graph = Digraph()\n\n    # get all the nodes from parents of this columngroup\n    # and add edges between each of them\n    allnodes = list(set(iter_nodes([column_group])))\n    node_ids = {v: str(k) for k, v in enumerate(allnodes)}\n    for node, nodeid in node_ids.items():\n        graph.node(nodeid, node.label)\n        for parent in node.parents:\n            graph.edge(node_ids[parent], nodeid)\n\n    # add a single 'output' node representing the final state\n    output_node_id = str(len(allnodes))\n    graph.node(output_node_id, f\"output cols=[{column_group._cols_repr}]\")\n    graph.edge(node_ids[column_group], output_node_id)\n    return graph\n\n\ndef _merge_add_nodes(graph):\n    \"\"\" merges repeat '+' nodes, leading to nicer looking outputs \"\"\"\n    # lets take a copy to avoid mutating the input\n    import copy\n\n    graph = copy.deepcopy(graph)\n\n    queue = [graph]\n    while queue:\n        current = queue.pop()\n        if current.kind == \"+\":\n            changed = True\n            while changed:\n                changed = False\n                parents = []\n                for i, parent in enumerate(current.parents):\n                    if parent.kind == \"+\" and len(parent.children) == 1:\n                        changed = True\n                        # disconnect parent, point all the grandparents at current instead\n                        parents.extend(parent.parents)\n                        for grandparent in parent.parents:\n                            grandparent.children = [\n                                current if child == parent else child\n                                for child in grandparent.children\n                            ]\n                    else:\n                        parents.append(parent)\n                current.parents = parents\n\n        queue.extend(current.parents)\n\n    return graph\n\n\ndef _convert_col(col):\n    if isinstance(col, (str, tuple)):\n        return col\n    elif isinstance(col, list):\n        return tuple(col)\n    else:\n        raise ValueError(\"Invalid column value for ColumnGroup: %s\", col)\n", "path": "nvtabular/column_group.py"}], "after_files": [{"content": "#\n# Copyright (c) 2020, NVIDIA CORPORATION.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\nimport collections.abc\n\nfrom dask.core import flatten\n\nfrom nvtabular.ops import LambdaOp, Operator\n\n\nclass ColumnGroup:\n    \"\"\"A ColumnGroup is a group of columns that you want to apply the same transformations to.\n    ColumnGroup's can be transformed by shifting operators on to them, which returns a new\n    ColumnGroup with the transformations applied. This lets you define a graph of operations\n    that makes up your workflow\n\n    Parameters\n    ----------\n    columns: list of (str or tuple of str)\n        The columns to select from the input Dataset. The elements of this list are strings\n        indicating the column names in most cases, but can also be tuples of strings\n        for feature crosses.\n    \"\"\"\n\n    def __init__(self, columns):\n        self.parents = []\n        self.children = []\n        self.op = None\n        self.kind = None\n        self.dependencies = None\n\n        if isinstance(columns, str):\n            columns = [columns]\n\n        # if any of the values we're passed are a columngroup\n        # we have to ourselves as a childnode in the graph.\n        if any(isinstance(col, ColumnGroup) for col in columns):\n            self.columns = []\n            self.kind = \"[...]\"\n            for col in columns:\n                if not isinstance(col, ColumnGroup):\n                    col = ColumnGroup(col)\n                else:\n                    # we can't handle nesting arbitrarily deep here\n                    # only accept non-nested (str) columns here\n                    if any(not isinstance(c, str) for c in col.columns):\n                        raise ValueError(\"Can't handle more than 1 level of nested columns\")\n\n                col.children.append(self)\n                self.parents.append(col)\n                self.columns.append(tuple(col.columns))\n\n        else:\n            self.columns = [_convert_col(col) for col in columns]\n\n    def __rshift__(self, operator):\n        \"\"\"Transforms this ColumnGroup by applying an Operator\n\n        Parameters\n        -----------\n        operators: Operator or callable\n\n        Returns\n        -------\n        ColumnGroup\n        \"\"\"\n        if isinstance(operator, type) and issubclass(operator, Operator):\n            # handle case where an operator class is passed\n            operator = operator()\n        elif callable(operator):\n            # implicit lambdaop conversion.\n            operator = LambdaOp(operator)\n\n        if not isinstance(operator, Operator):\n            raise ValueError(f\"Expected operator or callable, got {operator.__class__}\")\n\n        child = ColumnGroup(operator.output_column_names(self.columns))\n        child.parents = [self]\n        self.children.append(child)\n        child.op = operator\n\n        dependencies = operator.dependencies()\n        if dependencies:\n            child.dependencies = set()\n            if not isinstance(dependencies, collections.abc.Sequence):\n                dependencies = [dependencies]\n\n            for dependency in dependencies:\n                if not isinstance(dependency, ColumnGroup):\n                    dependency = ColumnGroup(dependency)\n                dependency.children.append(child)\n                child.parents.append(dependency)\n                child.dependencies.add(dependency)\n\n        return child\n\n    def __add__(self, other):\n        \"\"\"Adds columns from this ColumnGroup with another to return a new ColumnGroup\n\n        Parameters\n        -----------\n        other: ColumnGroup or str or list of str\n\n        Returns\n        -------\n        ColumnGroup\n        \"\"\"\n        if isinstance(other, str):\n            other = ColumnGroup([other])\n        elif isinstance(other, collections.abc.Sequence):\n            other = ColumnGroup(other)\n\n        # check if there are any columns with the same name in both column groups\n        overlap = set(self.columns).intersection(other.columns)\n        if overlap:\n            raise ValueError(f\"duplicate column names found: {overlap}\")\n\n        child = ColumnGroup(self.columns + other.columns)\n        child.parents = [self, other]\n        child.kind = \"+\"\n        self.children.append(child)\n        other.children.append(child)\n        return child\n\n    # handle the \"column_name\" + ColumnGroup case\n    __radd__ = __add__\n\n    def __sub__(self, other):\n        \"\"\"Adds columns from this ColumnGroup with another to return a new ColumnGroup\n\n        Parameters\n        -----------\n        other: ColumnGroup or str or list of str\n\n        Returns\n        -------\n        ColumnGroup\n        \"\"\"\n        if isinstance(other, ColumnGroup):\n            to_remove = set(other.columns)\n        elif isinstance(other, str):\n            to_remove = {other}\n        elif isinstance(other, collections.abc.Sequence):\n            to_remove = set(other)\n        else:\n            raise ValueError(f\"Expected ColumnGroup, str, or list of str. Got {other.__class__}\")\n        new_columns = [c for c in self.columns if c not in to_remove]\n        child = ColumnGroup(new_columns)\n        child.parents = [self]\n        self.children.append(child)\n        child.kind = f\"- {list(to_remove)}\"\n        return child\n\n    def __repr__(self):\n        output = \" output\" if not self.children else \"\"\n        return f\"<ColumnGroup {self.label}{output}>\"\n\n    @property\n    def flattened_columns(self):\n        return list(flatten(self.columns, container=tuple))\n\n    @property\n    def input_column_names(self):\n        \"\"\" returns the names of columns in the main chain \"\"\"\n        dependencies = self.dependencies or set()\n        return [\n            col for parent in self.parents for col in parent.columns if parent not in dependencies\n        ]\n\n    @property\n    def label(self):\n        if self.op:\n            return str(self.op.__class__.__name__)\n        elif self.kind:\n            return self.kind\n        elif not self.parents:\n            return f\"input cols=[{self._cols_repr}]\"\n        else:\n            return \"??\"\n\n    @property\n    def _cols_repr(self):\n        cols = \", \".join(map(str, self.columns[:3]))\n        if len(self.columns) > 3:\n            cols += \"...\"\n        return cols\n\n    @property\n    def graph(self):\n        return _to_graphviz(self)\n\n\ndef iter_nodes(nodes):\n    queue = nodes[:]\n    while queue:\n        current = queue.pop()\n        yield current\n        # TODO: deduplicate nodes?\n        for parent in current.parents:\n            queue.append(parent)\n\n\ndef _to_graphviz(column_group):\n    \"\"\" converts a columngroup to a GraphViz DiGraph object useful for display in notebooks \"\"\"\n    from graphviz import Digraph\n\n    column_group = _merge_add_nodes(column_group)\n    graph = Digraph()\n\n    # get all the nodes from parents of this columngroup\n    # and add edges between each of them\n    allnodes = list(set(iter_nodes([column_group])))\n    node_ids = {v: str(k) for k, v in enumerate(allnodes)}\n    for node, nodeid in node_ids.items():\n        graph.node(nodeid, node.label)\n        for parent in node.parents:\n            graph.edge(node_ids[parent], nodeid)\n\n    # add a single 'output' node representing the final state\n    output_node_id = str(len(allnodes))\n    graph.node(output_node_id, f\"output cols=[{column_group._cols_repr}]\")\n    graph.edge(node_ids[column_group], output_node_id)\n    return graph\n\n\ndef _merge_add_nodes(graph):\n    \"\"\" merges repeat '+' nodes, leading to nicer looking outputs \"\"\"\n    # lets take a copy to avoid mutating the input\n    import copy\n\n    graph = copy.deepcopy(graph)\n\n    queue = [graph]\n    while queue:\n        current = queue.pop()\n        if current.kind == \"+\":\n            changed = True\n            while changed:\n                changed = False\n                parents = []\n                for i, parent in enumerate(current.parents):\n                    if parent.kind == \"+\" and len(parent.children) == 1:\n                        changed = True\n                        # disconnect parent, point all the grandparents at current instead\n                        parents.extend(parent.parents)\n                        for grandparent in parent.parents:\n                            grandparent.children = [\n                                current if child == parent else child\n                                for child in grandparent.children\n                            ]\n                    else:\n                        parents.append(parent)\n                current.parents = parents\n\n        queue.extend(current.parents)\n\n    return graph\n\n\ndef _convert_col(col):\n    if isinstance(col, (str, tuple)):\n        return col\n    elif isinstance(col, list):\n        return tuple(col)\n    else:\n        raise ValueError(\"Invalid column value for ColumnGroup: %s\", col)\n", "path": "nvtabular/column_group.py"}]}