code: def test_merge_with_bloom_filter(setup):
      ns = np.random.RandomState(0)
      raw_df1 = pd.DataFrame(
          {
              "col1": ns.random(100),
              "col2": ns.randint(0, 10, size=(100,)),
              "col3": ns.randint(0, 10, size=(100,)),
          }
      )
      raw_df2 = pd.DataFrame(
          {
              "col1": ns.random(100),
              "col2": ns.randint(0, 10, size=(100,)),
              "col3": ns.randint(0, 10, size=(100,)),
          }
      )
  
      df1 = from_pandas(raw_df1, chunk_size=10)
      df2 = from_pandas(raw_df2, chunk_size=15)
  
      expected = raw_df1.merge(raw_df2, on="col2")
  
      result = (
          df1.merge(
              df2,
              on="col2",
              bloom_filter=True,
              bloom_filter_options={
                  "max_elements": 100,
                  "error_rate": 0.01,
              },
              auto_merge="none",
          )
          .execute()
          .fetch()
      )
      pd.testing.assert_frame_equal(
          expected.sort_values(by=["col1_x", "col2"]).reset_index(
              drop=True
          ),
          result.sort_values(by=["col1_x", "col2"]).reset_index(
              drop=True
          ),
      )
  
      result = (
          df2.merge(
              df1,
              on=["col2", "col3"],
              bloom_filter=True,
              auto_merge="none",
          )
          .execute()
          .fetch()
      )
      expected = raw_df2.merge(raw_df1, on=["col2", "col3"])
      pd.testing.assert_frame_equal(
          expected.sort_values(by=["col1_x", "col2"]).reset_index(
              drop=True
          ),
          result.sort_values(by=["col1_x", "col2"]).reset_index(
              drop=True
          ),
      )
  
      # on index
      result = (
          df2.merge(df1, bloom_filter=True, auto_merge="none")
          .execute()
          .fetch()
      )
      expected = raw_df2.merge(raw_df1)
      pd.testing.assert_frame_equal(
          expected.sort_index().reset_index(drop=True),
          result.sort_index().reset_index(drop=True),
      )
  
      # on float column
      result = (
          df2.merge(
              df1, on="col1", bloom_filter=True, auto_merge="none"
          )
          .execute()
          .fetch()
      )
      expected = raw_df2.merge(raw_df1, on="col1")
      pd.testing.assert_frame_equal(
          expected.sort_values(by=["col1", "col2_x"]).reset_index(
              drop=True
          ),
          result.sort_values(by=["col1", "col2_x"]).reset_index(
              drop=True
          ),
      )
  
      # on float columns
      result = (
          df2.merge(
              df1,
              on=["col1", "col2"],
              bloom_filter=True,
              auto_merge="none",
          )
          .execute()
          .fetch()
      )
      expected = raw_df2.merge(raw_df1, on=["col1", "col2"])
      pd.testing.assert_frame_equal(
          expected.sort_values(by=["col1", "col2"]).reset_index(
              drop=True
          ),
          result.sort_values(by=["col1", "col2"]).reset_index(
              drop=True
          ),
      )
  
      # multi index
      raw_df3 = raw_df1.copy()
      raw_df3.index = pd.MultiIndex.from_tuples(
          [(i, i + 1) for i in range(100)], names=["i1", "i2"]
      )
      df3 = from_pandas(raw_df3, chunk_size=8)
      result = (
          df3.merge(
              df1,
              left_on="i1",
              right_on="col2",
              bloom_filter=True,
              auto_merge="none",
          )
          .execute()
          .fetch()
      )
      expected = raw_df3.merge(raw_df1, left_on="i1", right_on="col2")
      pd.testing.assert_frame_equal(
          expected.sort_index()
          .sort_values(by=["col1_x"])
          .reset_index(drop=True),
          result.sort_index()
          .sort_values(by=["col1_x"])
          .reset_index(drop=True),
      )
  
      df4 = from_pandas(raw_df3, chunk_size=20)
      result = (
          df4.merge(
              df1,
              left_on="i1",
              right_on="col2",
              bloom_filter=True,
              auto_merge="none",
          )
          .execute()
          .fetch()
      )
      expected = raw_df3.merge(raw_df1, left_on="i1", right_on="col2")
      pd.testing.assert_frame_equal(
          expected.sort_index()
          .sort_values(by=["col1_x"])
          .reset_index(drop=True),
          result.sort_index()
          .sort_values(by=["col1_x"])
          .reset_index(drop=True),
      )
  
file_path: mars/mars/dataframe/merge/tests/test_merge_execution.py
imports: import numpy
  import pandas
  import pytest
  from ....core.graph.builder.utils import build_graph
  from ...datasource.dataframe import from_pandas
  from ...datasource.series import from_pandas
  from ...utils import sort_dataframe_inplace
  from ..None import concat, DataFrameConcat, DataFrameMergeAlign
label_desc: Use image node_id 6 for calling a global function with example usage: test_merge_with_bloom_filter(setup) without return types