We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
bug: filter() followed by drop() fails with Csv backends
This line fails with Csv backend:

https://github.com/ibis-project/ibis/pull/1967/files#diff-365b389d018694086c2001fa6cfe0d4fR39

Stacktrace:

```
ibis/tests/all/test_join.py:44: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
ibis/expr/types.py:202: in execute
    return execute(self, limit=limit, params=params, **kwargs)
ibis/client.py:285: in execute
    return backend.execute(expr, limit=limit, params=params, **kwargs)
ibis/file/client.py:38: in execute
    return execute_and_reset(expr, params=params, **kwargs)
ibis/pandas/core.py:414: in execute_and_reset
    expr, params=params, scope=scope, aggcontext=aggcontext, **kwargs
../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/multipledispatch/dispatcher.py:278: in __call__
    return func(*args, **kwargs)
ibis/pandas/core.py:371: in main_execute
    return execute_with_scope(expr, new_scope, aggcontext=aggcontext, **kwargs)
ibis/pandas/core.py:190: in execute_with_scope
    **kwargs,
ibis/pandas/core.py:226: in execute_until_in_scope
    **kwargs,
ibis/pandas/core.py:294: in execute_bottom_up
    for arg in computable_args
ibis/pandas/core.py:294: in <listcomp>
    for arg in computable_args
ibis/pandas/core.py:319: in execute_bottom_up
    **kwargs,
../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/multipledispatch/dispatcher.py:278: in __call__
    return func(*args, **kwargs)
ibis/pandas/execution/selection.py:309: in execute_selection_dataframe
    predicate = functools.reduce(operator.and_, predicates)
ibis/pandas/execution/selection.py:233: in _compute_predicates
    yield execute(predicate, scope=new_scope, **kwargs)
../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/multipledispatch/dispatcher.py:278: in __call__
    return func(*args, **kwargs)
ibis/pandas/core.py:371: in main_execute
    return execute_with_scope(expr, new_scope, aggcontext=aggcontext, **kwargs)
ibis/pandas/core.py:190: in execute_with_scope
    **kwargs,
ibis/pandas/core.py:226: in execute_until_in_scope
    **kwargs,
ibis/pandas/core.py:294: in execute_bottom_up
    for arg in computable_args
ibis/pandas/core.py:294: in <listcomp>
    for arg in computable_args
ibis/pandas/core.py:319: in execute_bottom_up
    **kwargs,
../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/multipledispatch/dispatcher.py:278: in __call__
    return func(*args, **kwargs)
ibis/pandas/execution/generic.py:379: in execute_table_column_df_or_df_groupby
    return data[op.name]
../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/pandas/core/frame.py:2927: in __getitem__
    indexer = self.columns.get_loc(key)
../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/pandas/core/indexes/base.py:2659: in get_loc
    return self._engine.get_loc(self._maybe_cast_indexer(key))
pandas/_libs/index.pyx:108: in pandas._libs.index.IndexEngine.get_loc
    ???
pandas/_libs/index.pyx:132: in pandas._libs.index.IndexEngine.get_loc
    ???
pandas/_libs/hashtable_class_helper.pxi:1601: in pandas._libs.hashtable.PyObjectHashTable.get_item
    ???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

>   ???
E   KeyError: 'lgID'
```

--- 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/csv/__init__.py`
Content:
```
1 import pandas as pd
2 import toolz
3 
4 import ibis.expr.operations as ops
5 import ibis.expr.schema as sch
6 from ibis.backends.base.file import BaseFileBackend
7 from ibis.backends.pandas.core import execute, execute_node, pre_execute
8 from ibis.backends.pandas.execution.selection import physical_tables
9 from ibis.expr.scope import Scope
10 from ibis.expr.typing import TimeContext
11 
12 
13 def _read_csv(path, schema, **kwargs):
14     dtypes = dict(schema.to_pandas())
15 
16     dates = list(toolz.valfilter(lambda s: s == 'datetime64[ns]', dtypes))
17     dtypes = toolz.dissoc(dtypes, *dates)
18 
19     return pd.read_csv(
20         str(path), dtype=dtypes, parse_dates=dates, encoding='utf-8', **kwargs
21     )
22 
23 
24 class CSVTable(ops.DatabaseTable):
25     def __init__(self, name, schema, source, **kwargs):
26         super().__init__(name, schema, source)
27         self.read_csv_kwargs = kwargs
28 
29 
30 class Backend(BaseFileBackend):
31     name = 'csv'
32     extension = 'csv'
33     table_class = CSVTable
34 
35     def insert(self, path, expr, index=False, **kwargs):
36         path = self.root / path
37         data = execute(expr)
38         data.to_csv(str(path), index=index, **kwargs)
39 
40     def table(self, name, path=None, schema=None, **kwargs):
41         if name not in self.list_tables(path):
42             raise AttributeError(name)
43 
44         if path is None:
45             path = self.root
46 
47         # get the schema
48         f = path / f"{name}.{self.extension}"
49 
50         # read sample
51         schema = schema or sch.schema([])
52         sample = _read_csv(f, schema=schema, header=0, nrows=50, **kwargs)
53 
54         # infer sample's schema and define table
55         schema = sch.infer(sample, schema=schema)
56         table = self.table_class(name, schema, self, **kwargs).to_expr()
57 
58         self.dictionary[name] = f
59 
60         return table
61 
62 
63 @pre_execute.register(ops.Selection, Backend)
64 def csv_pre_execute_selection(
65     op: ops.Node,
66     client: Backend,
67     scope: Scope,
68     timecontext: TimeContext = None,
69     **kwargs,
70 ):
71     tables = filter(
72         lambda t: scope.get_value(t, timecontext) is None,
73         physical_tables(op.table.op()),
74     )
75 
76     ops = Scope()
77     for table in tables:
78         path = client.dictionary[table.name]
79         usecols = None
80 
81         if op.selections:
82             header = _read_csv(path, schema=table.schema, header=0, nrows=1)
83             usecols = [
84                 getattr(s.op(), 'name', None) or s.get_name()
85                 for s in op.selections
86             ]
87 
88             # we cannot read all the columns that we would like
89             if len(pd.Index(usecols) & header.columns) != len(usecols):
90                 usecols = None
91         result = _read_csv(path, table.schema, usecols=usecols, header=0)
92         ops = ops.merge_scope(Scope({table: result}, timecontext))
93 
94     return ops
95 
96 
97 @execute_node.register(Backend.table_class, Backend)
98 def csv_read_table(op, client, scope, **kwargs):
99     path = client.dictionary[op.name]
100     df = _read_csv(path, schema=op.schema, header=0, **op.read_csv_kwargs)
101     return df
102
```
--- 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/csv/__init__.py b/ibis/backends/csv/__init__.py
--- a/ibis/backends/csv/__init__.py
+++ b/ibis/backends/csv/__init__.py
@@ -4,10 +4,7 @@
 import ibis.expr.operations as ops
 import ibis.expr.schema as sch
 from ibis.backends.base.file import BaseFileBackend
-from ibis.backends.pandas.core import execute, execute_node, pre_execute
-from ibis.backends.pandas.execution.selection import physical_tables
-from ibis.expr.scope import Scope
-from ibis.expr.typing import TimeContext
+from ibis.backends.pandas.core import execute, execute_node
 
 
 def _read_csv(path, schema, **kwargs):
@@ -60,40 +57,6 @@
         return table
 
 
-@pre_execute.register(ops.Selection, Backend)
-def csv_pre_execute_selection(
-    op: ops.Node,
-    client: Backend,
-    scope: Scope,
-    timecontext: TimeContext = None,
-    **kwargs,
-):
-    tables = filter(
-        lambda t: scope.get_value(t, timecontext) is None,
-        physical_tables(op.table.op()),
-    )
-
-    ops = Scope()
-    for table in tables:
-        path = client.dictionary[table.name]
-        usecols = None
-
-        if op.selections:
-            header = _read_csv(path, schema=table.schema, header=0, nrows=1)
-            usecols = [
-                getattr(s.op(), 'name', None) or s.get_name()
-                for s in op.selections
-            ]
-
-            # we cannot read all the columns that we would like
-            if len(pd.Index(usecols) & header.columns) != len(usecols):
-                usecols = None
-        result = _read_csv(path, table.schema, usecols=usecols, header=0)
-        ops = ops.merge_scope(Scope({table: result}, timecontext))
-
-    return ops
-
-
 @execute_node.register(Backend.table_class, Backend)
 def csv_read_table(op, client, scope, **kwargs):
     path = client.dictionary[op.name]

{"golden_diff": "diff --git a/ibis/backends/csv/__init__.py b/ibis/backends/csv/__init__.py\n--- a/ibis/backends/csv/__init__.py\n+++ b/ibis/backends/csv/__init__.py\n@@ -4,10 +4,7 @@\n import ibis.expr.operations as ops\n import ibis.expr.schema as sch\n from ibis.backends.base.file import BaseFileBackend\n-from ibis.backends.pandas.core import execute, execute_node, pre_execute\n-from ibis.backends.pandas.execution.selection import physical_tables\n-from ibis.expr.scope import Scope\n-from ibis.expr.typing import TimeContext\n+from ibis.backends.pandas.core import execute, execute_node\n \n \n def _read_csv(path, schema, **kwargs):\n@@ -60,40 +57,6 @@\n         return table\n \n \n-@pre_execute.register(ops.Selection, Backend)\n-def csv_pre_execute_selection(\n-    op: ops.Node,\n-    client: Backend,\n-    scope: Scope,\n-    timecontext: TimeContext = None,\n-    **kwargs,\n-):\n-    tables = filter(\n-        lambda t: scope.get_value(t, timecontext) is None,\n-        physical_tables(op.table.op()),\n-    )\n-\n-    ops = Scope()\n-    for table in tables:\n-        path = client.dictionary[table.name]\n-        usecols = None\n-\n-        if op.selections:\n-            header = _read_csv(path, schema=table.schema, header=0, nrows=1)\n-            usecols = [\n-                getattr(s.op(), 'name', None) or s.get_name()\n-                for s in op.selections\n-            ]\n-\n-            # we cannot read all the columns that we would like\n-            if len(pd.Index(usecols) & header.columns) != len(usecols):\n-                usecols = None\n-        result = _read_csv(path, table.schema, usecols=usecols, header=0)\n-        ops = ops.merge_scope(Scope({table: result}, timecontext))\n-\n-    return ops\n-\n-\n @execute_node.register(Backend.table_class, Backend)\n def csv_read_table(op, client, scope, **kwargs):\n     path = client.dictionary[op.name]\n", "issue": "bug: filter() followed by drop() fails with Csv backends\nThis line fails with Csv backend:\r\n\r\nhttps://github.com/ibis-project/ibis/pull/1967/files#diff-365b389d018694086c2001fa6cfe0d4fR39\r\n\r\nStacktrace:\r\n\r\n```\r\nibis/tests/all/test_join.py:44: \r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \r\nibis/expr/types.py:202: in execute\r\n    return execute(self, limit=limit, params=params, **kwargs)\r\nibis/client.py:285: in execute\r\n    return backend.execute(expr, limit=limit, params=params, **kwargs)\r\nibis/file/client.py:38: in execute\r\n    return execute_and_reset(expr, params=params, **kwargs)\r\nibis/pandas/core.py:414: in execute_and_reset\r\n    expr, params=params, scope=scope, aggcontext=aggcontext, **kwargs\r\n../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/multipledispatch/dispatcher.py:278: in __call__\r\n    return func(*args, **kwargs)\r\nibis/pandas/core.py:371: in main_execute\r\n    return execute_with_scope(expr, new_scope, aggcontext=aggcontext, **kwargs)\r\nibis/pandas/core.py:190: in execute_with_scope\r\n    **kwargs,\r\nibis/pandas/core.py:226: in execute_until_in_scope\r\n    **kwargs,\r\nibis/pandas/core.py:294: in execute_bottom_up\r\n    for arg in computable_args\r\nibis/pandas/core.py:294: in <listcomp>\r\n    for arg in computable_args\r\nibis/pandas/core.py:319: in execute_bottom_up\r\n    **kwargs,\r\n../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/multipledispatch/dispatcher.py:278: in __call__\r\n    return func(*args, **kwargs)\r\nibis/pandas/execution/selection.py:309: in execute_selection_dataframe\r\n    predicate = functools.reduce(operator.and_, predicates)\r\nibis/pandas/execution/selection.py:233: in _compute_predicates\r\n    yield execute(predicate, scope=new_scope, **kwargs)\r\n../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/multipledispatch/dispatcher.py:278: in __call__\r\n    return func(*args, **kwargs)\r\nibis/pandas/core.py:371: in main_execute\r\n    return execute_with_scope(expr, new_scope, aggcontext=aggcontext, **kwargs)\r\nibis/pandas/core.py:190: in execute_with_scope\r\n    **kwargs,\r\nibis/pandas/core.py:226: in execute_until_in_scope\r\n    **kwargs,\r\nibis/pandas/core.py:294: in execute_bottom_up\r\n    for arg in computable_args\r\nibis/pandas/core.py:294: in <listcomp>\r\n    for arg in computable_args\r\nibis/pandas/core.py:319: in execute_bottom_up\r\n    **kwargs,\r\n../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/multipledispatch/dispatcher.py:278: in __call__\r\n    return func(*args, **kwargs)\r\nibis/pandas/execution/generic.py:379: in execute_table_column_df_or_df_groupby\r\n    return data[op.name]\r\n../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/pandas/core/frame.py:2927: in __getitem__\r\n    indexer = self.columns.get_loc(key)\r\n../../miniconda3/envs/ibis-dev/lib/python3.7/site-packages/pandas/core/indexes/base.py:2659: in get_loc\r\n    return self._engine.get_loc(self._maybe_cast_indexer(key))\r\npandas/_libs/index.pyx:108: in pandas._libs.index.IndexEngine.get_loc\r\n    ???\r\npandas/_libs/index.pyx:132: in pandas._libs.index.IndexEngine.get_loc\r\n    ???\r\npandas/_libs/hashtable_class_helper.pxi:1601: in pandas._libs.hashtable.PyObjectHashTable.get_item\r\n    ???\r\n_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ \r\n\r\n>   ???\r\nE   KeyError: 'lgID'\r\n```\n", "before_files": [{"content": "import pandas as pd\nimport toolz\n\nimport ibis.expr.operations as ops\nimport ibis.expr.schema as sch\nfrom ibis.backends.base.file import BaseFileBackend\nfrom ibis.backends.pandas.core import execute, execute_node, pre_execute\nfrom ibis.backends.pandas.execution.selection import physical_tables\nfrom ibis.expr.scope import Scope\nfrom ibis.expr.typing import TimeContext\n\n\ndef _read_csv(path, schema, **kwargs):\n    dtypes = dict(schema.to_pandas())\n\n    dates = list(toolz.valfilter(lambda s: s == 'datetime64[ns]', dtypes))\n    dtypes = toolz.dissoc(dtypes, *dates)\n\n    return pd.read_csv(\n        str(path), dtype=dtypes, parse_dates=dates, encoding='utf-8', **kwargs\n    )\n\n\nclass CSVTable(ops.DatabaseTable):\n    def __init__(self, name, schema, source, **kwargs):\n        super().__init__(name, schema, source)\n        self.read_csv_kwargs = kwargs\n\n\nclass Backend(BaseFileBackend):\n    name = 'csv'\n    extension = 'csv'\n    table_class = CSVTable\n\n    def insert(self, path, expr, index=False, **kwargs):\n        path = self.root / path\n        data = execute(expr)\n        data.to_csv(str(path), index=index, **kwargs)\n\n    def table(self, name, path=None, schema=None, **kwargs):\n        if name not in self.list_tables(path):\n            raise AttributeError(name)\n\n        if path is None:\n            path = self.root\n\n        # get the schema\n        f = path / f\"{name}.{self.extension}\"\n\n        # read sample\n        schema = schema or sch.schema([])\n        sample = _read_csv(f, schema=schema, header=0, nrows=50, **kwargs)\n\n        # infer sample's schema and define table\n        schema = sch.infer(sample, schema=schema)\n        table = self.table_class(name, schema, self, **kwargs).to_expr()\n\n        self.dictionary[name] = f\n\n        return table\n\n\n@pre_execute.register(ops.Selection, Backend)\ndef csv_pre_execute_selection(\n    op: ops.Node,\n    client: Backend,\n    scope: Scope,\n    timecontext: TimeContext = None,\n    **kwargs,\n):\n    tables = filter(\n        lambda t: scope.get_value(t, timecontext) is None,\n        physical_tables(op.table.op()),\n    )\n\n    ops = Scope()\n    for table in tables:\n        path = client.dictionary[table.name]\n        usecols = None\n\n        if op.selections:\n            header = _read_csv(path, schema=table.schema, header=0, nrows=1)\n            usecols = [\n                getattr(s.op(), 'name', None) or s.get_name()\n                for s in op.selections\n            ]\n\n            # we cannot read all the columns that we would like\n            if len(pd.Index(usecols) & header.columns) != len(usecols):\n                usecols = None\n        result = _read_csv(path, table.schema, usecols=usecols, header=0)\n        ops = ops.merge_scope(Scope({table: result}, timecontext))\n\n    return ops\n\n\n@execute_node.register(Backend.table_class, Backend)\ndef csv_read_table(op, client, scope, **kwargs):\n    path = client.dictionary[op.name]\n    df = _read_csv(path, schema=op.schema, header=0, **op.read_csv_kwargs)\n    return df\n", "path": "ibis/backends/csv/__init__.py"}], "after_files": [{"content": "import pandas as pd\nimport toolz\n\nimport ibis.expr.operations as ops\nimport ibis.expr.schema as sch\nfrom ibis.backends.base.file import BaseFileBackend\nfrom ibis.backends.pandas.core import execute, execute_node\n\n\ndef _read_csv(path, schema, **kwargs):\n    dtypes = dict(schema.to_pandas())\n\n    dates = list(toolz.valfilter(lambda s: s == 'datetime64[ns]', dtypes))\n    dtypes = toolz.dissoc(dtypes, *dates)\n\n    return pd.read_csv(\n        str(path), dtype=dtypes, parse_dates=dates, encoding='utf-8', **kwargs\n    )\n\n\nclass CSVTable(ops.DatabaseTable):\n    def __init__(self, name, schema, source, **kwargs):\n        super().__init__(name, schema, source)\n        self.read_csv_kwargs = kwargs\n\n\nclass Backend(BaseFileBackend):\n    name = 'csv'\n    extension = 'csv'\n    table_class = CSVTable\n\n    def insert(self, path, expr, index=False, **kwargs):\n        path = self.root / path\n        data = execute(expr)\n        data.to_csv(str(path), index=index, **kwargs)\n\n    def table(self, name, path=None, schema=None, **kwargs):\n        if name not in self.list_tables(path):\n            raise AttributeError(name)\n\n        if path is None:\n            path = self.root\n\n        # get the schema\n        f = path / f\"{name}.{self.extension}\"\n\n        # read sample\n        schema = schema or sch.schema([])\n        sample = _read_csv(f, schema=schema, header=0, nrows=50, **kwargs)\n\n        # infer sample's schema and define table\n        schema = sch.infer(sample, schema=schema)\n        table = self.table_class(name, schema, self, **kwargs).to_expr()\n\n        self.dictionary[name] = f\n\n        return table\n\n\n@execute_node.register(Backend.table_class, Backend)\ndef csv_read_table(op, client, scope, **kwargs):\n    path = client.dictionary[op.name]\n    df = _read_csv(path, schema=op.schema, header=0, **op.read_csv_kwargs)\n    return df\n", "path": "ibis/backends/csv/__init__.py"}]}