[core] Optimize mars graph building performance

[chaokunyang]

Is your feature request related to a problem? Please describe.

num_rows = 100_0000_0000
df1 = md.DataFrame(
    mt.random.rand(num_rows, 4, chunk_size=100_0000),
    columns=list('abcd'))
df2 = md.DataFrame(
    mt.random.rand(num_rows, 4, chunk_size=100_0000),
    columns=list('abcd'))
df1.merge(df2, left_on='a', right_on='a').execute()

20000subtasks and 50 16C32Gworker

• without coloring, graph build time:
  • tile: 128s
  • assign: 40min
    
    

Describe the solution you'd like
A clear and concise description of what you want to happen.

Describe alternatives you've considered
A clear and concise description of any alternative solutions or features you've considered.

Additional context
Add any other context or screenshots about the feature request here.

[chaokunyang]

The process can be reproduced by:

from ..... import dataframe as md
from ..... import tensor as mt
from .....core.graph import TileableGraph, TileableGraphBuilder, ChunkGraphBuilder
from ..analyzer import GraphAnalyzer
from ..assigner import GraphAssigner


def test_assigner():
    num_rows = 100_0000_0000
    df1 = md.DataFrame(
        mt.random.rand(num_rows, 4, chunk_size=100_0000),
        columns=list('abcd'))
    df2 = md.DataFrame(
        mt.random.rand(num_rows, 4, chunk_size=100_0000),
        columns=list('abcd'))
    merged_df = df1.merge(df2, left_on='a', right_on='a')
    graph = TileableGraph([merged_df.data])
    next(TileableGraphBuilder(graph).build())
    chunk_graph = next(ChunkGraphBuilder(graph, fuse_enabled=False).build())
    start_ops = list(GraphAnalyzer._iter_start_ops(chunk_graph))
    band_slots = {
        (f"worker-{i}", "numa-0"): 16 for i in range(50)
    }
    current_assign = {}
    assigner = GraphAssigner(chunk_graph, start_ops, band_slots)
    assigned_result = assigner.assign(current_assign)
    print(len(assigned_result))