void buildPhysicalTopology() throws Exception {
        if (CollectionUtils.isNotEmpty(plan.getFragments()) &&
                plan.getTopFragment().getSink() instanceof OlapTableSink) {
            ConnectContext context = queryCoordinator.getConnectContext();
            context.getSessionVariable().setPreferComputeNode(false);
            context.getSessionVariable().setUseComputeNodes(0);
            OlapTableSink dataSink = getMVOlapTableSink();
            // NOTE use a fake transaction id, the real one would be generated when epoch started
            long fakeTransactionId = 1;
            long dbId = getView().getDbId();
            long timeout = context.getSessionVariable().getQueryTimeoutS();
            dataSink.init(context.getExecutionId(), fakeTransactionId, dbId, timeout);
            dataSink.complete();
        }
        queryCoordinator.prepareExec();

        List<ExecutionFragment> execFragments = queryCoordinator.getFragmentsInPreorder();
        TDescriptorTable descTable = queryCoordinator.getDescriptorTable();
        int tabletSinkDop = 1;

        // Group all fragment instances by BE id, and package them into a task
        BiMap<Long, TNetworkAddress> taskId2Addr = HashBiMap.create();
        BiMap<TNetworkAddress, Long> addr2TaskId = taskId2Addr.inverse();
        Map<Long, MVMaintenanceTask> tasksByBe = new HashMap<>();
        long taskIdGen = 0;
        int backendIdGen = 0;
        TFragmentInstanceFactory execPlanFragmentParamsFactory = queryCoordinator.createTFragmentInstanceFactory();
        for (ExecutionFragment execFragment : execFragments) {
            List<TExecPlanFragmentParams> tParams = execPlanFragmentParamsFactory.create(
                    execFragment, execFragment.getInstances(), descTable, tabletSinkDop, tabletSinkDop);
            for (int i = 0; i < execFragment.getInstances().size(); i++) {
                FragmentInstance instance = execFragment.getInstances().get(i);
                // Get brpc address instead of the default address
                TNetworkAddress beRpcAddr = queryCoordinator.getBrpcAddress(instance.getWorkerId());
                Long taskId = addr2TaskId.get(beRpcAddr);
                MVMaintenanceTask task;
                if (taskId == null) {
                    taskId = taskIdGen++;
                    task = MVMaintenanceTask.build(this, taskId, beRpcAddr, new ArrayList<>());
                    tasksByBe.put(taskId, task);
                    addr2TaskId.put(beRpcAddr, taskId);
                } else {
                    task = tasksByBe.get(taskId);
                }

                // TODO(murphy) is this necessary
                int backendId = backendIdGen++;
                instance.setIndexInJob(backendId);
                task.addFragmentInstance(tParams.get(i));
            }
        }
        this.taskId2Addr = taskId2Addr;
        this.taskMap = tasksByBe;
    }
@Test
    public void buildPhysicalTopology() throws Exception {
        String sql = "select count(distinct v5) from t1 join t2";
        Pair<String, ExecPlan> pair = UtFrameUtils.getPlanAndFragment(connectContext, sql);
        assertEquals("AGGREGATE ([GLOBAL] aggregate [{7: count=count(7: count)}] group by [[]] having [null]\n" +
                        "    EXCHANGE GATHER\n" +
                        "        AGGREGATE ([DISTINCT_LOCAL] aggregate [{7: count=count(2: v5)}] group by [[]] having [null]\n" +
                        "            AGGREGATE ([DISTINCT_GLOBAL] aggregate [{}] group by [[2: v5]] having [null]\n" +
                        "                EXCHANGE SHUFFLE[2]\n" +
                        "                    AGGREGATE ([LOCAL] aggregate [{}] group by [[2: v5]] having [null]\n" +
                        "                        CROSS JOIN (join-predicate [null] post-join-predicate [null])\n" +
                        "                            SCAN (columns[2: v5] predicate[null])\n" +
                        "                            EXCHANGE BROADCAST\n" +
                        "                                SCAN (columns[4: v7] predicate[null])",
                pair.first);

        String currentDb = connectContext.getDatabase();
        long dbId = GlobalStateMgr.getCurrentState().getDb(currentDb).getId();
        MaterializedView view = new MaterializedView();
        view.setDbId(dbId);
        view.setId(1024);
        view.setName("view1");
        view.setMaintenancePlan(pair.second);

        MVMaintenanceJob job = new MVMaintenanceJob(view);
        job.buildContext();
        job.buildPhysicalTopology();

        Map<Long, MVMaintenanceTask> taskMap = job.getTasks();
        System.err.println(taskMap);
        assertEquals(1, taskMap.size());
        MVMaintenanceTask task = taskMap.values().stream().findFirst().get();
        System.err.println(task);
        assertEquals(0, task.getTaskId());

        List<TExecPlanFragmentParams> instances = task.getFragmentInstances();
        assertEquals(4, instances.size());

        TExecPlanFragmentParams firstInstance = instances.get(2);
        System.err.println(firstInstance);
        List<TPlanNode> planNodes = firstInstance.getFragment().getPlan().getNodes();
        assertEquals(5, planNodes.size());
        assertEquals(TPlanNodeType.AGGREGATION_NODE, planNodes.get(0).getNode_type());
        assertEquals(TPlanNodeType.PROJECT_NODE, planNodes.get(1).getNode_type());
        assertEquals(TPlanNodeType.NESTLOOP_JOIN_NODE, planNodes.get(2).getNode_type());
        assertEquals(TPlanNodeType.OLAP_SCAN_NODE, planNodes.get(3).getNode_type());
        assertEquals(TPlanNodeType.EXCHANGE_NODE, planNodes.get(4).getNode_type());
    }