We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] Setitem for DataFrame leads to a wrong dtypes
<!--
Thank you for your contribution!

Please review https://github.com/mars-project/mars/blob/master/CONTRIBUTING.rst before opening an issue.
-->

**Describe the bug**
Add columns for a DataFrame will lead to a wrong dtypes of input DataFrame.

**To Reproduce**
```python
In [1]: import mars.dataframe as md                                                                 

In [2]: a = md.DataFrame({'a':[1,2,3]})                                                             

In [3]: a['new'] = 1                                                                                

In [4]: a.op.inputs                                                                                 
Out[4]: [DataFrame <op=DataFrameDataSource, key=c212164d24d96ed634711c3b97f334cb>]

In [5]: a.op.inputs[0].dtypes                                                                       
Out[5]: 
a      int64
new    int64
dtype: object
```
**Expected behavior**
Input DataFrame's dtypes should have only one column.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `mars/dataframe/indexing/setitem.py`
Content:
```
1 # Copyright 1999-2020 Alibaba Group Holding Ltd.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import numpy as np
16 import pandas as pd
17 from pandas.api.types import is_list_like
18 
19 from ... import opcodes
20 from ...core import OutputType
21 from ...serialize import KeyField, AnyField
22 from ...tensor.core import TENSOR_TYPE
23 from ...tiles import TilesError
24 from ..core import SERIES_TYPE, DataFrame
25 from ..initializer import Series as asseries
26 from ..operands import DataFrameOperand, DataFrameOperandMixin
27 from ..utils import parse_index
28 
29 
30 class DataFrameSetitem(DataFrameOperand, DataFrameOperandMixin):
31     _op_type_ = opcodes.INDEXSETVALUE
32 
33     _target = KeyField('target')
34     _indexes = AnyField('indexes')
35     _value = AnyField('value')
36 
37     def __init__(self, target=None, indexes=None, value=None, output_types=None, **kw):
38         super().__init__(_target=target, _indexes=indexes,
39                          _value=value, _output_types=output_types, **kw)
40         if self.output_types is None:
41             self.output_types = [OutputType.dataframe]
42 
43     @property
44     def target(self):
45         return self._target
46 
47     @property
48     def indexes(self):
49         return self._indexes
50 
51     @property
52     def value(self):
53         return self._value
54 
55     def _set_inputs(self, inputs):
56         super()._set_inputs(inputs)
57         self._target = self._inputs[0]
58         if len(inputs) > 1:
59             self._value = self._inputs[-1]
60 
61     def __call__(self, target: DataFrame, value):
62         inputs = [target]
63         if np.isscalar(value):
64             value_dtype = np.array(value).dtype
65         else:
66             if isinstance(value, (pd.Series, SERIES_TYPE)):
67                 value = asseries(value)
68                 inputs.append(value)
69                 value_dtype = value.dtype
70             elif is_list_like(value) or isinstance(value, TENSOR_TYPE):
71                 value = asseries(value, index=target.index)
72                 inputs.append(value)
73                 value_dtype = value.dtype
74             else:  # pragma: no cover
75                 raise TypeError('Wrong value type, could be one of scalar, Series or tensor')
76 
77             if value.index_value.key != target.index_value.key:  # pragma: no cover
78                 raise NotImplementedError('Does not support setting value '
79                                           'with different index for now')
80 
81         index_value = target.index_value
82         dtypes = target.dtypes
83         dtypes.loc[self._indexes] = value_dtype
84         columns_value = parse_index(dtypes.index, store_data=True)
85         ret = self.new_dataframe(inputs, shape=(target.shape[0], len(dtypes)),
86                                  dtypes=dtypes, index_value=index_value,
87                                  columns_value=columns_value)
88         target.data = ret.data
89 
90     @classmethod
91     def tile(cls, op):
92         out = op.outputs[0]
93         target = op.target
94         value = op.value
95         col = op.indexes
96         columns = target.columns_value.to_pandas()
97 
98         if not np.isscalar(value):
99             # check if all chunk's index_value are identical
100             target_chunk_index_values = [c.index_value for c in target.chunks
101                                          if c.index[1] == 0]
102             value_chunk_index_values = [v.index_value for v in value.chunks]
103             is_identical = len(target_chunk_index_values) == len(target_chunk_index_values) and \
104                 all(c.key == v.key for c, v in zip(target_chunk_index_values, value_chunk_index_values))
105             if not is_identical:
106                 # do rechunk
107                 if any(np.isnan(s) for s in target.nsplits[0]) or \
108                         any(np.isnan(s) for s in value.nsplits[0]):  # pragma: no cover
109                     raise TilesError('target or value has unknown chunk shape')
110 
111                 value = value.rechunk({0: target.nsplits[0]})._inplace_tile()
112 
113         out_chunks = []
114         nsplits = [list(ns) for ns in target.nsplits]
115         if col not in columns:
116             nsplits[1][-1] += 1
117             column_chunk_shape = target.chunk_shape[1]
118             # append to the last chunk on columns axis direction
119             for c in target.chunks:
120                 if c.index[-1] != column_chunk_shape - 1:
121                     # not effected, just output
122                     out_chunks.append(c)
123                 else:
124                     chunk_op = op.copy().reset_key()
125                     if np.isscalar(value):
126                         chunk_inputs = [c]
127                     else:
128                         value_chunk = value.cix[c.index[0], ]
129                         chunk_inputs = [c, value_chunk]
130 
131                     dtypes = c.dtypes
132                     dtypes.loc[out.dtypes.index[-1]] = out.dtypes.iloc[-1]
133                     chunk = chunk_op.new_chunk(chunk_inputs,
134                                                shape=(c.shape[0], c.shape[1] + 1),
135                                                dtypes=dtypes,
136                                                index_value=c.index_value,
137                                                columns_value=parse_index(dtypes.index, store_data=True),
138                                                index=c.index)
139                     out_chunks.append(chunk)
140         else:
141             # replace exist column
142             for c in target.chunks:
143                 if col in c.dtypes:
144                     chunk_inputs = [c]
145                     if not np.isscalar(value):
146                         chunk_inputs.append(value.cix[c.index[0], ])
147                     chunk_op = op.copy().reset_key()
148                     chunk = chunk_op.new_chunk(chunk_inputs,
149                                                shape=c.shape,
150                                                dtypes=c.dtypes,
151                                                index_value=c.index_value,
152                                                columns_value=c.columns_value,
153                                                index=c.index)
154                     out_chunks.append(chunk)
155                 else:
156                     out_chunks.append(c)
157 
158         params = out.params
159         params['nsplits'] = tuple(tuple(ns) for ns in nsplits)
160         params['chunks'] = out_chunks
161         new_op = op.copy()
162         return new_op.new_tileables(op.inputs, kws=[params])
163 
164     @classmethod
165     def execute(cls, ctx, op):
166         target = ctx[op.target.key].copy()
167         value = ctx[op.value.key] if not np.isscalar(op.value) else op.value
168         target[op.indexes] = value
169         ctx[op.outputs[0].key] = target
170 
171 
172 def dataframe_setitem(df, col, value):
173     op = DataFrameSetitem(target=df, indexes=col, value=value)
174     return op(df, value)
175
```
--- 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/mars/dataframe/indexing/setitem.py b/mars/dataframe/indexing/setitem.py
--- a/mars/dataframe/indexing/setitem.py
+++ b/mars/dataframe/indexing/setitem.py
@@ -79,7 +79,7 @@
                                           'with different index for now')
 
         index_value = target.index_value
-        dtypes = target.dtypes
+        dtypes = target.dtypes.copy(deep=True)
         dtypes.loc[self._indexes] = value_dtype
         columns_value = parse_index(dtypes.index, store_data=True)
         ret = self.new_dataframe(inputs, shape=(target.shape[0], len(dtypes)),

{"golden_diff": "diff --git a/mars/dataframe/indexing/setitem.py b/mars/dataframe/indexing/setitem.py\n--- a/mars/dataframe/indexing/setitem.py\n+++ b/mars/dataframe/indexing/setitem.py\n@@ -79,7 +79,7 @@\n                                           'with different index for now')\n \n         index_value = target.index_value\n-        dtypes = target.dtypes\n+        dtypes = target.dtypes.copy(deep=True)\n         dtypes.loc[self._indexes] = value_dtype\n         columns_value = parse_index(dtypes.index, store_data=True)\n         ret = self.new_dataframe(inputs, shape=(target.shape[0], len(dtypes)),\n", "issue": "[BUG] Setitem for DataFrame leads to a wrong dtypes\n<!--\r\nThank you for your contribution!\r\n\r\nPlease review https://github.com/mars-project/mars/blob/master/CONTRIBUTING.rst before opening an issue.\r\n-->\r\n\r\n**Describe the bug**\r\nAdd columns for a DataFrame will lead to a wrong dtypes of input DataFrame.\r\n\r\n**To Reproduce**\r\n```python\r\nIn [1]: import mars.dataframe as md                                                                 \r\n\r\nIn [2]: a = md.DataFrame({'a':[1,2,3]})                                                             \r\n\r\nIn [3]: a['new'] = 1                                                                                \r\n\r\nIn [4]: a.op.inputs                                                                                 \r\nOut[4]: [DataFrame <op=DataFrameDataSource, key=c212164d24d96ed634711c3b97f334cb>]\r\n\r\nIn [5]: a.op.inputs[0].dtypes                                                                       \r\nOut[5]: \r\na      int64\r\nnew    int64\r\ndtype: object\r\n```\r\n**Expected behavior**\r\nInput DataFrame's dtypes should have only one column.\r\n\n", "before_files": [{"content": "# Copyright 1999-2020 Alibaba Group Holding Ltd.\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 numpy as np\nimport pandas as pd\nfrom pandas.api.types import is_list_like\n\nfrom ... import opcodes\nfrom ...core import OutputType\nfrom ...serialize import KeyField, AnyField\nfrom ...tensor.core import TENSOR_TYPE\nfrom ...tiles import TilesError\nfrom ..core import SERIES_TYPE, DataFrame\nfrom ..initializer import Series as asseries\nfrom ..operands import DataFrameOperand, DataFrameOperandMixin\nfrom ..utils import parse_index\n\n\nclass DataFrameSetitem(DataFrameOperand, DataFrameOperandMixin):\n    _op_type_ = opcodes.INDEXSETVALUE\n\n    _target = KeyField('target')\n    _indexes = AnyField('indexes')\n    _value = AnyField('value')\n\n    def __init__(self, target=None, indexes=None, value=None, output_types=None, **kw):\n        super().__init__(_target=target, _indexes=indexes,\n                         _value=value, _output_types=output_types, **kw)\n        if self.output_types is None:\n            self.output_types = [OutputType.dataframe]\n\n    @property\n    def target(self):\n        return self._target\n\n    @property\n    def indexes(self):\n        return self._indexes\n\n    @property\n    def value(self):\n        return self._value\n\n    def _set_inputs(self, inputs):\n        super()._set_inputs(inputs)\n        self._target = self._inputs[0]\n        if len(inputs) > 1:\n            self._value = self._inputs[-1]\n\n    def __call__(self, target: DataFrame, value):\n        inputs = [target]\n        if np.isscalar(value):\n            value_dtype = np.array(value).dtype\n        else:\n            if isinstance(value, (pd.Series, SERIES_TYPE)):\n                value = asseries(value)\n                inputs.append(value)\n                value_dtype = value.dtype\n            elif is_list_like(value) or isinstance(value, TENSOR_TYPE):\n                value = asseries(value, index=target.index)\n                inputs.append(value)\n                value_dtype = value.dtype\n            else:  # pragma: no cover\n                raise TypeError('Wrong value type, could be one of scalar, Series or tensor')\n\n            if value.index_value.key != target.index_value.key:  # pragma: no cover\n                raise NotImplementedError('Does not support setting value '\n                                          'with different index for now')\n\n        index_value = target.index_value\n        dtypes = target.dtypes\n        dtypes.loc[self._indexes] = value_dtype\n        columns_value = parse_index(dtypes.index, store_data=True)\n        ret = self.new_dataframe(inputs, shape=(target.shape[0], len(dtypes)),\n                                 dtypes=dtypes, index_value=index_value,\n                                 columns_value=columns_value)\n        target.data = ret.data\n\n    @classmethod\n    def tile(cls, op):\n        out = op.outputs[0]\n        target = op.target\n        value = op.value\n        col = op.indexes\n        columns = target.columns_value.to_pandas()\n\n        if not np.isscalar(value):\n            # check if all chunk's index_value are identical\n            target_chunk_index_values = [c.index_value for c in target.chunks\n                                         if c.index[1] == 0]\n            value_chunk_index_values = [v.index_value for v in value.chunks]\n            is_identical = len(target_chunk_index_values) == len(target_chunk_index_values) and \\\n                all(c.key == v.key for c, v in zip(target_chunk_index_values, value_chunk_index_values))\n            if not is_identical:\n                # do rechunk\n                if any(np.isnan(s) for s in target.nsplits[0]) or \\\n                        any(np.isnan(s) for s in value.nsplits[0]):  # pragma: no cover\n                    raise TilesError('target or value has unknown chunk shape')\n\n                value = value.rechunk({0: target.nsplits[0]})._inplace_tile()\n\n        out_chunks = []\n        nsplits = [list(ns) for ns in target.nsplits]\n        if col not in columns:\n            nsplits[1][-1] += 1\n            column_chunk_shape = target.chunk_shape[1]\n            # append to the last chunk on columns axis direction\n            for c in target.chunks:\n                if c.index[-1] != column_chunk_shape - 1:\n                    # not effected, just output\n                    out_chunks.append(c)\n                else:\n                    chunk_op = op.copy().reset_key()\n                    if np.isscalar(value):\n                        chunk_inputs = [c]\n                    else:\n                        value_chunk = value.cix[c.index[0], ]\n                        chunk_inputs = [c, value_chunk]\n\n                    dtypes = c.dtypes\n                    dtypes.loc[out.dtypes.index[-1]] = out.dtypes.iloc[-1]\n                    chunk = chunk_op.new_chunk(chunk_inputs,\n                                               shape=(c.shape[0], c.shape[1] + 1),\n                                               dtypes=dtypes,\n                                               index_value=c.index_value,\n                                               columns_value=parse_index(dtypes.index, store_data=True),\n                                               index=c.index)\n                    out_chunks.append(chunk)\n        else:\n            # replace exist column\n            for c in target.chunks:\n                if col in c.dtypes:\n                    chunk_inputs = [c]\n                    if not np.isscalar(value):\n                        chunk_inputs.append(value.cix[c.index[0], ])\n                    chunk_op = op.copy().reset_key()\n                    chunk = chunk_op.new_chunk(chunk_inputs,\n                                               shape=c.shape,\n                                               dtypes=c.dtypes,\n                                               index_value=c.index_value,\n                                               columns_value=c.columns_value,\n                                               index=c.index)\n                    out_chunks.append(chunk)\n                else:\n                    out_chunks.append(c)\n\n        params = out.params\n        params['nsplits'] = tuple(tuple(ns) for ns in nsplits)\n        params['chunks'] = out_chunks\n        new_op = op.copy()\n        return new_op.new_tileables(op.inputs, kws=[params])\n\n    @classmethod\n    def execute(cls, ctx, op):\n        target = ctx[op.target.key].copy()\n        value = ctx[op.value.key] if not np.isscalar(op.value) else op.value\n        target[op.indexes] = value\n        ctx[op.outputs[0].key] = target\n\n\ndef dataframe_setitem(df, col, value):\n    op = DataFrameSetitem(target=df, indexes=col, value=value)\n    return op(df, value)\n", "path": "mars/dataframe/indexing/setitem.py"}], "after_files": [{"content": "# Copyright 1999-2020 Alibaba Group Holding Ltd.\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 numpy as np\nimport pandas as pd\nfrom pandas.api.types import is_list_like\n\nfrom ... import opcodes\nfrom ...core import OutputType\nfrom ...serialize import KeyField, AnyField\nfrom ...tensor.core import TENSOR_TYPE\nfrom ...tiles import TilesError\nfrom ..core import SERIES_TYPE, DataFrame\nfrom ..initializer import Series as asseries\nfrom ..operands import DataFrameOperand, DataFrameOperandMixin\nfrom ..utils import parse_index\n\n\nclass DataFrameSetitem(DataFrameOperand, DataFrameOperandMixin):\n    _op_type_ = opcodes.INDEXSETVALUE\n\n    _target = KeyField('target')\n    _indexes = AnyField('indexes')\n    _value = AnyField('value')\n\n    def __init__(self, target=None, indexes=None, value=None, output_types=None, **kw):\n        super().__init__(_target=target, _indexes=indexes,\n                         _value=value, _output_types=output_types, **kw)\n        if self.output_types is None:\n            self.output_types = [OutputType.dataframe]\n\n    @property\n    def target(self):\n        return self._target\n\n    @property\n    def indexes(self):\n        return self._indexes\n\n    @property\n    def value(self):\n        return self._value\n\n    def _set_inputs(self, inputs):\n        super()._set_inputs(inputs)\n        self._target = self._inputs[0]\n        if len(inputs) > 1:\n            self._value = self._inputs[-1]\n\n    def __call__(self, target: DataFrame, value):\n        inputs = [target]\n        if np.isscalar(value):\n            value_dtype = np.array(value).dtype\n        else:\n            if isinstance(value, (pd.Series, SERIES_TYPE)):\n                value = asseries(value)\n                inputs.append(value)\n                value_dtype = value.dtype\n            elif is_list_like(value) or isinstance(value, TENSOR_TYPE):\n                value = asseries(value, index=target.index)\n                inputs.append(value)\n                value_dtype = value.dtype\n            else:  # pragma: no cover\n                raise TypeError('Wrong value type, could be one of scalar, Series or tensor')\n\n            if value.index_value.key != target.index_value.key:  # pragma: no cover\n                raise NotImplementedError('Does not support setting value '\n                                          'with different index for now')\n\n        index_value = target.index_value\n        dtypes = target.dtypes.copy(deep=True)\n        dtypes.loc[self._indexes] = value_dtype\n        columns_value = parse_index(dtypes.index, store_data=True)\n        ret = self.new_dataframe(inputs, shape=(target.shape[0], len(dtypes)),\n                                 dtypes=dtypes, index_value=index_value,\n                                 columns_value=columns_value)\n        target.data = ret.data\n\n    @classmethod\n    def tile(cls, op):\n        out = op.outputs[0]\n        target = op.target\n        value = op.value\n        col = op.indexes\n        columns = target.columns_value.to_pandas()\n\n        if not np.isscalar(value):\n            # check if all chunk's index_value are identical\n            target_chunk_index_values = [c.index_value for c in target.chunks\n                                         if c.index[1] == 0]\n            value_chunk_index_values = [v.index_value for v in value.chunks]\n            is_identical = len(target_chunk_index_values) == len(target_chunk_index_values) and \\\n                all(c.key == v.key for c, v in zip(target_chunk_index_values, value_chunk_index_values))\n            if not is_identical:\n                # do rechunk\n                if any(np.isnan(s) for s in target.nsplits[0]) or \\\n                        any(np.isnan(s) for s in value.nsplits[0]):  # pragma: no cover\n                    raise TilesError('target or value has unknown chunk shape')\n\n                value = value.rechunk({0: target.nsplits[0]})._inplace_tile()\n\n        out_chunks = []\n        nsplits = [list(ns) for ns in target.nsplits]\n        if col not in columns:\n            nsplits[1][-1] += 1\n            column_chunk_shape = target.chunk_shape[1]\n            # append to the last chunk on columns axis direction\n            for c in target.chunks:\n                if c.index[-1] != column_chunk_shape - 1:\n                    # not effected, just output\n                    out_chunks.append(c)\n                else:\n                    chunk_op = op.copy().reset_key()\n                    if np.isscalar(value):\n                        chunk_inputs = [c]\n                    else:\n                        value_chunk = value.cix[c.index[0], ]\n                        chunk_inputs = [c, value_chunk]\n\n                    dtypes = c.dtypes\n                    dtypes.loc[out.dtypes.index[-1]] = out.dtypes.iloc[-1]\n                    chunk = chunk_op.new_chunk(chunk_inputs,\n                                               shape=(c.shape[0], c.shape[1] + 1),\n                                               dtypes=dtypes,\n                                               index_value=c.index_value,\n                                               columns_value=parse_index(dtypes.index, store_data=True),\n                                               index=c.index)\n                    out_chunks.append(chunk)\n        else:\n            # replace exist column\n            for c in target.chunks:\n                if col in c.dtypes:\n                    chunk_inputs = [c]\n                    if not np.isscalar(value):\n                        chunk_inputs.append(value.cix[c.index[0], ])\n                    chunk_op = op.copy().reset_key()\n                    chunk = chunk_op.new_chunk(chunk_inputs,\n                                               shape=c.shape,\n                                               dtypes=c.dtypes,\n                                               index_value=c.index_value,\n                                               columns_value=c.columns_value,\n                                               index=c.index)\n                    out_chunks.append(chunk)\n                else:\n                    out_chunks.append(c)\n\n        params = out.params\n        params['nsplits'] = tuple(tuple(ns) for ns in nsplits)\n        params['chunks'] = out_chunks\n        new_op = op.copy()\n        return new_op.new_tileables(op.inputs, kws=[params])\n\n    @classmethod\n    def execute(cls, ctx, op):\n        target = ctx[op.target.key].copy()\n        value = ctx[op.value.key] if not np.isscalar(op.value) else op.value\n        target[op.indexes] = value\n        ctx[op.outputs[0].key] = target\n\n\ndef dataframe_setitem(df, col, value):\n    op = DataFrameSetitem(target=df, indexes=col, value=value)\n    return op(df, value)\n", "path": "mars/dataframe/indexing/setitem.py"}]}