#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright 1999-2018 Alibaba Group Holding Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from .... import operands
from ....tiles import NotSupportTile
from ..core import TensorOperandMixin


class TensorFuseChunk(operands.Fuse, TensorOperandMixin):
    def __init__(self, dtype=None, **kw):
        super(TensorFuseChunk, self).__init__(_dtype=dtype, **kw)

    def calc_shape(self, *inputs_shape):
        in_shapes = inputs_shape
        out_shape = None

        # TODO: the logic will be changed when fusion is not only straight line
        for c in self.outputs[0].composed:
            out_shape = c.op.calc_shape(*in_shapes)
            in_shapes = [out_shape]
        return out_shape

    @classmethod
    def tile(cls, op):
        raise NotSupportTile('TensorFuseChunk is a chunk operand which does not support tile')


class TensorFuseChunkMixin(TensorOperandMixin):
    __slots__ = ()

    @classmethod
    def tile(cls, op):
        raise NotSupportTile('TensorFuseChunk is a chunk operand which does not support tile')

    def __call__(self, fuse_chunks):
        head_chunk = fuse_chunks[0]
        tail_chunk = fuse_chunks[-1]
        setattr(self, '_operands', [c.op for c in fuse_chunks])
        return self.new_chunk(head_chunk.inputs, tail_chunk.shape,
                              _composed=fuse_chunks, _key=tail_chunk.key)

mars/tensor/expressions/fuse/core.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# Copyright 1999-2018 Alibaba Group Holding Ltd.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from .... import operands
from ....tiles import NotSupportTile
from ..core import TensorOperandMixin


class TensorFuseChunk(operands.Fuse, TensorOperandMixin):
    def __init__(self, dtype=None, sparse=False, **kw):
        super(TensorFuseChunk, self).__init__(_dtype=dtype, _sparse=sparse, **kw)

    def calc_shape(self, *inputs_shape):
        in_shapes = inputs_shape
        out_shape = None

        # TODO: the logic will be changed when fusion is not only straight line
        for c in self.outputs[0].composed:
            out_shape = c.op.calc_shape(*in_shapes)
            in_shapes = [out_shape]
        return out_shape

    @classmethod
    def tile(cls, op):
        raise NotSupportTile('TensorFuseChunk is a chunk operand which does not support tile')


class TensorFuseChunkMixin(TensorOperandMixin):
    __slots__ = ()

    @classmethod
    def tile(cls, op):
        raise NotSupportTile('TensorFuseChunk is a chunk operand which does not support tile')

    def __call__(self, fuse_chunks):
        head_chunk = fuse_chunks[0]
        tail_chunk = fuse_chunks[-1]
        setattr(self, '_operands', [c.op for c in fuse_chunks])
        return self.new_chunk(head_chunk.inputs, tail_chunk.shape,
                              _composed=fuse_chunks, _key=tail_chunk.key)

mars/tensor/expressions/fuse/core.py
diff --git a/mars/fuse.pyx b/mars/fuse.pyx
index c12f2d6796..203191060e 100644
--- a/mars/fuse.pyx
+++ b/mars/fuse.pyx
@@ -41,7 +41,7 @@ class Fusion(object):
         for c in composes:
             head_node = c[0]
             tail_node = c[-1]
-            fuse_op = TensorFuseChunk(dtype=tail_node.dtype, _key=tail_node.op.key)
+            fuse_op = TensorFuseChunk(dtype=tail_node.dtype, sparse=tail_node.op.sparse, _key=tail_node.op.key)
             fuse_chunk = fuse_op.new_chunk(head_node.inputs, tail_node.shape,
                                            index=tail_node.index, _key=tail_node.key,
                                            _composed=c)
diff --git a/mars/tensor/expressions/fuse/core.py b/mars/tensor/expressions/fuse/core.py
index c7b0fd7e55..1df6882769 100644
--- a/mars/tensor/expressions/fuse/core.py
+++ b/mars/tensor/expressions/fuse/core.py
@@ -20,8 +20,8 @@
 
 
 class TensorFuseChunk(operands.Fuse, TensorOperandMixin):
-    def __init__(self, dtype=None, **kw):
-        super(TensorFuseChunk, self).__init__(_dtype=dtype, **kw)
+    def __init__(self, dtype=None, sparse=False, **kw):
+        super(TensorFuseChunk, self).__init__(_dtype=dtype, _sparse=sparse, **kw)
 
     def calc_shape(self, *inputs_shape):
         in_shapes = inputs_shape
diff --git a/mars/tensor/expressions/tests/test_fuse.py b/mars/tensor/expressions/tests/test_fuse.py
new file mode 100644
index 0000000000..b04c297927
--- /dev/null
+++ b/mars/tensor/expressions/tests/test_fuse.py
@@ -0,0 +1,99 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+# Copyright 1999-2018 Alibaba Group Holding Ltd.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+import scipy.sparse as sps
+
+import mars.tensor as mt
+from mars.tensor.expressions.fuse.core import TensorFuseChunk
+from mars.tensor.expressions.datasource import CSRMatrixDataSource
+from mars import operands
+
+
+class Test(unittest.TestCase):
+    def testBuildFusedGraph(self):
+        t = mt.random.rand(9, 9, chunk_size=3)
+        t1 = (t + 1) * 2
+
+        g = t1.build_graph(tiled=True, compose=True)
+        graph_nodes = list(g)
+        self.assertTrue(all(isinstance(n.op, TensorFuseChunk) for n in graph_nodes))
+        self.assertTrue(all(n.shape == (3, 3) for n in graph_nodes))
+
+        fuse_node = graph_nodes[0]
+        self.assertEqual(fuse_node.shape, (3, 3))
+        self.assertEqual(len(fuse_node.composed), 3)
+        self.assertIsInstance(fuse_node.composed[0].op, operands.Rand)
+        self.assertIsInstance(fuse_node.composed[1].op, operands.AddConstant)
+        self.assertIsInstance(fuse_node.composed[2].op, operands.MulConstant)
+
+        t2 = mt.sum((t / 2) - 1, axis=0)
+
+        g = t2.build_graph(tiled=True, compose=True)
+        graph_nodes = list(g.bfs())
+        self.assertTrue(all(isinstance(n.op, TensorFuseChunk) for n in graph_nodes))
+
+        reduction_node = graph_nodes[-1]
+        self.assertEqual(reduction_node.shape, (3,))
+        self.assertEqual(len(reduction_node.composed), 2)
+        self.assertEqual(reduction_node.inputs, reduction_node.composed[0].inputs)
+        self.assertIsInstance(reduction_node.composed[0].op, operands.Concatenate)
+        self.assertIsInstance(reduction_node.composed[1].op, operands.Sum)
+
+        agg_node = graph_nodes[0]
+        self.assertEqual(agg_node.shape, (1, 3))
+        self.assertEqual(len(agg_node.composed), 4)
+        self.assertIsInstance(agg_node.composed[0].op, operands.Rand)
+        self.assertIsInstance(agg_node.composed[1].op, (operands.TDivConstant, operands.DivConstant))
+        self.assertIsInstance(agg_node.composed[2].op, operands.SubConstant)
+        self.assertIsInstance(agg_node.composed[3].op, operands.Sum)
+
+    def testSparse(self):
+        data = sps.rand(9, 9, density=0.1)
+        t = mt.tensor(data, chunk_size=3)
+
+        t1 = t * 2 / 3
+        g = t1.build_graph(tiled=True, compose=True)
+        graph_nodes = list(g)
+        self.assertTrue(all(isinstance(n.op, TensorFuseChunk) for n in graph_nodes))
+        self.assertTrue(all(n.op.sparse for n in graph_nodes))
+        self.assertTrue(all(n.shape == (3, 3) for n in graph_nodes))
+
+        fuse_node = graph_nodes[0]
+        self.assertEqual(fuse_node.shape, (3, 3))
+        self.assertEqual(len(fuse_node.composed), 3)
+        self.assertIsInstance(fuse_node.composed[0].op, CSRMatrixDataSource)
+        self.assertIsInstance(fuse_node.composed[1].op, operands.MulConstant)
+        self.assertIsInstance(fuse_node.composed[2].op, (operands.TDivConstant, operands.DivConstant))
+        self.assertTrue(all(c.op.sparse for c in fuse_node.composed))
+
+        # add constant will convert sparse matrix to dense matrix
+        t2 = t * 2 + 3
+        g = t2.build_graph(tiled=True, compose=True)
+        graph_nodes = list(g)
+        self.assertTrue(all([isinstance(n.op, TensorFuseChunk) for n in graph_nodes]))
+        self.assertTrue(all([not n.op.sparse for n in graph_nodes]))
+        self.assertTrue(all(n.shape == (3, 3) for n in graph_nodes))
+
+        fuse_node = graph_nodes[0]
+        self.assertEqual(fuse_node.shape, (3, 3))
+        self.assertEqual(len(fuse_node.composed), 3)
+        self.assertIsInstance(fuse_node.composed[0].op, CSRMatrixDataSource)
+        self.assertIsInstance(fuse_node.composed[1].op, operands.MulConstant)
+        self.assertTrue(fuse_node.composed[1].op.sparse)
+        self.assertIsInstance(fuse_node.composed[2].op, operands.AddConstant)
+        self.assertFalse(fuse_node.composed[2].op.sparse)

[BUG] Fuse operand's sparse value is wrong
<!--
Thank you for your contribution!

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

**Describe the bug**
A fuse operand's sparseness should be the same as tail node's, it is not set correctly now.

**To Reproduce**
``` Python
In [1]: import scipy.sparse as sps                                                                                                 

In [2]: import mars.tensor as mt                                                                                                   

In [3]: data = sps.rand(10, 10, density=0.05)                                                                                      

In [4]: a = mt.tensor(data, chunk_size=3)                                                                                          

In [5]: b = (a * 2) * 2                                                                                                            

In [6]: g = b.build_graph(tiled=True, compose=True)                                                                                

In [7]: list(g)[0].op.sparse                                                                                                       
Out[7]: False

In [8]: list(g)[0].op                                                                                                              
Out[8]: <mars.tensor.expressions.fuse.core.TensorFuseChunk at 0xa208b7048>

In [9]: list(g)[0].composed[-1].op.sparse                                                                                         
Out[9]: True
```


