We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
perf: add fast path for simple selections in dask backend
For simple selections we don't need to bother with the `dd.concat` here: https://github.com/ibis-project/ibis/blob/master/ibis/backends/dask/execution/selection.py#L154 and should probably select on the data directly 

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `ibis/backends/dask/execution/selection.py`
Content:
```
1 """Dispatching code for Selection operations.
2 """
3 
4 
5 import functools
6 import operator
7 from typing import Optional
8 
9 import dask.dataframe as dd
10 import pandas
11 from toolz import concatv
12 
13 import ibis.expr.operations as ops
14 import ibis.expr.types as ir
15 from ibis.backends.pandas.execution.selection import (
16     compute_projection,
17     compute_projection_table_expr,
18     map_new_column_names_to_data,
19     remap_overlapping_column_names,
20 )
21 from ibis.expr.scope import Scope
22 from ibis.expr.typing import TimeContext
23 
24 from ..core import execute
25 from ..dispatch import execute_node
26 from ..execution import constants
27 from ..execution.util import (
28     add_partitioned_sorted_column,
29     coerce_to_output,
30     compute_sorted_frame,
31 )
32 
33 
34 @compute_projection.register(ir.ScalarExpr, ops.Selection, dd.DataFrame)
35 def compute_projection_scalar_expr(
36     expr,
37     parent,
38     data,
39     scope: Scope,
40     timecontext: Optional[TimeContext] = None,
41     **kwargs,
42 ):
43     name = expr._name
44     assert name is not None, 'Scalar selection name is None'
45 
46     op = expr.op()
47     parent_table_op = parent.table.op()
48 
49     data_columns = frozenset(data.columns)
50 
51     scope = scope.merge_scopes(
52         Scope(
53             {
54                 t: map_new_column_names_to_data(
55                     remap_overlapping_column_names(
56                         parent_table_op, t, data_columns
57                     ),
58                     data,
59                 )
60             },
61             timecontext,
62         )
63         for t in op.root_tables()
64     )
65     scalar = execute(expr, scope=scope, **kwargs)
66     return data.assign(**{name: scalar})[name]
67 
68 
69 @compute_projection.register(ir.ColumnExpr, ops.Selection, dd.DataFrame)
70 def compute_projection_column_expr(
71     expr,
72     parent,
73     data,
74     scope: Scope,
75     timecontext: Optional[TimeContext],
76     **kwargs,
77 ):
78     result_name = getattr(expr, '_name', None)
79     op = expr.op()
80     parent_table_op = parent.table.op()
81 
82     if isinstance(op, ops.TableColumn):
83         # slightly faster path for simple column selection
84         name = op.name
85 
86         if name in data:
87             return data[name].rename(result_name or name)
88 
89         if not isinstance(parent_table_op, ops.Join):
90             raise KeyError(name)
91         (root_table,) = op.root_tables()
92         left_root, right_root = ops.distinct_roots(
93             parent_table_op.left, parent_table_op.right
94         )
95         suffixes = {
96             left_root: constants.LEFT_JOIN_SUFFIX,
97             right_root: constants.RIGHT_JOIN_SUFFIX,
98         }
99         return data.loc[:, name + suffixes[root_table]].rename(
100             result_name or name
101         )
102 
103     data_columns = frozenset(data.columns)
104 
105     scope = scope.merge_scopes(
106         Scope(
107             {
108                 t: map_new_column_names_to_data(
109                     remap_overlapping_column_names(
110                         parent_table_op, t, data_columns
111                     ),
112                     data,
113                 )
114             },
115             timecontext,
116         )
117         for t in op.root_tables()
118     )
119 
120     result = execute(expr, scope=scope, timecontext=timecontext, **kwargs)
121     result = coerce_to_output(result, expr, data.index)
122     assert result_name is not None, 'Column selection name is None'
123 
124     return result
125 
126 
127 compute_projection.register(ir.TableExpr, ops.Selection, dd.DataFrame)(
128     compute_projection_table_expr
129 )
130 
131 
132 @execute_node.register(ops.Selection, dd.DataFrame)
133 def execute_selection_dataframe(
134     op, data, scope: Scope, timecontext: Optional[TimeContext], **kwargs
135 ):
136     selections = op.selections
137     predicates = op.predicates
138     sort_keys = op.sort_keys
139     result = data
140 
141     # Build up the individual dask structures from column expressions
142     if selections:
143         # Create a unique row identifier and set it as the index. This is used
144         # in dd.concat to merge the pieces back together.
145         data = add_partitioned_sorted_column(data)
146         data_pieces = []
147         for selection in selections:
148             dask_object = compute_projection(
149                 selection,
150                 op,
151                 data,
152                 scope=scope,
153                 timecontext=timecontext,
154                 **kwargs,
155             )
156             data_pieces.append(dask_object)
157 
158         result = dd.concat(data_pieces, axis=1)
159         result.reset_index(drop=True)
160 
161     if predicates:
162         predicates = _compute_predicates(
163             op.table.op(), predicates, data, scope, timecontext, **kwargs
164         )
165         predicate = functools.reduce(operator.and_, predicates)
166         result = result.loc[predicate]
167 
168     if sort_keys:
169         if len(sort_keys) > 1:
170             raise NotImplementedError(
171                 """
172                 Multi-key sorting is not implemented for the Dask backend
173                 """
174             )
175         sort_key = sort_keys[0]
176         ascending = getattr(sort_key.op(), 'ascending', True)
177         if not ascending:
178             raise NotImplementedError(
179                 "Descending sort is not supported for the Dask backend"
180             )
181         result = compute_sorted_frame(
182             result,
183             order_by=sort_key,
184             scope=scope,
185             timecontext=timecontext,
186             **kwargs,
187         )
188 
189         return result
190     else:
191         grouping_keys = ordering_keys = ()
192 
193     # return early if we do not have any temporary grouping or ordering columns
194     assert not grouping_keys, 'group by should never show up in Selection'
195     if not ordering_keys:
196         return result
197 
198     # create a sequence of columns that we need to drop
199     temporary_columns = pandas.Index(
200         concatv(grouping_keys, ordering_keys)
201     ).difference(data.columns)
202 
203     # no reason to call drop if we don't need to
204     if temporary_columns.empty:
205         return result
206 
207     # drop every temporary column we created for ordering or grouping
208     return result.drop(temporary_columns, axis=1)
209 
210 
211 def _compute_predicates(
212     table_op,
213     predicates,
214     data,
215     scope: Scope,
216     timecontext: Optional[TimeContext],
217     **kwargs,
218 ):
219     """Compute the predicates for a table operation.
220 
221     Parameters
222     ----------
223     table_op : TableNode
224     predicates : List[ir.ColumnExpr]
225     data : pd.DataFrame
226     scope : Scope
227     timecontext: Optional[TimeContext]
228     kwargs : dict
229 
230     Returns
231     -------
232     computed_predicate : pd.Series[bool]
233 
234     Notes
235     -----
236     This handles the cases where the predicates are computed columns, in
237     addition to the simple case of named columns coming directly from the input
238     table.
239     """
240     for predicate in predicates:
241         # Map each root table of the predicate to the data so that we compute
242         # predicates on the result instead of any left or right tables if the
243         # Selection is on a Join. Project data to only inlude columns from
244         # the root table.
245         root_tables = predicate.op().root_tables()
246 
247         # handle suffixes
248         data_columns = frozenset(data.columns)
249 
250         additional_scope = Scope()
251         for root_table in root_tables:
252             mapping = remap_overlapping_column_names(
253                 table_op, root_table, data_columns
254             )
255             if mapping is not None:
256                 new_data = data.loc[:, mapping.keys()].rename(columns=mapping)
257             else:
258                 new_data = data
259             additional_scope = additional_scope.merge_scope(
260                 Scope({root_table: new_data}, timecontext)
261             )
262 
263         scope = scope.merge_scope(additional_scope)
264         yield execute(predicate, scope=scope, **kwargs)
265
```
--- 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/ibis/backends/dask/execution/selection.py b/ibis/backends/dask/execution/selection.py
--- a/ibis/backends/dask/execution/selection.py
+++ b/ibis/backends/dask/execution/selection.py
@@ -4,7 +4,7 @@
 
 import functools
 import operator
-from typing import Optional
+from typing import List, Optional
 
 import dask.dataframe as dd
 import pandas
@@ -47,7 +47,6 @@
     parent_table_op = parent.table.op()
 
     data_columns = frozenset(data.columns)
-
     scope = scope.merge_scopes(
         Scope(
             {
@@ -129,6 +128,42 @@
 )
 
 
+def build_df_from_selection(
+    selections: List[ir.ColumnExpr], data: dd.DataFrame
+) -> dd.DataFrame:
+    """Build up a df by doing direct selections, renaming if necessary."""
+    cols = [
+        (s.op().name, getattr(s, "_name", s.op().name)) for s in selections
+    ]
+    renamed_cols = {
+        col: renamed_col for col, renamed_col in cols if col != renamed_col
+    }
+
+    result = data[[col for col, _ in cols]]
+    if renamed_cols:
+        result = result.rename(columns=renamed_cols)
+
+    return result
+
+
+def build_df_from_projection(
+    selections: List[ir.Expr], op: ops.Selection, data: dd.DataFrame, **kwargs
+) -> dd.DataFrame:
+    """
+    Build up a df from individual pieces by dispatching to `compute_projection`
+    for each expression.
+    """
+
+    # Create a unique row identifier and set it as the index. This is
+    # used in dd.concat to merge the pieces back together.
+    data = add_partitioned_sorted_column(data)
+    data_pieces = [
+        compute_projection(s, op, data, **kwargs) for s in selections
+    ]
+
+    return dd.concat(data_pieces, axis=1).reset_index(drop=True)
+
+
 @execute_node.register(ops.Selection, dd.DataFrame)
 def execute_selection_dataframe(
     op, data, scope: Scope, timecontext: Optional[TimeContext], **kwargs
@@ -138,25 +173,22 @@
     sort_keys = op.sort_keys
     result = data
 
-    # Build up the individual dask structures from column expressions
     if selections:
-        # Create a unique row identifier and set it as the index. This is used
-        # in dd.concat to merge the pieces back together.
-        data = add_partitioned_sorted_column(data)
-        data_pieces = []
-        for selection in selections:
-            dask_object = compute_projection(
-                selection,
+        # if we are just performing select operations and all columns are in
+        # the table we can do a direct selection
+        if all(isinstance(s.op(), ops.TableColumn) for s in selections) and {
+            s.op().name for s in selections
+        }.issubset(set(result.columns)):
+            result = build_df_from_selection(selections, data)
+        else:
+            result = build_df_from_projection(
+                selections,
                 op,
                 data,
                 scope=scope,
                 timecontext=timecontext,
                 **kwargs,
             )
-            data_pieces.append(dask_object)
-
-        result = dd.concat(data_pieces, axis=1)
-        result.reset_index(drop=True)
 
     if predicates:
         predicates = _compute_predicates(

{"golden_diff": "diff --git a/ibis/backends/dask/execution/selection.py b/ibis/backends/dask/execution/selection.py\n--- a/ibis/backends/dask/execution/selection.py\n+++ b/ibis/backends/dask/execution/selection.py\n@@ -4,7 +4,7 @@\n \n import functools\n import operator\n-from typing import Optional\n+from typing import List, Optional\n \n import dask.dataframe as dd\n import pandas\n@@ -47,7 +47,6 @@\n     parent_table_op = parent.table.op()\n \n     data_columns = frozenset(data.columns)\n-\n     scope = scope.merge_scopes(\n         Scope(\n             {\n@@ -129,6 +128,42 @@\n )\n \n \n+def build_df_from_selection(\n+    selections: List[ir.ColumnExpr], data: dd.DataFrame\n+) -> dd.DataFrame:\n+    \"\"\"Build up a df by doing direct selections, renaming if necessary.\"\"\"\n+    cols = [\n+        (s.op().name, getattr(s, \"_name\", s.op().name)) for s in selections\n+    ]\n+    renamed_cols = {\n+        col: renamed_col for col, renamed_col in cols if col != renamed_col\n+    }\n+\n+    result = data[[col for col, _ in cols]]\n+    if renamed_cols:\n+        result = result.rename(columns=renamed_cols)\n+\n+    return result\n+\n+\n+def build_df_from_projection(\n+    selections: List[ir.Expr], op: ops.Selection, data: dd.DataFrame, **kwargs\n+) -> dd.DataFrame:\n+    \"\"\"\n+    Build up a df from individual pieces by dispatching to `compute_projection`\n+    for each expression.\n+    \"\"\"\n+\n+    # Create a unique row identifier and set it as the index. This is\n+    # used in dd.concat to merge the pieces back together.\n+    data = add_partitioned_sorted_column(data)\n+    data_pieces = [\n+        compute_projection(s, op, data, **kwargs) for s in selections\n+    ]\n+\n+    return dd.concat(data_pieces, axis=1).reset_index(drop=True)\n+\n+\n @execute_node.register(ops.Selection, dd.DataFrame)\n def execute_selection_dataframe(\n     op, data, scope: Scope, timecontext: Optional[TimeContext], **kwargs\n@@ -138,25 +173,22 @@\n     sort_keys = op.sort_keys\n     result = data\n \n-    # Build up the individual dask structures from column expressions\n     if selections:\n-        # Create a unique row identifier and set it as the index. This is used\n-        # in dd.concat to merge the pieces back together.\n-        data = add_partitioned_sorted_column(data)\n-        data_pieces = []\n-        for selection in selections:\n-            dask_object = compute_projection(\n-                selection,\n+        # if we are just performing select operations and all columns are in\n+        # the table we can do a direct selection\n+        if all(isinstance(s.op(), ops.TableColumn) for s in selections) and {\n+            s.op().name for s in selections\n+        }.issubset(set(result.columns)):\n+            result = build_df_from_selection(selections, data)\n+        else:\n+            result = build_df_from_projection(\n+                selections,\n                 op,\n                 data,\n                 scope=scope,\n                 timecontext=timecontext,\n                 **kwargs,\n             )\n-            data_pieces.append(dask_object)\n-\n-        result = dd.concat(data_pieces, axis=1)\n-        result.reset_index(drop=True)\n \n     if predicates:\n         predicates = _compute_predicates(\n", "issue": "perf: add fast path for simple selections in dask backend\nFor simple selections we don't need to bother with the `dd.concat` here: https://github.com/ibis-project/ibis/blob/master/ibis/backends/dask/execution/selection.py#L154 and should probably select on the data directly \n", "before_files": [{"content": "\"\"\"Dispatching code for Selection operations.\n\"\"\"\n\n\nimport functools\nimport operator\nfrom typing import Optional\n\nimport dask.dataframe as dd\nimport pandas\nfrom toolz import concatv\n\nimport ibis.expr.operations as ops\nimport ibis.expr.types as ir\nfrom ibis.backends.pandas.execution.selection import (\n    compute_projection,\n    compute_projection_table_expr,\n    map_new_column_names_to_data,\n    remap_overlapping_column_names,\n)\nfrom ibis.expr.scope import Scope\nfrom ibis.expr.typing import TimeContext\n\nfrom ..core import execute\nfrom ..dispatch import execute_node\nfrom ..execution import constants\nfrom ..execution.util import (\n    add_partitioned_sorted_column,\n    coerce_to_output,\n    compute_sorted_frame,\n)\n\n\n@compute_projection.register(ir.ScalarExpr, ops.Selection, dd.DataFrame)\ndef compute_projection_scalar_expr(\n    expr,\n    parent,\n    data,\n    scope: Scope,\n    timecontext: Optional[TimeContext] = None,\n    **kwargs,\n):\n    name = expr._name\n    assert name is not None, 'Scalar selection name is None'\n\n    op = expr.op()\n    parent_table_op = parent.table.op()\n\n    data_columns = frozenset(data.columns)\n\n    scope = scope.merge_scopes(\n        Scope(\n            {\n                t: map_new_column_names_to_data(\n                    remap_overlapping_column_names(\n                        parent_table_op, t, data_columns\n                    ),\n                    data,\n                )\n            },\n            timecontext,\n        )\n        for t in op.root_tables()\n    )\n    scalar = execute(expr, scope=scope, **kwargs)\n    return data.assign(**{name: scalar})[name]\n\n\n@compute_projection.register(ir.ColumnExpr, ops.Selection, dd.DataFrame)\ndef compute_projection_column_expr(\n    expr,\n    parent,\n    data,\n    scope: Scope,\n    timecontext: Optional[TimeContext],\n    **kwargs,\n):\n    result_name = getattr(expr, '_name', None)\n    op = expr.op()\n    parent_table_op = parent.table.op()\n\n    if isinstance(op, ops.TableColumn):\n        # slightly faster path for simple column selection\n        name = op.name\n\n        if name in data:\n            return data[name].rename(result_name or name)\n\n        if not isinstance(parent_table_op, ops.Join):\n            raise KeyError(name)\n        (root_table,) = op.root_tables()\n        left_root, right_root = ops.distinct_roots(\n            parent_table_op.left, parent_table_op.right\n        )\n        suffixes = {\n            left_root: constants.LEFT_JOIN_SUFFIX,\n            right_root: constants.RIGHT_JOIN_SUFFIX,\n        }\n        return data.loc[:, name + suffixes[root_table]].rename(\n            result_name or name\n        )\n\n    data_columns = frozenset(data.columns)\n\n    scope = scope.merge_scopes(\n        Scope(\n            {\n                t: map_new_column_names_to_data(\n                    remap_overlapping_column_names(\n                        parent_table_op, t, data_columns\n                    ),\n                    data,\n                )\n            },\n            timecontext,\n        )\n        for t in op.root_tables()\n    )\n\n    result = execute(expr, scope=scope, timecontext=timecontext, **kwargs)\n    result = coerce_to_output(result, expr, data.index)\n    assert result_name is not None, 'Column selection name is None'\n\n    return result\n\n\ncompute_projection.register(ir.TableExpr, ops.Selection, dd.DataFrame)(\n    compute_projection_table_expr\n)\n\n\n@execute_node.register(ops.Selection, dd.DataFrame)\ndef execute_selection_dataframe(\n    op, data, scope: Scope, timecontext: Optional[TimeContext], **kwargs\n):\n    selections = op.selections\n    predicates = op.predicates\n    sort_keys = op.sort_keys\n    result = data\n\n    # Build up the individual dask structures from column expressions\n    if selections:\n        # Create a unique row identifier and set it as the index. This is used\n        # in dd.concat to merge the pieces back together.\n        data = add_partitioned_sorted_column(data)\n        data_pieces = []\n        for selection in selections:\n            dask_object = compute_projection(\n                selection,\n                op,\n                data,\n                scope=scope,\n                timecontext=timecontext,\n                **kwargs,\n            )\n            data_pieces.append(dask_object)\n\n        result = dd.concat(data_pieces, axis=1)\n        result.reset_index(drop=True)\n\n    if predicates:\n        predicates = _compute_predicates(\n            op.table.op(), predicates, data, scope, timecontext, **kwargs\n        )\n        predicate = functools.reduce(operator.and_, predicates)\n        result = result.loc[predicate]\n\n    if sort_keys:\n        if len(sort_keys) > 1:\n            raise NotImplementedError(\n                \"\"\"\n                Multi-key sorting is not implemented for the Dask backend\n                \"\"\"\n            )\n        sort_key = sort_keys[0]\n        ascending = getattr(sort_key.op(), 'ascending', True)\n        if not ascending:\n            raise NotImplementedError(\n                \"Descending sort is not supported for the Dask backend\"\n            )\n        result = compute_sorted_frame(\n            result,\n            order_by=sort_key,\n            scope=scope,\n            timecontext=timecontext,\n            **kwargs,\n        )\n\n        return result\n    else:\n        grouping_keys = ordering_keys = ()\n\n    # return early if we do not have any temporary grouping or ordering columns\n    assert not grouping_keys, 'group by should never show up in Selection'\n    if not ordering_keys:\n        return result\n\n    # create a sequence of columns that we need to drop\n    temporary_columns = pandas.Index(\n        concatv(grouping_keys, ordering_keys)\n    ).difference(data.columns)\n\n    # no reason to call drop if we don't need to\n    if temporary_columns.empty:\n        return result\n\n    # drop every temporary column we created for ordering or grouping\n    return result.drop(temporary_columns, axis=1)\n\n\ndef _compute_predicates(\n    table_op,\n    predicates,\n    data,\n    scope: Scope,\n    timecontext: Optional[TimeContext],\n    **kwargs,\n):\n    \"\"\"Compute the predicates for a table operation.\n\n    Parameters\n    ----------\n    table_op : TableNode\n    predicates : List[ir.ColumnExpr]\n    data : pd.DataFrame\n    scope : Scope\n    timecontext: Optional[TimeContext]\n    kwargs : dict\n\n    Returns\n    -------\n    computed_predicate : pd.Series[bool]\n\n    Notes\n    -----\n    This handles the cases where the predicates are computed columns, in\n    addition to the simple case of named columns coming directly from the input\n    table.\n    \"\"\"\n    for predicate in predicates:\n        # Map each root table of the predicate to the data so that we compute\n        # predicates on the result instead of any left or right tables if the\n        # Selection is on a Join. Project data to only inlude columns from\n        # the root table.\n        root_tables = predicate.op().root_tables()\n\n        # handle suffixes\n        data_columns = frozenset(data.columns)\n\n        additional_scope = Scope()\n        for root_table in root_tables:\n            mapping = remap_overlapping_column_names(\n                table_op, root_table, data_columns\n            )\n            if mapping is not None:\n                new_data = data.loc[:, mapping.keys()].rename(columns=mapping)\n            else:\n                new_data = data\n            additional_scope = additional_scope.merge_scope(\n                Scope({root_table: new_data}, timecontext)\n            )\n\n        scope = scope.merge_scope(additional_scope)\n        yield execute(predicate, scope=scope, **kwargs)\n", "path": "ibis/backends/dask/execution/selection.py"}], "after_files": [{"content": "\"\"\"Dispatching code for Selection operations.\n\"\"\"\n\n\nimport functools\nimport operator\nfrom typing import List, Optional\n\nimport dask.dataframe as dd\nimport pandas\nfrom toolz import concatv\n\nimport ibis.expr.operations as ops\nimport ibis.expr.types as ir\nfrom ibis.backends.pandas.execution.selection import (\n    compute_projection,\n    compute_projection_table_expr,\n    map_new_column_names_to_data,\n    remap_overlapping_column_names,\n)\nfrom ibis.expr.scope import Scope\nfrom ibis.expr.typing import TimeContext\n\nfrom ..core import execute\nfrom ..dispatch import execute_node\nfrom ..execution import constants\nfrom ..execution.util import (\n    add_partitioned_sorted_column,\n    coerce_to_output,\n    compute_sorted_frame,\n)\n\n\n@compute_projection.register(ir.ScalarExpr, ops.Selection, dd.DataFrame)\ndef compute_projection_scalar_expr(\n    expr,\n    parent,\n    data,\n    scope: Scope,\n    timecontext: Optional[TimeContext] = None,\n    **kwargs,\n):\n    name = expr._name\n    assert name is not None, 'Scalar selection name is None'\n\n    op = expr.op()\n    parent_table_op = parent.table.op()\n\n    data_columns = frozenset(data.columns)\n    scope = scope.merge_scopes(\n        Scope(\n            {\n                t: map_new_column_names_to_data(\n                    remap_overlapping_column_names(\n                        parent_table_op, t, data_columns\n                    ),\n                    data,\n                )\n            },\n            timecontext,\n        )\n        for t in op.root_tables()\n    )\n    scalar = execute(expr, scope=scope, **kwargs)\n    return data.assign(**{name: scalar})[name]\n\n\n@compute_projection.register(ir.ColumnExpr, ops.Selection, dd.DataFrame)\ndef compute_projection_column_expr(\n    expr,\n    parent,\n    data,\n    scope: Scope,\n    timecontext: Optional[TimeContext],\n    **kwargs,\n):\n    result_name = getattr(expr, '_name', None)\n    op = expr.op()\n    parent_table_op = parent.table.op()\n\n    if isinstance(op, ops.TableColumn):\n        # slightly faster path for simple column selection\n        name = op.name\n\n        if name in data:\n            return data[name].rename(result_name or name)\n\n        if not isinstance(parent_table_op, ops.Join):\n            raise KeyError(name)\n        (root_table,) = op.root_tables()\n        left_root, right_root = ops.distinct_roots(\n            parent_table_op.left, parent_table_op.right\n        )\n        suffixes = {\n            left_root: constants.LEFT_JOIN_SUFFIX,\n            right_root: constants.RIGHT_JOIN_SUFFIX,\n        }\n        return data.loc[:, name + suffixes[root_table]].rename(\n            result_name or name\n        )\n\n    data_columns = frozenset(data.columns)\n\n    scope = scope.merge_scopes(\n        Scope(\n            {\n                t: map_new_column_names_to_data(\n                    remap_overlapping_column_names(\n                        parent_table_op, t, data_columns\n                    ),\n                    data,\n                )\n            },\n            timecontext,\n        )\n        for t in op.root_tables()\n    )\n\n    result = execute(expr, scope=scope, timecontext=timecontext, **kwargs)\n    result = coerce_to_output(result, expr, data.index)\n    assert result_name is not None, 'Column selection name is None'\n\n    return result\n\n\ncompute_projection.register(ir.TableExpr, ops.Selection, dd.DataFrame)(\n    compute_projection_table_expr\n)\n\n\ndef build_df_from_selection(\n    selections: List[ir.ColumnExpr], data: dd.DataFrame\n) -> dd.DataFrame:\n    \"\"\"Build up a df by doing direct selections, renaming if necessary.\"\"\"\n    cols = [\n        (s.op().name, getattr(s, \"_name\", s.op().name)) for s in selections\n    ]\n    renamed_cols = {\n        col: renamed_col for col, renamed_col in cols if col != renamed_col\n    }\n\n    result = data[[col for col, _ in cols]]\n    if renamed_cols:\n        result = result.rename(columns=renamed_cols)\n\n    return result\n\n\ndef build_df_from_projection(\n    selections: List[ir.Expr], op: ops.Selection, data: dd.DataFrame, **kwargs\n) -> dd.DataFrame:\n    \"\"\"\n    Build up a df from individual pieces by dispatching to `compute_projection`\n    for each expression.\n    \"\"\"\n\n    # Create a unique row identifier and set it as the index. This is\n    # used in dd.concat to merge the pieces back together.\n    data = add_partitioned_sorted_column(data)\n    data_pieces = [\n        compute_projection(s, op, data, **kwargs) for s in selections\n    ]\n\n    return dd.concat(data_pieces, axis=1).reset_index(drop=True)\n\n\n@execute_node.register(ops.Selection, dd.DataFrame)\ndef execute_selection_dataframe(\n    op, data, scope: Scope, timecontext: Optional[TimeContext], **kwargs\n):\n    selections = op.selections\n    predicates = op.predicates\n    sort_keys = op.sort_keys\n    result = data\n\n    if selections:\n        # if we are just performing select operations and all columns are in\n        # the table we can do a direct selection\n        if all(isinstance(s.op(), ops.TableColumn) for s in selections) and {\n            s.op().name for s in selections\n        }.issubset(set(result.columns)):\n            result = build_df_from_selection(selections, data)\n        else:\n            result = build_df_from_projection(\n                selections,\n                op,\n                data,\n                scope=scope,\n                timecontext=timecontext,\n                **kwargs,\n            )\n\n    if predicates:\n        predicates = _compute_predicates(\n            op.table.op(), predicates, data, scope, timecontext, **kwargs\n        )\n        predicate = functools.reduce(operator.and_, predicates)\n        result = result.loc[predicate]\n\n    if sort_keys:\n        if len(sort_keys) > 1:\n            raise NotImplementedError(\n                \"\"\"\n                Multi-key sorting is not implemented for the Dask backend\n                \"\"\"\n            )\n        sort_key = sort_keys[0]\n        ascending = getattr(sort_key.op(), 'ascending', True)\n        if not ascending:\n            raise NotImplementedError(\n                \"Descending sort is not supported for the Dask backend\"\n            )\n        result = compute_sorted_frame(\n            result,\n            order_by=sort_key,\n            scope=scope,\n            timecontext=timecontext,\n            **kwargs,\n        )\n\n        return result\n    else:\n        grouping_keys = ordering_keys = ()\n\n    # return early if we do not have any temporary grouping or ordering columns\n    assert not grouping_keys, 'group by should never show up in Selection'\n    if not ordering_keys:\n        return result\n\n    # create a sequence of columns that we need to drop\n    temporary_columns = pandas.Index(\n        concatv(grouping_keys, ordering_keys)\n    ).difference(data.columns)\n\n    # no reason to call drop if we don't need to\n    if temporary_columns.empty:\n        return result\n\n    # drop every temporary column we created for ordering or grouping\n    return result.drop(temporary_columns, axis=1)\n\n\ndef _compute_predicates(\n    table_op,\n    predicates,\n    data,\n    scope: Scope,\n    timecontext: Optional[TimeContext],\n    **kwargs,\n):\n    \"\"\"Compute the predicates for a table operation.\n\n    Parameters\n    ----------\n    table_op : TableNode\n    predicates : List[ir.ColumnExpr]\n    data : pd.DataFrame\n    scope : Scope\n    timecontext: Optional[TimeContext]\n    kwargs : dict\n\n    Returns\n    -------\n    computed_predicate : pd.Series[bool]\n\n    Notes\n    -----\n    This handles the cases where the predicates are computed columns, in\n    addition to the simple case of named columns coming directly from the input\n    table.\n    \"\"\"\n    for predicate in predicates:\n        # Map each root table of the predicate to the data so that we compute\n        # predicates on the result instead of any left or right tables if the\n        # Selection is on a Join. Project data to only inlude columns from\n        # the root table.\n        root_tables = predicate.op().root_tables()\n\n        # handle suffixes\n        data_columns = frozenset(data.columns)\n\n        additional_scope = Scope()\n        for root_table in root_tables:\n            mapping = remap_overlapping_column_names(\n                table_op, root_table, data_columns\n            )\n            if mapping is not None:\n                new_data = data.loc[:, mapping.keys()].rename(columns=mapping)\n            else:\n                new_data = data\n            additional_scope = additional_scope.merge_scope(\n                Scope({root_table: new_data}, timecontext)\n            )\n\n        scope = scope.merge_scope(additional_scope)\n        yield execute(predicate, scope=scope, **kwargs)\n", "path": "ibis/backends/dask/execution/selection.py"}]}