code: def test_two_merges(setup, gen_data2):
      file_path, file_path2 = gen_data2
      df1 = md.read_parquet(file_path)
      df2 = md.read_parquet(file_path2)
      c = (
          (df1 + 1)
          .merge(
              (df2 + 2), left_on=["c2", "c3"], right_on=["cc1", "cc4"]
          )[["c2", "c4", "cc1", "cc2"]]
          .merge(df2, left_on=["cc1"], right_on=["cc3"])
      )
      graph = c.build_graph()
      optimize(graph)
      r = c.execute().fetch()
  
      raw1 = pd.read_parquet(file_path)
      raw2 = pd.read_parquet(file_path2)
  
      expected = (
          (raw1 + 1)
          .merge(
              (raw2 + 2), left_on=["c2", "c3"], right_on=["cc1", "cc4"]
          )[["c2", "c4", "cc1", "cc2"]]
          .merge(raw2, left_on=["cc1"], right_on=["cc3"])
      )
      pd.testing.assert_frame_equal(r, expected)
  
      parquet_nodes = [
          n for n in graph._nodes if type(n.op) is DataFrameReadParquet
      ]
      assert len(parquet_nodes) == 2
  
      # df1 read parquet push down
      df1_node = [n for n in parquet_nodes if n.op.path == file_path][0]
      assert set(df1_node.op.get_columns()) == {"c2", "c3", "c4"}
  
      # df2 read parquet not push down since it needs all the columns
      df2_node = [n for n in parquet_nodes if n.op.path == file_path2][
          0
      ]
      assert df2_node.op.columns is None
  
      # prove that inserted nodes take effect
      inserted_nodes = [
          n for n in graph._nodes if type(n.op) is DataFrameIndex
      ]
      assert len(inserted_nodes) == 3
  
      index_after_merge_node = [
          n
          for n in inserted_nodes
          if type(graph.predecessors(n)[0].op) is DataFrameMerge
      ][0]
      assert set(index_after_merge_node.op.col_names) == {
          "c2",
          "c4",
          "cc1",
          "cc2",
      }
  
imports: import os
  import tempfile
  import pandas
  import pytest
  from ......None import dataframe
  from ......None import tensor
  from ......dataframe.arithmetic import DataFrameMul
  from ......dataframe.base.eval import DataFrameEval
  from ......dataframe.base.isin import DataFrameIsin
  from ......dataframe.core import DataFrameData, SeriesData, DataFrameGroupByData
  from ......dataframe.datasource.dataframe import DataFrameDataSource
  from ......dataframe.datasource.read_csv import DataFrameReadCSV
  from ......dataframe.datasource.read_parquet import DataFrameReadParquet
  from ......dataframe.groupby.aggregation import DataFrameGroupByAgg
  from ......dataframe.groupby.core import DataFrameGroupByOperand
  from ......dataframe.indexing.getitem import DataFrameIndex
  from ......dataframe.indexing.setitem import DataFrameSetitem
  from ......dataframe.merge import DataFrameMerge
  from ......optimization.logical.tileable import optimize
  from ......tensor.core import TensorData
  from ......tensor.datasource import ArrayDataSource
label_desc: Use image node_id 11 for calling a global function with example usage: test_two_merges(setup, gen_data2) without return types