code: def test_merge():
      df1 = pd.DataFrame(
          np.arange(20).reshape((4, 5)) + 1,
          columns=["a", "b", "c", "d", "e"],
      )
      df2 = pd.DataFrame(
          np.arange(20).reshape((5, 4)) + 1,
          columns=["a", "b", "x", "y"],
      )
  
      mdf1 = from_pandas(df1, chunk_size=2)
      mdf2 = from_pandas(df2, chunk_size=3)
  
      parameters = [
          {},
          {"how": "left", "right_on": "x", "left_index": True},
          {"how": "right", "left_on": "a", "right_index": True},
          {"how": "left", "left_on": "a", "right_on": "x"},
          {"how": "right", "left_on": "a", "right_index": True},
          {"how": "right", "on": "a"},
          {"how": "inner", "on": ["a", "b"]},
      ]
  
      for kw in parameters:
          df = mdf1.merge(mdf2, **kw)
          df = tile(df)
  
          assert df.chunk_shape == (2, 1)
          for chunk in df.chunks:
              assert isinstance(chunk.op, DataFrameMerge)
              assert chunk.op.how == kw.get("how", "inner")
              left, right = chunk.op.inputs
              assert isinstance(left.op, DataFrameMergeAlign)
              assert left.op.stage == OperandStage.reduce
              assert isinstance(right.op, DataFrameMergeAlign)
              assert right.op.stage == OperandStage.reduce
              assert len(left.inputs[0].inputs) == 4
              assert len(right.inputs[0].inputs) == 4
              for lchunk in left.inputs[0].inputs[:2]:
                  assert isinstance(lchunk.op, DataFrameMergeAlign)
                  assert lchunk.op.stage == OperandStage.map
                  assert lchunk.op.index_shuffle_size == 2
                  if kw.get("on", None) or kw.get("left_on", None):
                      # defaults to common columns
                      assert lchunk.op.shuffle_on == kw.get(
                          "on", None
                      ) or kw.get("left_on", None)
              for rchunk in right.inputs[0].inputs[2:]:
                  assert isinstance(rchunk.op, DataFrameMergeAlign)
                  assert rchunk.op.stage == OperandStage.map
                  assert rchunk.op.index_shuffle_size == 2
                  if kw.get("on", None) or kw.get("right_on", None):
                      # defaults to common columns
                      assert rchunk.op.shuffle_on == kw.get(
                          "on", None
                      ) or kw.get("right_on", None)
              pd.testing.assert_index_equal(
                  chunk.columns_value.to_pandas(),
                  df.columns_value.to_pandas(),
              )
  
file_path: mars/mars/dataframe/merge/tests/test_merge.py
imports: import numpy
  import pandas
  import pytest
  from ....core import tile
  from ....core.operand import OperandStage
  from ...core import IndexValue
  from ...datasource.dataframe import from_pandas
  from ..None import DataFrameMergeAlign, DataFrameMerge, concat
label_desc: Use image node_id 1 for calling a global function with example usage: test_merge() without return types