private static void handleUpdateTableSchema(long backendId, Map<Long, TTablet> backendTablets) {
        Table<Long, Long, List<Long>> tableToIndexTabletMap = HashBasedTable.create();
        Map<Long, Long> tableToDb = Maps.newHashMap();

        TabletInvertedIndex invertedIndex = GlobalStateMgr.getCurrentState().getTabletInvertedIndex();
        // split tablets by db, table and index
        for (TTablet backendTablet : backendTablets.values()) {
            for (TTabletInfo tabletInfo : backendTablet.tablet_infos) {
                if (!tabletInfo.isSetTablet_schema_version()) {
                    continue;
                }
                long tabletId = tabletInfo.getTablet_id();
                TabletMeta tabletMeta = invertedIndex.getTabletMeta(tabletId);
                if (tabletMeta == null) {
                    continue;
                }

                long dbId = tabletMeta.getDbId();
                long tableId = tabletMeta.getTableId();
                long indexId = tabletMeta.getIndexId();

                Database db = GlobalStateMgr.getCurrentState().getDb(dbId);
                if (db == null) {
                    continue;
                }

                OlapTable olapTable = (OlapTable) db.getTable(tableId);
                if (olapTable == null) {
                    continue;
                }

                Locker locker = new Locker();
                locker.lockTablesWithIntensiveDbLock(db, Lists.newArrayList(olapTable.getId()), LockType.READ);
                try {
                    if (olapTable.getMaxColUniqueId() <= Column.COLUMN_UNIQUE_ID_INIT_VALUE) {
                        continue;
                    }
                    MaterializedIndexMeta indexMeta = olapTable.getIndexMetaByIndexId(indexId);
                    if (indexMeta == null) {
                        continue;
                    }
                    int schemaVersion = tabletInfo.tablet_schema_version;
                    int latestSchemaVersion = indexMeta.getSchemaVersion();
                    if (schemaVersion < latestSchemaVersion) {
                        List<Long> tabletsList = tableToIndexTabletMap.get(tableId, indexId);
                        if (tabletsList != null) {
                            tabletsList.add(Long.valueOf(tabletId));
                        } else {
                            tabletsList = Lists.newArrayList();
                            tabletsList.add(Long.valueOf(tabletId));
                            tableToIndexTabletMap.put(tableId, indexId, tabletsList);
                        }
                        tableToDb.put(tableId, dbId);
                    }
                } finally {
                    locker.unLockTablesWithIntensiveDbLock(db, Lists.newArrayList(olapTable.getId()), LockType.READ);
                }
            }
        }

        // create AgentBatch Task
        AgentBatchTask updateSchemaBatchTask = new AgentBatchTask();
        for (Table.Cell<Long, Long, List<Long>> cell : tableToIndexTabletMap.cellSet()) {
            Long tableId = cell.getRowKey();
            Long indexId = cell.getColumnKey();
            List<Long> tablets = cell.getValue();
            Long dbId = tableToDb.get(tableId);

            Database db = GlobalStateMgr.getCurrentState().getDb(dbId);
            if (db == null) {
                continue;
            }

            OlapTable olapTable = (OlapTable) db.getTable(tableId);
            if (olapTable == null) {
                continue;
            }
            Locker locker = new Locker();
            locker.lockTablesWithIntensiveDbLock(db, Lists.newArrayList(olapTable.getId()), LockType.READ);
            try {
                MaterializedIndexMeta indexMeta = olapTable.getIndexMetaByIndexId(indexId);
                if (indexMeta == null) {
                    continue;
                }

                // already has one update scheam task, ignore to prevent send too many task
                if (indexMeta.hasUpdateSchemaTask(backendId)) {
                    continue;
                }

                List<TColumn> columnsDesc = Lists.newArrayList();
                List<Integer> columnSortKeyUids = Lists.newArrayList();

                for (Column column : indexMeta.getSchema()) {
                    TColumn tColumn = column.toThrift();
                    tColumn.setColumn_name(column.getColumnId().getId());
                    column.setIndexFlag(tColumn, olapTable.getIndexes(), olapTable.getBfColumnIds());
                    columnsDesc.add(tColumn);
                }
                if (indexMeta.getSortKeyUniqueIds() != null) {
                    columnSortKeyUids.addAll(indexMeta.getSortKeyUniqueIds());
                }
                TOlapTableColumnParam columnParam = new TOlapTableColumnParam(columnsDesc, columnSortKeyUids,
                        indexMeta.getShortKeyColumnCount());

                UpdateSchemaTask task = new UpdateSchemaTask(backendId, db.getId(), olapTable.getId(),
                        indexId, tablets, indexMeta.getSchemaId(), indexMeta.getSchemaVersion(),
                        columnParam);
                updateSchemaBatchTask.addTask(task);
                indexMeta.addUpdateSchemaBackend(backendId);
            } finally {
                locker.unLockTablesWithIntensiveDbLock(db, Lists.newArrayList(olapTable.getId()), LockType.READ);
            }
        }
        // send agent batch task
        if (updateSchemaBatchTask.getTaskNum() > 0) {
            for (AgentTask task : updateSchemaBatchTask.getAllTasks()) {
                AgentTaskQueue.addTask(task);
            }
            AgentTaskExecutor.submit(updateSchemaBatchTask);
        }

    }
@Test
    public void testHandleUpdateTableSchema() throws Exception {
        Database db = GlobalStateMgr.getCurrentState().getDb("test");
        long dbId = db.getId();
        OlapTable olapTable = (OlapTable) db.getTable("update_schema");

        String stmt = "alter table update_schema add column add_v int default '1'";
        StarRocksAssert starRocksAssert = new StarRocksAssert(connectContext);
        AlterTableStmt alterTableStmt =
                (AlterTableStmt) UtFrameUtils.parseStmtWithNewParser(stmt, starRocksAssert.getCtx());
        SchemaChangeHandler schemaChangeHandler = GlobalStateMgr.getCurrentState().getSchemaChangeHandler();
        schemaChangeHandler.process(alterTableStmt.getAlterClauseList(), db, olapTable);
        Assert.assertEquals(OlapTableState.NORMAL, olapTable.getState());

        long backendId = 10001L;
        List<Long> tabletIds =
                GlobalStateMgr.getCurrentState().getTabletInvertedIndex().getTabletIdsByBackendId(10001);
        Assert.assertFalse(tabletIds.isEmpty());

        Map<Long, TTablet> backendTablets = new HashMap<Long, TTablet>();
        List<TTabletInfo> tabletInfos = Lists.newArrayList();
        TTablet tablet = new TTablet(tabletInfos);
        for (Long tabletId : tabletIds) {
            TTabletInfo tabletInfo = new TTabletInfo();
            tabletInfo.setTablet_id(tabletId);
            tabletInfo.setSchema_hash(60000);
            tabletInfo.setTablet_schema_version(0);
            tablet.tablet_infos.add(tabletInfo);
        }
        backendTablets.put(backendId, tablet);

        ReportHandler handler = new ReportHandler();
        handler.testHandleUpdateTableSchema(backendId, backendTablets);
    }