We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[BUG] Tile of rechunk() outputs inconsistent dtypes and columns_value
**Describe the bug**
DataFrame.rechunk() produces chunks with different number of dtypes and columns_value, which may impair descendant chunks.

Code to reproduce this:

```python                                                                                                                                                                   
In [2]: import pandas as pd                                                                                                                                                                                          

In [4]: from mars.dataframe.datasource.dataframe import from_pandas as from_pandas_df 

In [6]: cols = [chr(ord('A') + i) for i in range(10)] 

In [8]: df_raw = pd.DataFrame(dict((c, [i ** 2 for i in range(20)]) for c in cols))                                                                                                                           

In [9]: df = from_pandas_df(df_raw, chunk_size=5)       

In [13]: rechunked = df.rechunk((20, 1)).tiles()                                                                                                                                                              

In [14]: rechunked.chunks[0].columns_value.to_pandas()                                                                                                                                                        
Out[14]: Index(['A'], dtype='object')

In [15]: rechunked.chunks[0].dtypes                                                                                                                                                                           
Out[15]: 
A    int64
B    int64
C    int64
D    int64
E    int64
dtype: object
```

--- 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/base/rechunk.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 itertools
16 
17 from ... import opcodes as OperandDef
18 from ...serialize import KeyField, AnyField, Int32Field, Int64Field
19 from ...tensor.rechunk.core import get_nsplits, plan_rechunks, compute_rechunk_slices
20 from ...tensor.utils import calc_sliced_size
21 from ...utils import check_chunks_unknown_shape
22 from ...tiles import TilesError
23 from ..operands import DataFrameOperand, DataFrameOperandMixin, DATAFRAME_TYPE, ObjectType
24 from ..utils import indexing_index_value, merge_index_value
25 
26 
27 class DataFrameRechunk(DataFrameOperand, DataFrameOperandMixin):
28     _op_type_ = OperandDef.RECHUNK
29 
30     _input = KeyField('input')
31     _chunk_size = AnyField('chunk_size')
32     _threshold = Int32Field('threshold')
33     _chunk_size_limit = Int64Field('chunk_size_limit')
34 
35     def __init__(self, chunk_size=None, threshold=None, chunk_size_limit=None, object_type=None, **kw):
36         super().__init__(_chunk_size=chunk_size, _threshold=threshold,
37                          _chunk_size_limit=chunk_size_limit, _object_type=object_type, **kw)
38 
39     @property
40     def chunk_size(self):
41         return self._chunk_size
42 
43     @property
44     def threshold(self):
45         return self._threshold
46 
47     @property
48     def chunk_size_limit(self):
49         return self._chunk_size_limit
50 
51     def _set_inputs(self, inputs):
52         super()._set_inputs(inputs)
53         self._input = self._inputs[0]
54 
55     def __call__(self, x):
56         if isinstance(x, DATAFRAME_TYPE):
57             self._object_type = ObjectType.dataframe
58             return self.new_dataframe([x], shape=x.shape, dtypes=x.dtypes,
59                                       columns_value=x.columns_value, index_value=x.index_value)
60         else:
61             self._object_type = ObjectType.series
62             return self.new_series([x], shape=x.shape, dtype=x.dtype, index_value=x.index_value, name=x.name)
63 
64     @classmethod
65     def tile(cls, op):
66         check_chunks_unknown_shape(op.inputs, TilesError)
67         out = op.outputs[0]
68         new_chunk_size = op.chunk_size
69         if isinstance(out, DATAFRAME_TYPE):
70             itemsize = max(dt.itemsize for dt in out.dtypes)
71         else:
72             itemsize = out.dtype.itemsize
73         steps = plan_rechunks(op.inputs[0], new_chunk_size, itemsize,
74                               threshold=op.threshold,
75                               chunk_size_limit=op.chunk_size_limit)
76         for c in steps:
77             out = compute_rechunk(out.inputs[0], c)
78 
79         return [out]
80 
81 
82 def rechunk(a, chunk_size, threshold=None, chunk_size_limit=None):
83     if isinstance(a, DATAFRAME_TYPE):
84         itemsize = max(dt.itemsize for dt in a.dtypes)
85     else:
86         itemsize = a.dtype.itemsize
87     chunk_size = get_nsplits(a, chunk_size, itemsize)
88     if chunk_size == a.nsplits:
89         return a
90 
91     op = DataFrameRechunk(chunk_size, threshold, chunk_size_limit)
92     return op(a)
93 
94 
95 def _concat_dataframe_index_and_columns(to_concat_chunks):
96     if to_concat_chunks[0].index_value.to_pandas().empty:
97         index_value = to_concat_chunks[0].index_value
98     else:
99         idx_to_index_value = dict((c.index[0], c.index_value) for c in to_concat_chunks if c.index[1] == 0)
100         index_value = merge_index_value(idx_to_index_value)
101 
102     idx_to_columns_value = dict((c.index[1], c.columns_value) for c in to_concat_chunks if c.index[0] == 0)
103     columns_value = merge_index_value(idx_to_columns_value, store_data=True)
104     return index_value, columns_value
105 
106 
107 def _concat_series_index(to_concat_chunks):
108     if to_concat_chunks[0].index_value.to_pandas().empty:
109         index_value = to_concat_chunks[0].index_value
110     else:
111         idx_to_index_value = dict((c.index[0], c.index_value) for c in to_concat_chunks)
112         index_value = merge_index_value(idx_to_index_value)
113     return index_value
114 
115 
116 def compute_rechunk(a, chunk_size):
117     from ..indexing.iloc import DataFrameIlocGetItem, SeriesIlocGetItem
118     from ..merge.concat import DataFrameConcat
119 
120     result_slices = compute_rechunk_slices(a, chunk_size)
121     result_chunks = []
122     idxes = itertools.product(*[range(len(c)) for c in chunk_size])
123     chunk_slices = itertools.product(*result_slices)
124     chunk_shapes = itertools.product(*chunk_size)
125     is_dataframe = isinstance(a, DATAFRAME_TYPE)
126     for idx, chunk_slice, chunk_shape in zip(idxes, chunk_slices, chunk_shapes):
127         to_merge = []
128         merge_idxes = itertools.product(*[range(len(i)) for i in chunk_slice])
129         for merge_idx, index_slices in zip(merge_idxes, itertools.product(*chunk_slice)):
130             chunk_index, chunk_slice = zip(*index_slices)
131             old_chunk = a.cix[chunk_index]
132             merge_chunk_shape = tuple(calc_sliced_size(s, chunk_slice[0]) for s in old_chunk.shape)
133             new_index_value = indexing_index_value(old_chunk.index_value, chunk_slice[0])
134             if is_dataframe:
135                 new_columns_value = indexing_index_value(old_chunk.columns_value, chunk_slice[1], store_data=True)
136                 merge_chunk_op = DataFrameIlocGetItem(chunk_slice, sparse=old_chunk.op.sparse,
137                                                       object_type=ObjectType.dataframe)
138                 merge_chunk = merge_chunk_op.new_chunk([old_chunk], shape=merge_chunk_shape,
139                                                        index=merge_idx, index_value=new_index_value,
140                                                        columns_value=new_columns_value, dtypes=old_chunk.dtypes)
141             else:
142                 merge_chunk_op = SeriesIlocGetItem(chunk_slice, sparse=old_chunk.op.sparse,
143                                                    object_type=ObjectType.series)
144                 merge_chunk = merge_chunk_op.new_chunk([old_chunk], shape=merge_chunk_shape,
145                                                        index=merge_idx, index_value=new_index_value,
146                                                        dtype=old_chunk.dtype)
147             to_merge.append(merge_chunk)
148         if len(to_merge) == 1:
149             chunk_op = to_merge[0].op.copy()
150             if is_dataframe:
151                 out_chunk = chunk_op.new_chunk(to_merge[0].op.inputs, shape=chunk_shape,
152                                                index=idx, index_value=to_merge[0].index_value,
153                                                columns_value=to_merge[0].columns_value,
154                                                dtypes=to_merge[0].dtypes)
155             else:
156                 out_chunk = chunk_op.new_chunk(to_merge[0].op.inputs, shape=chunk_shape,
157                                                index=idx, index_value=to_merge[0].index_value,
158                                                name=to_merge[0].name, dtype=to_merge[0].dtype)
159             result_chunks.append(out_chunk)
160         else:
161             if is_dataframe:
162                 chunk_op = DataFrameConcat(object_type=ObjectType.dataframe)
163                 index_value, columns_value = _concat_dataframe_index_and_columns(to_merge)
164                 out_chunk = chunk_op.new_chunk(to_merge, shape=chunk_shape,
165                                                index=idx, index_value=index_value,
166                                                columns_value=columns_value,
167                                                dtypes=to_merge[0].dtypes)
168             else:
169                 chunk_op = DataFrameConcat(object_type=ObjectType.series)
170                 index_value = _concat_series_index(to_merge)
171                 out_chunk = chunk_op.new_chunk(to_merge, shape=chunk_shape,
172                                                index=idx, index_value=index_value,
173                                                dtype=to_merge[0].dtype)
174             result_chunks.append(out_chunk)
175 
176     if is_dataframe:
177         op = DataFrameRechunk(chunk_size, object_type=ObjectType.dataframe)
178         return op.new_dataframe([a], a.shape, dtypes=a.dtypes, columns_value=a.columns_value,
179                                 index_value=a.index_value, nsplits=chunk_size, chunks=result_chunks)
180     else:
181         op = DataFrameRechunk(chunk_size, object_type=ObjectType.series)
182         return op.new_series([a], a.shape, dtype=a.dtype, index_value=a.index_value,
183                              nsplits=chunk_size, chunks=result_chunks)
184
```
--- 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/base/rechunk.py b/mars/dataframe/base/rechunk.py
--- a/mars/dataframe/base/rechunk.py
+++ b/mars/dataframe/base/rechunk.py
@@ -14,6 +14,8 @@
 
 import itertools
 
+import pandas as pd
+
 from ... import opcodes as OperandDef
 from ...serialize import KeyField, AnyField, Int32Field, Int64Field
 from ...tensor.rechunk.core import get_nsplits, plan_rechunks, compute_rechunk_slices
@@ -92,7 +94,7 @@
     return op(a)
 
 
-def _concat_dataframe_index_and_columns(to_concat_chunks):
+def _concat_dataframe_meta(to_concat_chunks):
     if to_concat_chunks[0].index_value.to_pandas().empty:
         index_value = to_concat_chunks[0].index_value
     else:
@@ -101,7 +103,10 @@
 
     idx_to_columns_value = dict((c.index[1], c.columns_value) for c in to_concat_chunks if c.index[0] == 0)
     columns_value = merge_index_value(idx_to_columns_value, store_data=True)
-    return index_value, columns_value
+
+    idx_to_dtypes = dict((c.index[1], c.dtypes) for c in to_concat_chunks if c.index[0] == 0)
+    dtypes = pd.concat([v[1] for v in list(sorted(idx_to_dtypes.items()))])
+    return index_value, columns_value, dtypes
 
 
 def _concat_series_index(to_concat_chunks):
@@ -137,7 +142,8 @@
                                                       object_type=ObjectType.dataframe)
                 merge_chunk = merge_chunk_op.new_chunk([old_chunk], shape=merge_chunk_shape,
                                                        index=merge_idx, index_value=new_index_value,
-                                                       columns_value=new_columns_value, dtypes=old_chunk.dtypes)
+                                                       columns_value=new_columns_value,
+                                                       dtypes=old_chunk.dtypes.iloc[chunk_slice[1]])
             else:
                 merge_chunk_op = SeriesIlocGetItem(chunk_slice, sparse=old_chunk.op.sparse,
                                                    object_type=ObjectType.series)
@@ -160,11 +166,11 @@
         else:
             if is_dataframe:
                 chunk_op = DataFrameConcat(object_type=ObjectType.dataframe)
-                index_value, columns_value = _concat_dataframe_index_and_columns(to_merge)
+                index_value, columns_value, dtypes = _concat_dataframe_meta(to_merge)
                 out_chunk = chunk_op.new_chunk(to_merge, shape=chunk_shape,
                                                index=idx, index_value=index_value,
                                                columns_value=columns_value,
-                                               dtypes=to_merge[0].dtypes)
+                                               dtypes=dtypes)
             else:
                 chunk_op = DataFrameConcat(object_type=ObjectType.series)
                 index_value = _concat_series_index(to_merge)

{"golden_diff": "diff --git a/mars/dataframe/base/rechunk.py b/mars/dataframe/base/rechunk.py\n--- a/mars/dataframe/base/rechunk.py\n+++ b/mars/dataframe/base/rechunk.py\n@@ -14,6 +14,8 @@\n \n import itertools\n \n+import pandas as pd\n+\n from ... import opcodes as OperandDef\n from ...serialize import KeyField, AnyField, Int32Field, Int64Field\n from ...tensor.rechunk.core import get_nsplits, plan_rechunks, compute_rechunk_slices\n@@ -92,7 +94,7 @@\n     return op(a)\n \n \n-def _concat_dataframe_index_and_columns(to_concat_chunks):\n+def _concat_dataframe_meta(to_concat_chunks):\n     if to_concat_chunks[0].index_value.to_pandas().empty:\n         index_value = to_concat_chunks[0].index_value\n     else:\n@@ -101,7 +103,10 @@\n \n     idx_to_columns_value = dict((c.index[1], c.columns_value) for c in to_concat_chunks if c.index[0] == 0)\n     columns_value = merge_index_value(idx_to_columns_value, store_data=True)\n-    return index_value, columns_value\n+\n+    idx_to_dtypes = dict((c.index[1], c.dtypes) for c in to_concat_chunks if c.index[0] == 0)\n+    dtypes = pd.concat([v[1] for v in list(sorted(idx_to_dtypes.items()))])\n+    return index_value, columns_value, dtypes\n \n \n def _concat_series_index(to_concat_chunks):\n@@ -137,7 +142,8 @@\n                                                       object_type=ObjectType.dataframe)\n                 merge_chunk = merge_chunk_op.new_chunk([old_chunk], shape=merge_chunk_shape,\n                                                        index=merge_idx, index_value=new_index_value,\n-                                                       columns_value=new_columns_value, dtypes=old_chunk.dtypes)\n+                                                       columns_value=new_columns_value,\n+                                                       dtypes=old_chunk.dtypes.iloc[chunk_slice[1]])\n             else:\n                 merge_chunk_op = SeriesIlocGetItem(chunk_slice, sparse=old_chunk.op.sparse,\n                                                    object_type=ObjectType.series)\n@@ -160,11 +166,11 @@\n         else:\n             if is_dataframe:\n                 chunk_op = DataFrameConcat(object_type=ObjectType.dataframe)\n-                index_value, columns_value = _concat_dataframe_index_and_columns(to_merge)\n+                index_value, columns_value, dtypes = _concat_dataframe_meta(to_merge)\n                 out_chunk = chunk_op.new_chunk(to_merge, shape=chunk_shape,\n                                                index=idx, index_value=index_value,\n                                                columns_value=columns_value,\n-                                               dtypes=to_merge[0].dtypes)\n+                                               dtypes=dtypes)\n             else:\n                 chunk_op = DataFrameConcat(object_type=ObjectType.series)\n                 index_value = _concat_series_index(to_merge)\n", "issue": "[BUG] Tile of rechunk() outputs inconsistent dtypes and columns_value\n**Describe the bug**\r\nDataFrame.rechunk() produces chunks with different number of dtypes and columns_value, which may impair descendant chunks.\r\n\r\nCode to reproduce this:\r\n\r\n```python                                                                                                                                                                   \r\nIn [2]: import pandas as pd                                                                                                                                                                                          \r\n\r\nIn [4]: from mars.dataframe.datasource.dataframe import from_pandas as from_pandas_df \r\n\r\nIn [6]: cols = [chr(ord('A') + i) for i in range(10)] \r\n\r\nIn [8]: df_raw = pd.DataFrame(dict((c, [i ** 2 for i in range(20)]) for c in cols))                                                                                                                           \r\n\r\nIn [9]: df = from_pandas_df(df_raw, chunk_size=5)       \r\n\r\nIn [13]: rechunked = df.rechunk((20, 1)).tiles()                                                                                                                                                              \r\n\r\nIn [14]: rechunked.chunks[0].columns_value.to_pandas()                                                                                                                                                        \r\nOut[14]: Index(['A'], dtype='object')\r\n\r\nIn [15]: rechunked.chunks[0].dtypes                                                                                                                                                                           \r\nOut[15]: \r\nA    int64\r\nB    int64\r\nC    int64\r\nD    int64\r\nE    int64\r\ndtype: object\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 itertools\n\nfrom ... import opcodes as OperandDef\nfrom ...serialize import KeyField, AnyField, Int32Field, Int64Field\nfrom ...tensor.rechunk.core import get_nsplits, plan_rechunks, compute_rechunk_slices\nfrom ...tensor.utils import calc_sliced_size\nfrom ...utils import check_chunks_unknown_shape\nfrom ...tiles import TilesError\nfrom ..operands import DataFrameOperand, DataFrameOperandMixin, DATAFRAME_TYPE, ObjectType\nfrom ..utils import indexing_index_value, merge_index_value\n\n\nclass DataFrameRechunk(DataFrameOperand, DataFrameOperandMixin):\n    _op_type_ = OperandDef.RECHUNK\n\n    _input = KeyField('input')\n    _chunk_size = AnyField('chunk_size')\n    _threshold = Int32Field('threshold')\n    _chunk_size_limit = Int64Field('chunk_size_limit')\n\n    def __init__(self, chunk_size=None, threshold=None, chunk_size_limit=None, object_type=None, **kw):\n        super().__init__(_chunk_size=chunk_size, _threshold=threshold,\n                         _chunk_size_limit=chunk_size_limit, _object_type=object_type, **kw)\n\n    @property\n    def chunk_size(self):\n        return self._chunk_size\n\n    @property\n    def threshold(self):\n        return self._threshold\n\n    @property\n    def chunk_size_limit(self):\n        return self._chunk_size_limit\n\n    def _set_inputs(self, inputs):\n        super()._set_inputs(inputs)\n        self._input = self._inputs[0]\n\n    def __call__(self, x):\n        if isinstance(x, DATAFRAME_TYPE):\n            self._object_type = ObjectType.dataframe\n            return self.new_dataframe([x], shape=x.shape, dtypes=x.dtypes,\n                                      columns_value=x.columns_value, index_value=x.index_value)\n        else:\n            self._object_type = ObjectType.series\n            return self.new_series([x], shape=x.shape, dtype=x.dtype, index_value=x.index_value, name=x.name)\n\n    @classmethod\n    def tile(cls, op):\n        check_chunks_unknown_shape(op.inputs, TilesError)\n        out = op.outputs[0]\n        new_chunk_size = op.chunk_size\n        if isinstance(out, DATAFRAME_TYPE):\n            itemsize = max(dt.itemsize for dt in out.dtypes)\n        else:\n            itemsize = out.dtype.itemsize\n        steps = plan_rechunks(op.inputs[0], new_chunk_size, itemsize,\n                              threshold=op.threshold,\n                              chunk_size_limit=op.chunk_size_limit)\n        for c in steps:\n            out = compute_rechunk(out.inputs[0], c)\n\n        return [out]\n\n\ndef rechunk(a, chunk_size, threshold=None, chunk_size_limit=None):\n    if isinstance(a, DATAFRAME_TYPE):\n        itemsize = max(dt.itemsize for dt in a.dtypes)\n    else:\n        itemsize = a.dtype.itemsize\n    chunk_size = get_nsplits(a, chunk_size, itemsize)\n    if chunk_size == a.nsplits:\n        return a\n\n    op = DataFrameRechunk(chunk_size, threshold, chunk_size_limit)\n    return op(a)\n\n\ndef _concat_dataframe_index_and_columns(to_concat_chunks):\n    if to_concat_chunks[0].index_value.to_pandas().empty:\n        index_value = to_concat_chunks[0].index_value\n    else:\n        idx_to_index_value = dict((c.index[0], c.index_value) for c in to_concat_chunks if c.index[1] == 0)\n        index_value = merge_index_value(idx_to_index_value)\n\n    idx_to_columns_value = dict((c.index[1], c.columns_value) for c in to_concat_chunks if c.index[0] == 0)\n    columns_value = merge_index_value(idx_to_columns_value, store_data=True)\n    return index_value, columns_value\n\n\ndef _concat_series_index(to_concat_chunks):\n    if to_concat_chunks[0].index_value.to_pandas().empty:\n        index_value = to_concat_chunks[0].index_value\n    else:\n        idx_to_index_value = dict((c.index[0], c.index_value) for c in to_concat_chunks)\n        index_value = merge_index_value(idx_to_index_value)\n    return index_value\n\n\ndef compute_rechunk(a, chunk_size):\n    from ..indexing.iloc import DataFrameIlocGetItem, SeriesIlocGetItem\n    from ..merge.concat import DataFrameConcat\n\n    result_slices = compute_rechunk_slices(a, chunk_size)\n    result_chunks = []\n    idxes = itertools.product(*[range(len(c)) for c in chunk_size])\n    chunk_slices = itertools.product(*result_slices)\n    chunk_shapes = itertools.product(*chunk_size)\n    is_dataframe = isinstance(a, DATAFRAME_TYPE)\n    for idx, chunk_slice, chunk_shape in zip(idxes, chunk_slices, chunk_shapes):\n        to_merge = []\n        merge_idxes = itertools.product(*[range(len(i)) for i in chunk_slice])\n        for merge_idx, index_slices in zip(merge_idxes, itertools.product(*chunk_slice)):\n            chunk_index, chunk_slice = zip(*index_slices)\n            old_chunk = a.cix[chunk_index]\n            merge_chunk_shape = tuple(calc_sliced_size(s, chunk_slice[0]) for s in old_chunk.shape)\n            new_index_value = indexing_index_value(old_chunk.index_value, chunk_slice[0])\n            if is_dataframe:\n                new_columns_value = indexing_index_value(old_chunk.columns_value, chunk_slice[1], store_data=True)\n                merge_chunk_op = DataFrameIlocGetItem(chunk_slice, sparse=old_chunk.op.sparse,\n                                                      object_type=ObjectType.dataframe)\n                merge_chunk = merge_chunk_op.new_chunk([old_chunk], shape=merge_chunk_shape,\n                                                       index=merge_idx, index_value=new_index_value,\n                                                       columns_value=new_columns_value, dtypes=old_chunk.dtypes)\n            else:\n                merge_chunk_op = SeriesIlocGetItem(chunk_slice, sparse=old_chunk.op.sparse,\n                                                   object_type=ObjectType.series)\n                merge_chunk = merge_chunk_op.new_chunk([old_chunk], shape=merge_chunk_shape,\n                                                       index=merge_idx, index_value=new_index_value,\n                                                       dtype=old_chunk.dtype)\n            to_merge.append(merge_chunk)\n        if len(to_merge) == 1:\n            chunk_op = to_merge[0].op.copy()\n            if is_dataframe:\n                out_chunk = chunk_op.new_chunk(to_merge[0].op.inputs, shape=chunk_shape,\n                                               index=idx, index_value=to_merge[0].index_value,\n                                               columns_value=to_merge[0].columns_value,\n                                               dtypes=to_merge[0].dtypes)\n            else:\n                out_chunk = chunk_op.new_chunk(to_merge[0].op.inputs, shape=chunk_shape,\n                                               index=idx, index_value=to_merge[0].index_value,\n                                               name=to_merge[0].name, dtype=to_merge[0].dtype)\n            result_chunks.append(out_chunk)\n        else:\n            if is_dataframe:\n                chunk_op = DataFrameConcat(object_type=ObjectType.dataframe)\n                index_value, columns_value = _concat_dataframe_index_and_columns(to_merge)\n                out_chunk = chunk_op.new_chunk(to_merge, shape=chunk_shape,\n                                               index=idx, index_value=index_value,\n                                               columns_value=columns_value,\n                                               dtypes=to_merge[0].dtypes)\n            else:\n                chunk_op = DataFrameConcat(object_type=ObjectType.series)\n                index_value = _concat_series_index(to_merge)\n                out_chunk = chunk_op.new_chunk(to_merge, shape=chunk_shape,\n                                               index=idx, index_value=index_value,\n                                               dtype=to_merge[0].dtype)\n            result_chunks.append(out_chunk)\n\n    if is_dataframe:\n        op = DataFrameRechunk(chunk_size, object_type=ObjectType.dataframe)\n        return op.new_dataframe([a], a.shape, dtypes=a.dtypes, columns_value=a.columns_value,\n                                index_value=a.index_value, nsplits=chunk_size, chunks=result_chunks)\n    else:\n        op = DataFrameRechunk(chunk_size, object_type=ObjectType.series)\n        return op.new_series([a], a.shape, dtype=a.dtype, index_value=a.index_value,\n                             nsplits=chunk_size, chunks=result_chunks)\n", "path": "mars/dataframe/base/rechunk.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 itertools\n\nimport pandas as pd\n\nfrom ... import opcodes as OperandDef\nfrom ...serialize import KeyField, AnyField, Int32Field, Int64Field\nfrom ...tensor.rechunk.core import get_nsplits, plan_rechunks, compute_rechunk_slices\nfrom ...tensor.utils import calc_sliced_size\nfrom ...utils import check_chunks_unknown_shape\nfrom ...tiles import TilesError\nfrom ..operands import DataFrameOperand, DataFrameOperandMixin, DATAFRAME_TYPE, ObjectType\nfrom ..utils import indexing_index_value, merge_index_value\n\n\nclass DataFrameRechunk(DataFrameOperand, DataFrameOperandMixin):\n    _op_type_ = OperandDef.RECHUNK\n\n    _input = KeyField('input')\n    _chunk_size = AnyField('chunk_size')\n    _threshold = Int32Field('threshold')\n    _chunk_size_limit = Int64Field('chunk_size_limit')\n\n    def __init__(self, chunk_size=None, threshold=None, chunk_size_limit=None, object_type=None, **kw):\n        super().__init__(_chunk_size=chunk_size, _threshold=threshold,\n                         _chunk_size_limit=chunk_size_limit, _object_type=object_type, **kw)\n\n    @property\n    def chunk_size(self):\n        return self._chunk_size\n\n    @property\n    def threshold(self):\n        return self._threshold\n\n    @property\n    def chunk_size_limit(self):\n        return self._chunk_size_limit\n\n    def _set_inputs(self, inputs):\n        super()._set_inputs(inputs)\n        self._input = self._inputs[0]\n\n    def __call__(self, x):\n        if isinstance(x, DATAFRAME_TYPE):\n            self._object_type = ObjectType.dataframe\n            return self.new_dataframe([x], shape=x.shape, dtypes=x.dtypes,\n                                      columns_value=x.columns_value, index_value=x.index_value)\n        else:\n            self._object_type = ObjectType.series\n            return self.new_series([x], shape=x.shape, dtype=x.dtype, index_value=x.index_value, name=x.name)\n\n    @classmethod\n    def tile(cls, op):\n        check_chunks_unknown_shape(op.inputs, TilesError)\n        out = op.outputs[0]\n        new_chunk_size = op.chunk_size\n        if isinstance(out, DATAFRAME_TYPE):\n            itemsize = max(dt.itemsize for dt in out.dtypes)\n        else:\n            itemsize = out.dtype.itemsize\n        steps = plan_rechunks(op.inputs[0], new_chunk_size, itemsize,\n                              threshold=op.threshold,\n                              chunk_size_limit=op.chunk_size_limit)\n        for c in steps:\n            out = compute_rechunk(out.inputs[0], c)\n\n        return [out]\n\n\ndef rechunk(a, chunk_size, threshold=None, chunk_size_limit=None):\n    if isinstance(a, DATAFRAME_TYPE):\n        itemsize = max(dt.itemsize for dt in a.dtypes)\n    else:\n        itemsize = a.dtype.itemsize\n    chunk_size = get_nsplits(a, chunk_size, itemsize)\n    if chunk_size == a.nsplits:\n        return a\n\n    op = DataFrameRechunk(chunk_size, threshold, chunk_size_limit)\n    return op(a)\n\n\ndef _concat_dataframe_meta(to_concat_chunks):\n    if to_concat_chunks[0].index_value.to_pandas().empty:\n        index_value = to_concat_chunks[0].index_value\n    else:\n        idx_to_index_value = dict((c.index[0], c.index_value) for c in to_concat_chunks if c.index[1] == 0)\n        index_value = merge_index_value(idx_to_index_value)\n\n    idx_to_columns_value = dict((c.index[1], c.columns_value) for c in to_concat_chunks if c.index[0] == 0)\n    columns_value = merge_index_value(idx_to_columns_value, store_data=True)\n\n    idx_to_dtypes = dict((c.index[1], c.dtypes) for c in to_concat_chunks if c.index[0] == 0)\n    dtypes = pd.concat([v[1] for v in list(sorted(idx_to_dtypes.items()))])\n    return index_value, columns_value, dtypes\n\n\ndef _concat_series_index(to_concat_chunks):\n    if to_concat_chunks[0].index_value.to_pandas().empty:\n        index_value = to_concat_chunks[0].index_value\n    else:\n        idx_to_index_value = dict((c.index[0], c.index_value) for c in to_concat_chunks)\n        index_value = merge_index_value(idx_to_index_value)\n    return index_value\n\n\ndef compute_rechunk(a, chunk_size):\n    from ..indexing.iloc import DataFrameIlocGetItem, SeriesIlocGetItem\n    from ..merge.concat import DataFrameConcat\n\n    result_slices = compute_rechunk_slices(a, chunk_size)\n    result_chunks = []\n    idxes = itertools.product(*[range(len(c)) for c in chunk_size])\n    chunk_slices = itertools.product(*result_slices)\n    chunk_shapes = itertools.product(*chunk_size)\n    is_dataframe = isinstance(a, DATAFRAME_TYPE)\n    for idx, chunk_slice, chunk_shape in zip(idxes, chunk_slices, chunk_shapes):\n        to_merge = []\n        merge_idxes = itertools.product(*[range(len(i)) for i in chunk_slice])\n        for merge_idx, index_slices in zip(merge_idxes, itertools.product(*chunk_slice)):\n            chunk_index, chunk_slice = zip(*index_slices)\n            old_chunk = a.cix[chunk_index]\n            merge_chunk_shape = tuple(calc_sliced_size(s, chunk_slice[0]) for s in old_chunk.shape)\n            new_index_value = indexing_index_value(old_chunk.index_value, chunk_slice[0])\n            if is_dataframe:\n                new_columns_value = indexing_index_value(old_chunk.columns_value, chunk_slice[1], store_data=True)\n                merge_chunk_op = DataFrameIlocGetItem(chunk_slice, sparse=old_chunk.op.sparse,\n                                                      object_type=ObjectType.dataframe)\n                merge_chunk = merge_chunk_op.new_chunk([old_chunk], shape=merge_chunk_shape,\n                                                       index=merge_idx, index_value=new_index_value,\n                                                       columns_value=new_columns_value,\n                                                       dtypes=old_chunk.dtypes.iloc[chunk_slice[1]])\n            else:\n                merge_chunk_op = SeriesIlocGetItem(chunk_slice, sparse=old_chunk.op.sparse,\n                                                   object_type=ObjectType.series)\n                merge_chunk = merge_chunk_op.new_chunk([old_chunk], shape=merge_chunk_shape,\n                                                       index=merge_idx, index_value=new_index_value,\n                                                       dtype=old_chunk.dtype)\n            to_merge.append(merge_chunk)\n        if len(to_merge) == 1:\n            chunk_op = to_merge[0].op.copy()\n            if is_dataframe:\n                out_chunk = chunk_op.new_chunk(to_merge[0].op.inputs, shape=chunk_shape,\n                                               index=idx, index_value=to_merge[0].index_value,\n                                               columns_value=to_merge[0].columns_value,\n                                               dtypes=to_merge[0].dtypes)\n            else:\n                out_chunk = chunk_op.new_chunk(to_merge[0].op.inputs, shape=chunk_shape,\n                                               index=idx, index_value=to_merge[0].index_value,\n                                               name=to_merge[0].name, dtype=to_merge[0].dtype)\n            result_chunks.append(out_chunk)\n        else:\n            if is_dataframe:\n                chunk_op = DataFrameConcat(object_type=ObjectType.dataframe)\n                index_value, columns_value, dtypes = _concat_dataframe_meta(to_merge)\n                out_chunk = chunk_op.new_chunk(to_merge, shape=chunk_shape,\n                                               index=idx, index_value=index_value,\n                                               columns_value=columns_value,\n                                               dtypes=dtypes)\n            else:\n                chunk_op = DataFrameConcat(object_type=ObjectType.series)\n                index_value = _concat_series_index(to_merge)\n                out_chunk = chunk_op.new_chunk(to_merge, shape=chunk_shape,\n                                               index=idx, index_value=index_value,\n                                               dtype=to_merge[0].dtype)\n            result_chunks.append(out_chunk)\n\n    if is_dataframe:\n        op = DataFrameRechunk(chunk_size, object_type=ObjectType.dataframe)\n        return op.new_dataframe([a], a.shape, dtypes=a.dtypes, columns_value=a.columns_value,\n                                index_value=a.index_value, nsplits=chunk_size, chunks=result_chunks)\n    else:\n        op = DataFrameRechunk(chunk_size, object_type=ObjectType.series)\n        return op.new_series([a], a.shape, dtype=a.dtype, index_value=a.index_value,\n                             nsplits=chunk_size, chunks=result_chunks)\n", "path": "mars/dataframe/base/rechunk.py"}]}