def METHOD_NAME(backend):
    session = new_ray_session(
        session_id="abc", worker_num=2, worker_mem=512 * 1024**2, backend=backend
    )
    mt.random.RandomState(0).rand(100, 5).sum().execute()
    session.execute(mt.random.RandomState(0).rand(100, 5).sum())
    mars.execute(mt.random.RandomState(0).rand(100, 5).sum())
    session = new_ray_session(
        session_id="abcd",
        worker_num=2,
        default=True,
        worker_mem=512 * 1024**2,
        backend=backend,
    )
    session.execute(mt.random.RandomState(0).rand(100, 5).sum())
    mars.execute(mt.random.RandomState(0).rand(100, 5).sum())
    df = md.DataFrame(mt.random.rand(100, 4), columns=list("abcd"))
    # Convert mars dataframe to ray dataset
    ds = md.to_ray_dataset(df)
    print(ds.schema(), ds.count())
    ds.filter(lambda row: row["a"] > 0.5).show(5)
    # Convert ray dataset to mars dataframe
    df2 = md.read_ray_dataset(ds)
    print(df2.head(5).execute())
    # Test ray cluster exists after session got gc.
    del session
    import gc
    gc.collect()
    mars.execute(mt.random.RandomState(0).rand(100, 5).sum())