public void modifyTableAddOrDrop(Database db, OlapTable olapTable,
                                     Map<Long, List<Column>> indexSchemaMap,
                                     List<Index> indexes, long jobId, long txnId,
                                     Map<Long, Long> indexToNewSchemaId, boolean isReplay)
            throws DdlException, NotImplementedException {
        Locker locker = new Locker();
        locker.lockTablesWithIntensiveDbLock(db, Lists.newArrayList(olapTable.getId()), LockType.WRITE);
        try {
            LOG.debug("indexSchemaMap:{}, indexes:{}", indexSchemaMap, indexes);
            if (olapTable.getState() == OlapTableState.ROLLUP) {
                throw new DdlException("Table[" + olapTable.getName() + "] is doing ROLLUP job");
            }

            // for now table's state can only be NORMAL
            Preconditions.checkState(olapTable.getState() == OlapTableState.NORMAL, olapTable.getState().name());
            olapTable.setState(OlapTableState.UPDATING_META);
            SchemaChangeJobV2 schemaChangeJob = new SchemaChangeJobV2(jobId, db.getId(), olapTable.getId(),
                    olapTable.getName(), 1000);
            // update base index schema
            Set<String> modifiedColumns = Sets.newHashSet();
            boolean hasMv = !olapTable.getRelatedMaterializedViews().isEmpty();
            for (Map.Entry<Long, List<Column>> entry : indexSchemaMap.entrySet()) {
                Long idx = entry.getKey();
                List<Column> indexSchema = entry.getValue();
                // modify the copied indexMeta and put the update result in the indexIdToMeta
                MaterializedIndexMeta currentIndexMeta = olapTable.getIndexMetaByIndexId(idx).shallowCopy();
                List<Column> originSchema = currentIndexMeta.getSchema();

                if (hasMv) {
                    modifiedColumns.addAll(AlterHelper.collectDroppedOrModifiedColumns(originSchema, indexSchema));
                }

                List<Integer> sortKeyUniqueIds = currentIndexMeta.getSortKeyUniqueIds();
                List<Integer> newSortKeyIdxes = new ArrayList<>();
                if (sortKeyUniqueIds != null) {
                    for (Integer uniqueId : sortKeyUniqueIds) {
                        Optional<Column> col = indexSchema.stream().filter(c -> c.getUniqueId() == uniqueId).findFirst();
                        if (col.isEmpty()) {
                            throw new DdlException("Sork key col with unique id: " + uniqueId + " not exists");
                        }
                        int sortKeyIdx = indexSchema.indexOf(col.get());
                        newSortKeyIdxes.add(sortKeyIdx);
                    }
                }

                currentIndexMeta.setSchema(indexSchema);
                if (!newSortKeyIdxes.isEmpty()) {
                    currentIndexMeta.setSortKeyIdxes(newSortKeyIdxes);
                }

                int currentSchemaVersion = currentIndexMeta.getSchemaVersion();
                int newSchemaVersion = currentSchemaVersion + 1;
                currentIndexMeta.setSchemaVersion(newSchemaVersion);
                // update the indexIdToMeta
                olapTable.getIndexIdToMeta().put(idx, currentIndexMeta);
                // if FE upgrade from old version and replay journal, the indexToNewSchemaId maybe null
                if (indexToNewSchemaId != null) {
                    currentIndexMeta.setSchemaId(indexToNewSchemaId.get(idx));
                }

                schemaChangeJob.addIndexSchema(idx, idx, olapTable.getIndexNameById(idx), newSchemaVersion,
                        currentIndexMeta.getSchemaHash(), currentIndexMeta.getShortKeyColumnCount(),
                        indexSchema);
            }
            olapTable.setIndexes(indexes);
            olapTable.rebuildFullSchema();

            // If modified columns are already done, inactive related mv
            inactiveRelatedMaterializedViews(db, olapTable, modifiedColumns);

            if (!isReplay) {
                TableAddOrDropColumnsInfo info = new TableAddOrDropColumnsInfo(db.getId(), olapTable.getId(),
                        indexSchemaMap, indexes, jobId, txnId, indexToNewSchemaId);
                LOG.debug("logModifyTableAddOrDrop info:{}", info);
                GlobalStateMgr.getCurrentState().getEditLog().logModifyTableAddOrDrop(info);
            }

            schemaChangeJob.setWatershedTxnId(txnId);
            schemaChangeJob.setJobState(AlterJobV2.JobState.FINISHED);
            schemaChangeJob.setFinishedTimeMs(System.currentTimeMillis());
            this.addAlterJobV2(schemaChangeJob);

            olapTable.lastSchemaUpdateTime.set(System.nanoTime());
            LOG.info("finished modify table's add or drop column(field). table: {}, is replay: {}", olapTable.getName(),
                    isReplay);
        } finally {
            olapTable.setState(OlapTableState.NORMAL);
            locker.unLockTablesWithIntensiveDbLock(db, Lists.newArrayList(olapTable.getId()), LockType.WRITE);
        }
    }
@Test
    public void testModifyTableAddOrDropColumns() {
        GlobalStateMgr globalStateMgr = GlobalStateMgr.getCurrentState();
        Database db = globalStateMgr.getDb("test");
        OlapTable tbl = (OlapTable) db.getTable("sc_dup2");
        Map<Long, AlterJobV2> alterJobs = globalStateMgr.getSchemaChangeHandler().getAlterJobsV2();

        // origin columns
        Map<Long, List<Column>> indexSchemaMap = new HashMap<>();
        Map<Long, Long> indexToNewSchemaId = new HashMap<>();
        for (Map.Entry<Long, List<Column>> entry : tbl.getIndexIdToSchema().entrySet()) {
            indexSchemaMap.put(entry.getKey(), new LinkedList<>(entry.getValue()));
            indexToNewSchemaId.put(entry.getKey(), globalStateMgr.getNextId());
        }
        List<Index> newIndexes = tbl.getCopiedIndexes();

        Assertions.assertDoesNotThrow(
                () -> ((SchemaChangeHandler) GlobalStateMgr.getCurrentState().getAlterJobMgr().getSchemaChangeHandler())
                        .modifyTableAddOrDrop(db, tbl, indexSchemaMap, newIndexes, 100, 100,
                                indexToNewSchemaId, false));
        jobSize++;
        Assertions.assertEquals(jobSize, alterJobs.size());

        Assertions.assertDoesNotThrow(
                () -> ((SchemaChangeHandler) GlobalStateMgr.getCurrentState().getAlterJobMgr().getSchemaChangeHandler())
                        .modifyTableAddOrDrop(db, tbl, indexSchemaMap, newIndexes, 101, 101,
                                indexToNewSchemaId, true));
        jobSize++;
        Assertions.assertEquals(jobSize, alterJobs.size());

        OlapTableState beforeState = tbl.getState();
        tbl.setState(OlapTableState.ROLLUP);
        Assertions.assertThrows(DdlException.class,
                () -> ((SchemaChangeHandler) GlobalStateMgr.getCurrentState().getAlterJobMgr().getSchemaChangeHandler())
                        .modifyTableAddOrDrop(db, tbl, indexSchemaMap, newIndexes, 102, 102, indexToNewSchemaId,
                                false));
        tbl.setState(beforeState);
    }