/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.kylin.rest.service;

import static java.util.stream.Collectors.groupingBy;
import static org.apache.kylin.common.exception.ServerErrorCode.COMPUTED_COLUMN_CASCADE_ERROR;
import static org.apache.kylin.common.exception.ServerErrorCode.COMPUTED_COLUMN_DEPENDS_ANTI_FLATTEN_LOOKUP;
import static org.apache.kylin.common.exception.ServerErrorCode.DUPLICATE_COMPUTED_COLUMN_NAME;
import static org.apache.kylin.common.exception.ServerErrorCode.DUPLICATE_DIMENSION_NAME;
import static org.apache.kylin.common.exception.ServerErrorCode.DUPLICATE_JOIN_CONDITION;
import static org.apache.kylin.common.exception.ServerErrorCode.DUPLICATE_MEASURE_EXPRESSION;
import static org.apache.kylin.common.exception.ServerErrorCode.DUPLICATE_MEASURE_NAME;
import static org.apache.kylin.common.exception.ServerErrorCode.FAILED_CREATE_MODEL;
import static org.apache.kylin.common.exception.ServerErrorCode.FAILED_DETECT_DATA_RANGE;
import static org.apache.kylin.common.exception.ServerErrorCode.FAILED_EXECUTE_MODEL_SQL;
import static org.apache.kylin.common.exception.ServerErrorCode.FAILED_UPDATE_MODEL;
import static org.apache.kylin.common.exception.ServerErrorCode.FILTER_CONDITION_DEPENDS_ANTI_FLATTEN_LOOKUP;
import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_COMPUTED_COLUMN_EXPRESSION;
import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_FILTER_CONDITION;
import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_MODEL_TYPE;
import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_MULTI_PARTITION_MAPPING_REQUEST;
import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_NAME;
import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_PARAMETER;
import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_PARTITION_COLUMN;
import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_PARTITION_VALUES;
import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_RANGE;
import static org.apache.kylin.common.exception.ServerErrorCode.INVALID_SEGMENT_PARAMETER;
import static org.apache.kylin.common.exception.ServerErrorCode.MODEL_BROKEN;
import static org.apache.kylin.common.exception.ServerErrorCode.MODEL_ONLINE_ABANDON;
import static org.apache.kylin.common.exception.ServerErrorCode.MODEL_STORAGE_UPDATE_FAILED;
import static org.apache.kylin.common.exception.ServerErrorCode.PERMISSION_DENIED;
import static org.apache.kylin.common.exception.ServerErrorCode.STREAMING_INDEX_UPDATE_DISABLE;
import static org.apache.kylin.common.exception.ServerErrorCode.TABLE_NOT_EXIST;
import static org.apache.kylin.common.exception.ServerErrorCode.TIMESTAMP_COLUMN_NOT_EXIST;
import static org.apache.kylin.common.exception.ServerErrorCode.VIEW_PARTITION_DATE_FORMAT_DETECTION_FORBIDDEN;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.COMPUTED_COLUMN_CONFLICT;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.COMPUTED_COLUMN_NAME_OR_EXPR_EMPTY;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.DATETIME_FORMAT_EMPTY;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.DATETIME_FORMAT_PARSE_ERROR;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.MODEL_ID_NOT_EXIST;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.MODEL_NAME_DUPLICATE;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.MODEL_NAME_NOT_EXIST;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.MODEL_NAME_TOO_LONG;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.PROJECT_NOT_EXIST;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.SEGMENT_LOCKED;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.SEGMENT_MERGE_CONTAINS_GAPS;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.SEGMENT_NOT_EXIST_ID;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.SEGMENT_NOT_EXIST_NAME;
import static org.apache.kylin.common.exception.code.ErrorCodeServer.SEGMENT_STATUS;
import static org.apache.kylin.metadata.model.FunctionDesc.PARAMETER_TYPE_COLUMN;

import java.io.IOException;
import java.math.BigDecimal;
import java.sql.SQLException;
import java.text.MessageFormat;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.calcite.sql.SqlIdentifier;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.dialect.CalciteSqlDialect;
import org.apache.calcite.sql.util.SqlBasicVisitor;
import org.apache.calcite.sql.util.SqlVisitor;
import org.apache.calcite.util.Util;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.QueryContext;
import org.apache.kylin.common.constant.Constant;
import org.apache.kylin.common.event.ModelAddEvent;
import org.apache.kylin.common.event.ModelDropEvent;
import org.apache.kylin.common.exception.JobErrorCode;
import org.apache.kylin.common.exception.KylinException;
import org.apache.kylin.common.exception.KylinTimeoutException;
import org.apache.kylin.common.exception.ServerErrorCode;
import org.apache.kylin.common.exception.code.ErrorCodeServer;
import org.apache.kylin.common.msg.Message;
import org.apache.kylin.common.msg.MsgPicker;
import org.apache.kylin.common.persistence.RootPersistentEntity;
import org.apache.kylin.common.persistence.transaction.TransactionException;
import org.apache.kylin.common.persistence.transaction.UnitOfWork;
import org.apache.kylin.common.persistence.transaction.UnitOfWorkContext;
import org.apache.kylin.common.persistence.transaction.UnitOfWorkParams;
import org.apache.kylin.common.scheduler.EventBusFactory;
import org.apache.kylin.common.util.DateFormat;
import org.apache.kylin.common.util.JsonUtil;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.common.util.RandomUtil;
import org.apache.kylin.common.util.SqlIdentifierFormatterVisitor;
import org.apache.kylin.common.util.StringHelper;
import org.apache.kylin.common.util.ThreadUtil;
import org.apache.kylin.engine.spark.job.NSparkCubingJob;
import org.apache.kylin.engine.spark.utils.ComputedColumnEvalUtil;
import org.apache.kylin.fileseg.FileSegments;
import org.apache.kylin.fileseg.FileSegments.ModelFileSegments;
import org.apache.kylin.guava30.shaded.common.annotations.VisibleForTesting;
import org.apache.kylin.guava30.shaded.common.base.Preconditions;
import org.apache.kylin.guava30.shaded.common.base.Strings;
import org.apache.kylin.guava30.shaded.common.collect.ImmutableList;
import org.apache.kylin.guava30.shaded.common.collect.Lists;
import org.apache.kylin.guava30.shaded.common.collect.Maps;
import org.apache.kylin.guava30.shaded.common.collect.Sets;
import org.apache.kylin.job.common.IndexBuildJobUtil;
import org.apache.kylin.job.common.SegmentUtil;
import org.apache.kylin.job.execution.AbstractExecutable;
import org.apache.kylin.job.execution.ExecutableHandler.HandlerType;
import org.apache.kylin.job.execution.ExecutableManager;
import org.apache.kylin.job.execution.ExecutableState;
import org.apache.kylin.job.execution.JobTypeEnum;
import org.apache.kylin.job.execution.MergerInfo;
import org.apache.kylin.job.manager.JobManager;
import org.apache.kylin.job.manager.SegmentAutoMergeUtil;
import org.apache.kylin.job.model.JobParam;
import org.apache.kylin.metadata.acl.AclTCRManager;
import org.apache.kylin.metadata.acl.NDataModelAclParams;
import org.apache.kylin.metadata.cube.cuboid.NAggregationGroup;
import org.apache.kylin.metadata.cube.model.IndexEntity;
import org.apache.kylin.metadata.cube.model.IndexPlan;
import org.apache.kylin.metadata.cube.model.LayoutEntity;
import org.apache.kylin.metadata.cube.model.NBatchConstants;
import org.apache.kylin.metadata.cube.model.NDataLayout;
import org.apache.kylin.metadata.cube.model.NDataLayoutDetails;
import org.apache.kylin.metadata.cube.model.NDataLayoutDetailsManager;
import org.apache.kylin.metadata.cube.model.NDataSegment;
import org.apache.kylin.metadata.cube.model.NDataflow;
import org.apache.kylin.metadata.cube.model.NDataflowManager;
import org.apache.kylin.metadata.cube.model.NDataflowUpdate;
import org.apache.kylin.metadata.cube.model.NIndexPlanManager;
import org.apache.kylin.metadata.cube.model.RuleBasedIndex;
import org.apache.kylin.metadata.model.AntiFlatChecker;
import org.apache.kylin.metadata.model.AutoMergeTimeEnum;
import org.apache.kylin.metadata.model.ColumnDesc;
import org.apache.kylin.metadata.model.ComputedColumnDesc;
import org.apache.kylin.metadata.model.DataCheckDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.FusionModel;
import org.apache.kylin.metadata.model.FusionModelManager;
import org.apache.kylin.metadata.model.ISourceAware;
import org.apache.kylin.metadata.model.JoinDesc;
import org.apache.kylin.metadata.model.JoinTableDesc;
import org.apache.kylin.metadata.model.ManagementType;
import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.MultiPartitionDesc;
import org.apache.kylin.metadata.model.NDataModel;
import org.apache.kylin.metadata.model.NDataModelManager;
import org.apache.kylin.metadata.model.NTableMetadataManager;
import org.apache.kylin.metadata.model.ParameterDesc;
import org.apache.kylin.metadata.model.PartitionDesc;
import org.apache.kylin.metadata.model.RetentionRange;
import org.apache.kylin.metadata.model.SegmentRange;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
import org.apache.kylin.metadata.model.SegmentStatusEnumToDisplay;
import org.apache.kylin.metadata.model.Segments;
import org.apache.kylin.metadata.model.TableDesc;
import org.apache.kylin.metadata.model.TableRef;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.model.UpdateImpact;
import org.apache.kylin.metadata.model.VolatileRange;
import org.apache.kylin.metadata.model.schema.AffectedModelContext;
import org.apache.kylin.metadata.model.tool.CalciteParser;
import org.apache.kylin.metadata.model.util.ComputedColumnUtil;
import org.apache.kylin.metadata.model.util.MultiPartitionUtil;
import org.apache.kylin.metadata.model.util.scd2.SCD2CondChecker;
import org.apache.kylin.metadata.project.EnhancedUnitOfWork;
import org.apache.kylin.metadata.project.NProjectLoader;
import org.apache.kylin.metadata.project.NProjectManager;
import org.apache.kylin.metadata.project.ProjectInstance;
import org.apache.kylin.metadata.realization.RealizationStatusEnum;
import org.apache.kylin.metadata.streaming.KafkaConfig;
import org.apache.kylin.query.util.PushDownUtil;
import org.apache.kylin.query.util.QueryParams;
import org.apache.kylin.query.util.QueryUtil;
import org.apache.kylin.rest.aspect.Transaction;
import org.apache.kylin.rest.constant.ModelStatusToDisplayEnum;
import org.apache.kylin.rest.feign.MetadataContract;
import org.apache.kylin.rest.request.AddSegmentRequest;
import org.apache.kylin.rest.request.MergeSegmentRequest;
import org.apache.kylin.rest.request.ModelConfigRequest;
import org.apache.kylin.rest.request.ModelParatitionDescRequest;
import org.apache.kylin.rest.request.ModelRequest;
import org.apache.kylin.rest.request.MultiPartitionMappingRequest;
import org.apache.kylin.rest.request.OptimizeLayoutDataRequest;
import org.apache.kylin.rest.request.OwnerChangeRequest;
import org.apache.kylin.rest.request.SegmentTimeRequest;
import org.apache.kylin.rest.response.AffectedModelsResponse;
import org.apache.kylin.rest.response.AggGroupResponse;
import org.apache.kylin.rest.response.BuildBaseIndexResponse;
import org.apache.kylin.rest.response.CheckSegmentResponse;
import org.apache.kylin.rest.response.ComputedColumnCheckResponse;
import org.apache.kylin.rest.response.ComputedColumnConflictResponse;
import org.apache.kylin.rest.response.ComputedColumnUsageResponse;
import org.apache.kylin.rest.response.DataResult;
import org.apache.kylin.rest.response.ExistedDataRangeResponse;
import org.apache.kylin.rest.response.FusionModelResponse;
import org.apache.kylin.rest.response.IndicesResponse;
import org.apache.kylin.rest.response.InvalidIndexesResponse;
import org.apache.kylin.rest.response.JobInfoResponse;
import org.apache.kylin.rest.response.LayoutRecDetailResponse;
import org.apache.kylin.rest.response.ModelConfigResponse;
import org.apache.kylin.rest.response.ModelSaveCheckResponse;
import org.apache.kylin.rest.response.MultiPartitionValueResponse;
import org.apache.kylin.rest.response.NCubeDescResponse;
import org.apache.kylin.rest.response.NDataModelLiteResponse;
import org.apache.kylin.rest.response.NDataModelOldParams;
import org.apache.kylin.rest.response.NDataModelResponse;
import org.apache.kylin.rest.response.NDataSegmentResponse;
import org.apache.kylin.rest.response.NModelDescResponse;
import org.apache.kylin.rest.response.PurgeModelAffectedResponse;
import org.apache.kylin.rest.response.RelatedModelResponse;
import org.apache.kylin.rest.response.SegmentCheckResponse;
import org.apache.kylin.rest.response.SegmentPartitionResponse;
import org.apache.kylin.rest.response.SegmentRangeResponse;
import org.apache.kylin.rest.response.SimplifiedMeasure;
import org.apache.kylin.rest.service.merger.MetadataMerger;
import org.apache.kylin.rest.service.params.FullBuildSegmentParams;
import org.apache.kylin.rest.service.params.IncrementBuildSegmentParams;
import org.apache.kylin.rest.service.params.MergeSegmentParams;
import org.apache.kylin.rest.service.params.ModelQueryParams;
import org.apache.kylin.rest.util.AclPermissionUtil;
import org.apache.kylin.rest.util.ModelTriple;
import org.apache.kylin.rest.util.ModelUtils;
import org.apache.kylin.rest.util.PagingUtil;
import org.apache.kylin.source.SourceFactory;
import org.apache.kylin.source.adhocquery.PushDownConverterKeyWords;
import org.apache.kylin.streaming.event.StreamingJobDropEvent;
import org.apache.kylin.streaming.event.StreamingJobKillEvent;
import org.apache.kylin.streaming.manager.StreamingJobManager;
import org.apache.spark.sql.SparderEnv;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.security.access.AccessDeniedException;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.security.core.userdetails.UserDetails;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.core.JsonProcessingException;

import lombok.Setter;
import lombok.val;
import lombok.var;
import lombok.experimental.Delegate;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component("modelService")
public class ModelService extends AbstractModelService
        implements TableModelSupporter, ProjectModelSupporter, MetadataContract {

    private static final Logger logger = LoggerFactory.getLogger(ModelService.class);

    private static final String LAST_MODIFY = "last_modify";
    public static final String RECOMMENDATIONS_COUNT_LOWER_UNDERSCORE = "recommendations_count";

    public static final Pattern VALID_NAME_FOR_DIMENSION = Pattern.compile("^[\\u4E00-\\u9FA5a-zA-Z0-9 _\\-()%?（）]+$");

    public static final Pattern VALID_NAME_FOR_MEASURE = Pattern.compile("^[\\u4E00-\\u9FA5a-zA-Z0-9 _\\-()%?（）.]+$");

    private static final List<String> MODEL_CONFIG_BLOCK_LIST = Lists.newArrayList("kylin.index.rule-scheduler-data");
    private static final Set<String> STRING_TYPE_SET = Sets.newHashSet("STRING", "CHAR", "VARCHAR");

    //The front-end supports only the following formats
    private static final List<String> SUPPORTED_FORMATS = ImmutableList.of("ZZ", "DD", "D", "Do", "dddd", "ddd", "dd", //
            "d", "MMM", "MM", "M", "yyyy", "yy", "hh", "hh", "h", "HH", "H", "m", "mm", "ss", "s", "SSS", "SS", "S", //
            "A", "a");
    private static final Pattern QUOTE_PATTERN = Pattern.compile("\'(.*?)\'");

    @Setter
    @Autowired
    private ModelSemanticHelper semanticUpdater;

    @Autowired
    private ProjectService projectService;

    @Setter
    @Autowired(required = false)
    private ModelQuerySupporter modelQuerySupporter;

    @Setter
    @Autowired
    private IndexPlanService indexPlanService;

    @Autowired(required = false)
    @Qualifier("modelBuildService")
    private ModelBuildSupporter modelBuildService;

    @Setter
    @Autowired(required = false)
    private List<ModelChangeSupporter> modelChangeSupporters = Lists.newArrayList();

    @Delegate
    private ModelMetadataBaseService modelMetadataBaseService = new ModelMetadataBaseService();

    public NDataModel getModelById(String modelId, String project) {
        NDataModelManager modelManager = getManager(NDataModelManager.class, project);
        NDataModel nDataModel = modelManager.getDataModelDesc(modelId);
        if (null == nDataModel) {
            throw new KylinException(MODEL_ID_NOT_EXIST, modelId);
        }
        return nDataModel;
    }

    public NDataModel getModelByAlias(String modelAlias, String project) {
        NDataModelManager modelManager = getManager(NDataModelManager.class, project);
        NDataModel nDataModel = modelManager.getDataModelDescByAlias(modelAlias);
        if (null == nDataModel) {
            throw new KylinException(MODEL_NAME_NOT_EXIST, modelAlias);
        }
        return nDataModel;
    }

    public NDataModel getModel(String modelAliasOrUuid, String project) {
        NDataModel model = getManager(NDataModelManager.class, project).listAllModels().stream() //
                .filter(dataModel -> dataModel.getUuid().equals(modelAliasOrUuid) //
                        || dataModel.getAlias().equalsIgnoreCase(modelAliasOrUuid))
                .findFirst().orElse(null);

        if (model == null) {
            throw new KylinException(MODEL_NAME_NOT_EXIST, modelAliasOrUuid);
        }
        if (model.isBroken()) {
            throw new KylinException(ServerErrorCode.MODEL_BROKEN,
                    String.format(Locale.ROOT, MsgPicker.getMsg().getBrokenModelOperationDenied(), modelAliasOrUuid));
        }
        return model;
    }

    public NDataModel getModelWithoutBrokenCheck(String modelAliasOrUuid, String project) {
        NDataModel model = getManager(NDataModelManager.class, project).listAllModels().stream() //
                .filter(dataModel -> dataModel.getUuid().equals(modelAliasOrUuid) //
                        || dataModel.getAlias().equalsIgnoreCase(modelAliasOrUuid))
                .findFirst().orElse(null);

        if (model == null) {
            throw new KylinException(MODEL_NAME_NOT_EXIST, modelAliasOrUuid);
        }
        return model;
    }

    public List<String> getModelNamesByFuzzyName(String fuzzyName, String project, boolean exact) {
        if (StringUtils.isNotEmpty(project)) {
            NDataModelManager modelManager = getManager(NDataModelManager.class, project);
            return matchModelNameByAlias(modelManager, fuzzyName, exact);
        }

        List<String> modelAlias = new ArrayList<>();
        // query from all projects
        List<ProjectInstance> projectInstances = projectService.getReadableProjects(null, false);
        for (ProjectInstance projectInstance : projectInstances) {
            NDataModelManager modelManager = getManager(NDataModelManager.class, projectInstance.getName());
            modelAlias.addAll(matchModelNameByAlias(modelManager, fuzzyName, exact));
        }
        return modelAlias;
    }

    private List<String> matchModelNameByAlias(NDataModelManager modelManager, String fuzzyName, boolean exact) {
        if (!exact) {
            return modelManager.getModelNamesByFuzzyName(fuzzyName);
        }
        List<String> modelAlias = Lists.newArrayList();
        NDataModel nDataModel = modelManager.getDataModelDescByAlias(fuzzyName);
        if (null != nDataModel) {
            modelAlias.add(nDataModel.getAlias());
        }
        return modelAlias;
    }

    /**
     * for 3x rest api
     *
     * @param modelList
     * @return
     */
    public List<NDataModel> addOldParams(String project, List<NDataModel> modelList) {
        val executables = getAllRunningExecutable(project);
        modelList.forEach(model -> {
            NDataModelOldParams oldParams = new NDataModelOldParams();
            oldParams.setName(model.getAlias());
            oldParams.setJoinTables(model.getJoinTables());

            if (!model.isBroken()) {
                addOldSegmentParams(model, oldParams, executables);
            }

            if (model instanceof NDataModelResponse) {
                oldParams.setProjectName(model.getProject());
                oldParams.setSizeKB(((NDataModelResponse) model).getStorage() / 1024);
                oldParams.setDimensions(((NDataModelResponse) model).getNamedColumns());
                ((NDataModelResponse) model).setOldParams(oldParams);
            } else if (model instanceof RelatedModelResponse) {
                ((RelatedModelResponse) model).setOldParams(oldParams);
            }
        });

        return modelList;
    }

    private void addOldSegmentParams(NDataModel model, NDataModelOldParams oldParams,
            List<AbstractExecutable> executables) {
        List<NDataSegmentResponse> segments = getSegmentsResponse(model.getId(), model.getProject(), "1",
                String.valueOf(Long.MAX_VALUE - 1), null, executables, LAST_MODIFY, true,
                (model instanceof NDataModelLiteResponse));
        calculateRecordSizeAndCount(segments, oldParams);
        if (model instanceof NDataModelResponse) {
            ((NDataModelResponse) model).setSegments(segments);
            ((NDataModelResponse) model).setHasSegments(
                    ((NDataModelResponse) model).isHasSegments() || CollectionUtils.isNotEmpty(segments));
            checkSegmentOverlap((NDataModelResponse) model, segments);
        }

        if (model.isFusionModel()) {
            FusionModel fusionModel = getManager(FusionModelManager.class, model.getProject())
                    .getFusionModel(model.getId());
            NDataModel batchModel = fusionModel.getBatchModel();
            if (!batchModel.isBroken()) {
                List<NDataSegmentResponse> batchSegments = getSegmentsResponse(batchModel.getUuid(), model.getProject(),
                        "1", String.valueOf(Long.MAX_VALUE - 1), null, executables, LAST_MODIFY, true, false);
                calculateRecordSizeAndCount(batchSegments, oldParams);
                if (model instanceof FusionModelResponse) {
                    ((FusionModelResponse) model).setBatchSegments(batchSegments);
                }
            }
        }
    }

    private void calculateRecordSizeAndCount(List<NDataSegmentResponse> segments, NDataModelOldParams oldParams) {
        long inputRecordCnt = oldParams.getInputRecordCnt();
        long inputRecordSizeBytes = oldParams.getInputRecordSizeBytes();
        for (NDataSegmentResponse segment : segments) {
            long sourceRows = segment.getSourceCount();
            inputRecordCnt += sourceRows;
            inputRecordSizeBytes += segment.getSourceBytesSize();

            NDataSegmentResponse.OldParams segmentOldParams = new NDataSegmentResponse.OldParams();
            segmentOldParams.setSizeKB(segment.getBytesSize() / 1024);
            segmentOldParams.setInputRecords(sourceRows);
            segment.setOldParams(segmentOldParams);
        }
        oldParams.setInputRecordCnt(inputRecordCnt);
        oldParams.setInputRecordSizeBytes(inputRecordSizeBytes);
    }

    @VisibleForTesting
    public Set<String> getAllProjects() {
        return projectService.getReadableProjects().stream().map(ProjectInstance::getName).collect(Collectors.toSet());
    }

    public boolean isProjectNotExist(String project) {
        List<ProjectInstance> projectInstances = projectService.getReadableProjects(project, true);
        return CollectionUtils.isEmpty(projectInstances);
    }

    /**
     * for 3x rest api
     */
    public NDataModelResponse getCube(String modelAlias, String projectName) {
        if (Objects.nonNull(projectName)) {
            List<NDataModelResponse> cubes = getCubes(modelAlias, projectName);
            if (!CollectionUtils.isEmpty(cubes)) {
                return cubes.get(0);
            }
        } else {
            for (String project : getAllProjects()) {
                List<NDataModelResponse> cubes = getCubes(modelAlias, project);
                if (!CollectionUtils.isEmpty(cubes)) {
                    return cubes.get(0);
                }
            }
        }

        return null;
    }

    /**
     * for 3x rest api
     *
     * @param modelAlias
     * @param projectName
     * @return
     */
    public List<NDataModelResponse> getCubes(String modelAlias, String projectName) {
        if (Objects.nonNull(projectName)) {
            return getCubes0(modelAlias, projectName);
        }

        List<NDataModelResponse> cubes = Lists.newArrayList();
        for (String project : getAllProjects()) {
            cubes.addAll(getCubes0(modelAlias, project));
        }

        return cubes;
    }

    private boolean isAggGroupIncludeAllJoinCol(List<Set<String>> aggIncludes, List<String> needCols) {
        for (Set<String> includes : aggIncludes) {
            // if one of aggGroup contains all join column then the column would return
            boolean allIn = includes.containsAll(needCols);
            if (allIn) {
                return true;
            }
        }
        return false;
    }

    private List<NCubeDescResponse.Dimension3X> getDimension3XES(IndexPlan indexPlan, NDataModelResponse cube,
            List<AggGroupResponse> aggGroupResponses, NDataModel dataModel) {
        String rootFactTable = cube.getRootFactTableName();
        List<NCubeDescResponse.Dimension3X> dims = new ArrayList<>();
        HashMap<String, String> fk2Pk = Maps.newHashMap();
        cube.getJoinTables().forEach(join -> {
            String[] pks = join.getJoin().getPrimaryKey();
            String[] fks = join.getJoin().getForeignKey();
            for (int i = 0; i < pks.length; ++i)
                fk2Pk.put(fks[i], pks[i]);
        });

        HashMap<String, List<String>> tableToAllDim = Maps.newHashMap();
        cube.getNamedColumns().forEach(namedColumn -> {
            String aliasDotColumn = namedColumn.getAliasDotColumn();
            String table = aliasDotColumn.split("\\.")[0];
            String column = aliasDotColumn.split("\\.")[1];
            if (!tableToAllDim.containsKey(table)) {
                tableToAllDim.put(table, Lists.newArrayList());
            }
            tableToAllDim.get(table).add(column);
        });

        Set<String> allAggDim = Sets.newHashSet();//table.col
        indexPlan.getRuleBasedIndex().getDimensions().stream().map(x -> dataModel.getEffectiveDimensions().get(x))
                .forEach(x -> allAggDim.add(x.getIdentity()));

        List<Set<String>> aggIncludes = new ArrayList<>();
        aggGroupResponses.forEach(agg -> aggIncludes.add(Sets.newHashSet(agg.getIncludes())));

        cube.getNamedColumns().forEach(namedColumn -> {
            String aliasDotColumn = namedColumn.getAliasDotColumn();
            String table = aliasDotColumn.split("\\.")[0];
            boolean isRootFactTable = rootFactTable.endsWith("." + table);
            if (isRootFactTable) {
                if (allAggDim.contains(aliasDotColumn)) {
                    dims.add(new NCubeDescResponse.Dimension3X(namedColumn, false));
                }
            } else {
                List<String> needCols = new ArrayList<>();
                List<String> dimTableCol = tableToAllDim.get(table);
                dimTableCol.stream().filter(x -> fk2Pk.containsKey(table + "." + x)).forEach(x -> {
                    needCols.add(x);
                    needCols.add(fk2Pk.get(x));
                });
                if (isAggGroupIncludeAllJoinCol(aggIncludes, needCols)) {
                    boolean isDerived = !allAggDim.contains(aliasDotColumn);
                    dims.add(new NCubeDescResponse.Dimension3X(namedColumn, isDerived));
                }
            }
        });
        return dims;
    }

    public NCubeDescResponse getCubeWithExactModelName(String modelAlias, String projectName) {
        NDataModel dataModel = getManager(NDataModelManager.class, projectName).getDataModelDescByAlias(modelAlias);
        if (dataModel == null) {
            throw new KylinException(MODEL_NAME_NOT_EXIST, modelAlias);
        }
        NDataModelResponse cube = new NDataModelResponse(dataModel);
        NCubeDescResponse result = new NCubeDescResponse();

        result.setUuid(cube.getUuid());
        result.setName(cube.getAlias());
        result.setMeasures(
                cube.getMeasures().stream().map(NCubeDescResponse.Measure3X::new).collect(Collectors.toList()));

        IndexPlan indexPlan = getIndexPlan(result.getUuid(), projectName);
        if (!dataModel.isBroken() && indexPlan.getRuleBasedIndex() != null) {
            List<AggGroupResponse> aggGroupResponses = indexPlan.getRuleBasedIndex().getAggregationGroups().stream()
                    .map(x -> new AggGroupResponse(dataModel, x)).collect(Collectors.toList());
            result.setAggregationGroups(aggGroupResponses);
            result.setDimensions(getDimension3XES(indexPlan, cube, aggGroupResponses, dataModel));
        } else {
            result.setAggregationGroups(new ArrayList<>());
            result.setDimensions(new ArrayList<>());
        }

        return result;
    }

    /**
     * for 3x rest api
     *
     * @param modelAlias
     * @param project
     * @return
     */
    public List<NDataModelResponse> getCubes0(String modelAlias, String project) {
        Preconditions.checkNotNull(project);

        List<NDataModelResponse> modelResponseList = getModels(modelAlias, project, true, null, null, LAST_MODIFY,
                true);

        modelResponseList.forEach(modelResponse -> {
            NDataModelOldParams oldParams = new NDataModelOldParams();

            long inputRecordCnt = 0L;
            long inputRecordSizeBytes = 0L;
            if (!modelResponse.isModelBroken()) {
                List<NDataSegmentResponse> segments = getSegmentsResponse(modelResponse.getId(), project, "1",
                        String.valueOf(Long.MAX_VALUE - 1), null, LAST_MODIFY, true);
                for (NDataSegmentResponse segment : segments) {
                    long sourceRows = segment.getSourceCount();
                    inputRecordCnt += sourceRows;
                    inputRecordSizeBytes += segment.getSourceBytesSize();

                    NDataSegmentResponse.OldParams segmentOldParams = new NDataSegmentResponse.OldParams();
                    segmentOldParams.setSizeKB(segment.getBytesSize() / 1024);
                    segmentOldParams.setInputRecords(sourceRows);
                    segment.setOldParams(segmentOldParams);
                }

                modelResponse.setSegments(segments);
                modelResponse.setHasSegments(modelResponse.isHasSegments() || CollectionUtils.isNotEmpty(segments));

                checkSegmentOverlap(modelResponse, segments);
            }

            oldParams.setName(modelResponse.getAlias());
            oldParams.setProjectName(project);
            oldParams.setStreaming(false);
            oldParams.setSizeKB(modelResponse.getStorage() / 1024);
            oldParams.setInputRecordSizeBytes(inputRecordSizeBytes);
            oldParams.setInputRecordCnt(inputRecordCnt);
            oldParams.setJoinTables(modelResponse.getJoinTables());
            modelResponse.setOldParams(oldParams);
        });

        return modelResponseList;
    }

    private boolean isListContains(List<String> status, ModelStatusToDisplayEnum modelStatus) {
        return status == null || status.isEmpty() || (modelStatus != null && status.contains(modelStatus.name()));
    }

    public List<NDataModelResponse> getModels(final String modelAlias, final String projectName, boolean exactMatch,
            String owner, List<String> status, String sortBy, boolean reverse) {
        return getModels(modelAlias, projectName, exactMatch, owner, status, sortBy, reverse, null, null, null);
    }

    public List<NDataModelResponse> getSCD2ModelsByStatus(final String projectName, final List<String> status) {
        return getModels(null, projectName, false, null, status, LAST_MODIFY, true, null, null, null).stream()
                .filter(SCD2CondChecker.INSTANCE::isScd2Model).collect(Collectors.toList());
    }

    public List<String> getSCD2ModelsAliasByStatus(final String projectName, final List<String> status) {
        return getSCD2ModelsByStatus(projectName, status).stream().map(NDataModel::getAlias)
                .collect(Collectors.toList());
    }

    private List<String> getSCD2Models(final String projectName) {
        return getSCD2ModelsByStatus(projectName, null).stream().map(NDataModel::getId).collect(Collectors.toList());
    }

    public List<String> getModelNonOffOnlineStatus() {
        return Arrays.asList(ModelStatusToDisplayEnum.ONLINE.name(), ModelStatusToDisplayEnum.WARNING.name());
    }

    public List<String> getMultiPartitionModelsAlias(final String projectName, final List<String> status) {
        return getMultiPartitionModelsByStatus(projectName, status).stream().map(NDataModel::getAlias)
                .collect(Collectors.toList());
    }

    public List<NDataModelResponse> getMultiPartitionModelsByStatus(final String projectName,
            final List<String> status) {
        return getModels(null, projectName, false, null, status, LAST_MODIFY, true, null, null, null).stream()
                .filter(NDataModel::isMultiPartitionModel).collect(Collectors.toList());
    }

    public DataResult<List<NDataModel>> getModels(ModelQueryParams params) {
        List<NDataModel> models = new ArrayList<>();
        DataResult<List<NDataModel>> filterModels;
        val table = params.getTable();
        val project = params.getProjectName();
        val modelAttributes = params.getModelAttributes();
        val modelId = params.getModelId();
        val offset = params.getOffset();
        val limit = params.getLimit();
        if (StringUtils.isEmpty(table)) {
            val tripleList = modelQuerySupporter.getModels(params);
            val pair = getModelsOfCurrentPage(params, tripleList, params.isLite());
            models.addAll(pair.getFirst());
            filterModels = new DataResult<>(models, pair.getSecond(), offset, limit);
            filterModels.setValue(addOldParams(project, filterModels.getValue()));
            filterModels.setValue(updateResponseAcl(filterModels.getValue(), project));
            return filterModels;
        }
        Set<NDataModel> filteredModels = ModelUtils.getFilteredModels(modelAttributes, models);

        if (CollectionUtils.isNotEmpty(modelAttributes)) {
            models = models.stream().filter(filteredModels::contains).collect(Collectors.toList());
        }
        if (StringUtils.isNotEmpty(modelId)) {
            models.removeIf(model -> !model.getUuid().equals(modelId));
        }
        filterModels = DataResult.get(models, offset, limit);

        filterModels.setValue(addOldParams(project, filterModels.getValue()));
        filterModels.setValue(updateResponseAcl(filterModels.getValue(), project));
        return filterModels;
    }

    public Pair<List<NDataModelResponse>, Integer> getModelsOfCurrentPage(ModelQueryParams queryElem,
            List<ModelTriple> modelTripleList, boolean lite) {
        val projectName = queryElem.getProjectName();
        val offset = queryElem.getOffset();
        val limit = queryElem.getLimit();
        val status = queryElem.getStatus();
        val dfManager = getManager(NDataflowManager.class, projectName);
        List<NDataModelResponse> filterModels = new ArrayList<>();
        final AtomicInteger totalSize = new AtomicInteger();
        modelTripleList.stream().map(t -> {
            if ((status == null || status.isEmpty()) && !PagingUtil.isInCurrentPage(totalSize.get(), offset, limit)) {
                totalSize.getAndIncrement();
                return null;
            }
            NDataModel dataModel = t.getDataModel();
            try {
                NDataModelResponse nDataModelResponse = convertToDataModelResponse(dataModel, projectName, dfManager,
                        status, queryElem.isOnlyNormalDim());
                if (lite && nDataModelResponse != null) {
                    return new NDataModelLiteResponse(nDataModelResponse, dataModel);
                } else {
                    return nDataModelResponse;
                }
            } catch (Exception e) {
                String errorMsg = String.format(Locale.ROOT,
                        "convert NDataModelResponse failed, mark to broken. %s, %s", dataModel.getAlias(),
                        dataModel.getUuid());
                logger.error(errorMsg, e);
                return convertToDataModelResponseBroken(dataModel);
            }
        }).filter(Objects::nonNull).forEach(nDataModelResponse -> {
            if (PagingUtil.isInCurrentPage(totalSize.get(), offset, limit)) {
                filterModels.add(nDataModelResponse);
            }
            totalSize.getAndIncrement();
        });

        return new Pair<>(filterModels, totalSize.get());
    }

    public NDataModelResponse convertToDataModelResponseBroken(NDataModel modelDesc) {
        NDataModelResponse response = new NDataModelResponse(modelDesc);
        response.setStatus(ModelStatusToDisplayEnum.BROKEN);
        return response;
    }

    public List<NDataModelResponse> getModels(final String modelAlias, final String projectName, boolean exactMatch,
            String owner, List<String> status, String sortBy, boolean reverse, String modelAliasOrOwner,
            Long lastModifyFrom, Long lastModifyTo) {
        return getModels(modelAlias, projectName, exactMatch, owner, status, sortBy, reverse, modelAliasOrOwner,
                lastModifyFrom, lastModifyTo, true);
    }

    public List<NDataModelResponse> getModels(final String modelAlias, final String projectName, boolean exactMatch,
            String owner, List<String> status, String sortBy, boolean reverse, String modelAliasOrOwner,
            Long lastModifyFrom, Long lastModifyTo, boolean onlyNormalDim) {
        aclEvaluate.checkProjectReadPermission(projectName);
        List<Pair<NDataflow, NDataModel>> pairs = getFirstMatchModels(modelAlias, projectName, exactMatch, owner,
                modelAliasOrOwner, lastModifyFrom, lastModifyTo);
        val dfManager = getManager(NDataflowManager.class, projectName);
        List<NDataModelResponse> filterModels = new ArrayList<>();
        pairs.forEach(p -> {
            val modelDesc = p.getValue();
            val dataModelResponse = convertToDataModelResponse(modelDesc, projectName, dfManager, status,
                    onlyNormalDim);
            if (dataModelResponse != null) {
                filterModels.add(dataModelResponse);
            }
        });

        if ("expansionrate".equalsIgnoreCase(sortBy)) {
            return sortExpansionRate(reverse, filterModels);
        } else if (getManager(NProjectManager.class).getProject(projectName).isSemiAutoMode()
                || RECOMMENDATIONS_COUNT_LOWER_UNDERSCORE.equalsIgnoreCase(sortBy)) {
            Comparator<NDataModelResponse> comparator = BasicService.propertyComparator(
                    StringUtils.isEmpty(sortBy) ? RECOMMENDATIONS_COUNT_LOWER_UNDERSCORE : sortBy, !reverse);
            filterModels.sort(comparator);
            return filterModels;
        } else {
            Comparator<NDataModelResponse> comparator = BasicService
                    .propertyComparator(StringUtils.isEmpty(sortBy) ? ModelService.LAST_MODIFY : sortBy, !reverse);
            filterModels.sort(comparator);
            return filterModels;
        }
    }

    public NDataModelResponse convertToDataModelResponse(NDataModel modelDesc, String projectName,
            NDataflowManager dfManager, List<String> status, boolean onlyNormalDim) {
        long inconsistentSegmentCount = dfManager.getDataflow(modelDesc.getId()).getSegments(SegmentStatusEnum.WARNING)
                .size();
        ModelStatusToDisplayEnum modelResponseStatus = convertModelStatusToDisplay(modelDesc, projectName,
                inconsistentSegmentCount);
        if (modelDesc.isFusionModel()) {
            modelResponseStatus = convertFusionModelStatusToDisplay(modelDesc, modelResponseStatus, projectName,
                    dfManager);
        }
        boolean isModelStatusMatch = isListContains(status, modelResponseStatus);
        if (isModelStatusMatch) {
            boolean isScd2ForbiddenOnline = checkSCD2ForbiddenOnline(modelDesc, projectName);
            NDataModelResponse nDataModelResponse = enrichModelResponse(modelDesc, projectName);
            nDataModelResponse.computedInfo(inconsistentSegmentCount, modelResponseStatus, isScd2ForbiddenOnline,
                    modelDesc, onlyNormalDim);
            return nDataModelResponse;
        } else {
            return null;
        }
    }

    private ModelStatusToDisplayEnum convertFusionModelStatusToDisplay(NDataModel modelDesc,
            ModelStatusToDisplayEnum modelResponseStatus, String projectName, NDataflowManager dfManager) {
        FusionModel fusionModel = FusionModelManager.getInstance(KylinConfig.getInstanceFromEnv(), projectName)
                .getFusionModel(modelDesc.getFusionId());
        val batchModel = fusionModel.getBatchModel();
        long inconsistentBatchSegmentCount = dfManager.getDataflow(batchModel.getId())
                .getSegments(SegmentStatusEnum.WARNING).size();
        ModelStatusToDisplayEnum batchModelResponseStatus = convertModelStatusToDisplay(batchModel, projectName,
                inconsistentBatchSegmentCount);
        if (!batchModel.isBroken() && !modelDesc.isBroken()) {
            switch (modelResponseStatus) {
            case ONLINE:
                return (batchModelResponseStatus == ModelStatusToDisplayEnum.WARNING ? ModelStatusToDisplayEnum.WARNING
                        : modelResponseStatus);
            case OFFLINE:
                return batchModelResponseStatus;
            default:
                return modelResponseStatus;
            }
        } else {
            return modelResponseStatus;
        }
    }

    private List<Pair<NDataflow, NDataModel>> getFirstMatchModels(final String modelAlias, final String projectName,
            boolean exactMatch, String owner, String modelAliasOrOwner, Long lastModifyFrom, Long lastModifyTo) {
        return getManager(NDataflowManager.class, projectName).listAllDataflows(true).stream()
                .map(df -> Pair.newPair(df,
                        df.checkBrokenWithRelatedInfo() ? modelQuerySupporter.getBrokenModel(projectName, df.getId())
                                : df.getModel()))
                .filter(p -> !(Objects.nonNull(lastModifyFrom) && lastModifyFrom > p.getValue().getLastModified())
                        && !(Objects.nonNull(lastModifyTo) && lastModifyTo <= p.getValue().getLastModified())
                        && (ModelUtils.isArgMatch(modelAliasOrOwner, exactMatch, p.getValue().getAlias())
                                || ModelUtils.isArgMatch(modelAliasOrOwner, exactMatch, p.getValue().getOwner()))
                        && ModelUtils.isArgMatch(modelAlias, exactMatch, p.getValue().getAlias())
                        && ModelUtils.isArgMatch(owner, exactMatch, p.getValue().getOwner())
                        && !p.getValue().fusionModelBatchPart())
                .collect(Collectors.toList());
    }

    public ModelStatusToDisplayEnum convertModelStatusToDisplay(NDataModel modelDesc, final String projectName,
            long inconsistentSegmentCount) {
        RealizationStatusEnum modelStatus = modelDesc.isBroken() ? RealizationStatusEnum.BROKEN
                : getModelStatus(modelDesc.getUuid(), projectName);
        ModelStatusToDisplayEnum modelResponseStatus = ModelStatusToDisplayEnum.convert(modelStatus);
        val segmentHoles = getManager(NDataflowManager.class, projectName).calculateSegHoles(modelDesc.getUuid());
        if (modelResponseStatus == ModelStatusToDisplayEnum.BROKEN
                || modelResponseStatus == ModelStatusToDisplayEnum.OFFLINE) {
            return modelResponseStatus;
        }
        if (getEmptyIndexesCount(projectName, modelDesc.getId()) > 0 || CollectionUtils.isNotEmpty(segmentHoles)
                || inconsistentSegmentCount > 0) {
            modelResponseStatus = ModelStatusToDisplayEnum.WARNING;
        }
        return modelResponseStatus;
    }

    private long getEmptyIndexesCount(String project, String modelId) {
        val indexPlanManager = getManager(NIndexPlanManager.class, project);
        val indexPlan = indexPlanManager.getIndexPlan(modelId);
        return indexPlan.getAllLayoutsReadOnly().size() - indexPlanManager.getAvailableIndexesCount(project, modelId);
    }

    private List<NDataModelResponse> sortExpansionRate(boolean reverse, List<NDataModelResponse> filterModels) {
        List<NDataModelResponse> sorted;
        if (!reverse) {
            sorted = filterModels.stream().sorted(Comparator.comparing(a -> new BigDecimal(a.getExpansionrate())))
                    .collect(Collectors.toList());
        } else {
            sorted = filterModels.stream().sorted(
                    (a, b) -> new BigDecimal(b.getExpansionrate()).compareTo(new BigDecimal(a.getExpansionrate())))
                    .collect(Collectors.toList());
        }
        List<NDataModelResponse> unknowModels = sorted.stream()
                .filter(modle -> "-1".equalsIgnoreCase(modle.getExpansionrate())).collect(Collectors.toList());

        List<NDataModelResponse> nDataModelResponses = sorted.stream()
                .filter(modle -> !"-1".equalsIgnoreCase(modle.getExpansionrate())).collect(Collectors.toList());
        nDataModelResponses.addAll(unknowModels);
        return nDataModelResponses;
    }

    private NDataModelResponse enrichModelResponse(NDataModel modelDesc, String projectName) {
        NDataModelResponse nDataModelResponse = modelDesc.isFusionModel() ? new FusionModelResponse(modelDesc)
                : new NDataModelResponse(modelDesc);

        if (modelDesc.isBroken()) {
            val tableManager = NTableMetadataManager.getInstance(KylinConfig.getInstanceFromEnv(), projectName);
            if (tableManager.getTableDesc(modelDesc.getRootFactTableName()) == null) {
                nDataModelResponse.setRootFactTableName(nDataModelResponse.getRootFactTableName() + " deleted");
                nDataModelResponse.setRootFactTableDeleted(true);
            }
            nDataModelResponse.setBroken(modelDesc.isBroken());
            return nDataModelResponse;
        }
        nDataModelResponse.setAllTableRefs(modelDesc.getAllTables());
        nDataModelResponse.setBroken(modelDesc.isBroken());
        if (ManagementType.MODEL_BASED == modelDesc.getManagementType()) {
            Segments<NDataSegment> segments = getSegmentsByRange(modelDesc.getUuid(), projectName, "0",
                    "" + Long.MAX_VALUE);
            if (CollectionUtils.isNotEmpty(segments)) {
                NDataSegment lastSegment = segments.get(segments.size() - 1);
                nDataModelResponse.setLastBuildEnd(lastSegment.getSegRange().getEnd().toString());
            } else {
                nDataModelResponse.setLastBuildEnd("");
            }
        }
        return nDataModelResponse;
    }

    private boolean checkSCD2ForbiddenOnline(NDataModel modelDesc, String projectName) {
        boolean isSCD2 = SCD2CondChecker.INSTANCE.isScd2Model(modelDesc);
        return !getManager(NProjectManager.class).getProject(projectName).getConfig().isQueryNonEquiJoinModelEnabled()
                && isSCD2;
    }

    public RealizationStatusEnum getModelStatus(String modelId, String projectName) {
        val indexPlan = getIndexPlan(modelId, projectName);
        if (indexPlan != null) {
            return getManager(NDataflowManager.class, projectName).getDataflow(indexPlan.getUuid()).getStatus();
        } else {
            return null;
        }
    }

    public List<NDataSegmentResponse> getSegmentsResponse(String modelId, String project, String start, String end,
            String status, String sortBy, boolean reverse) {
        return getSegmentsResponse(modelId, project, start, end, status, null, null, false, sortBy, reverse, null);
    }

    public List<NDataSegmentResponse> getSegmentsResponse(String modelId, String project, String start, String end,
            String status, List<AbstractExecutable> executables, String sortBy, boolean reverse, boolean lite) {
        return getSegmentsResponse(modelId, project, start, end, status, null, null, executables, false, sortBy,
                reverse, lite, null);
    }

    private List<AbstractExecutable> getAllRunningExecutable(String project) {
        KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
        ExecutableManager execManager = ExecutableManager.getInstance(kylinConfig, project);
        return execManager.getNotFinalExecutablesByType(
                Lists.newArrayList(JobTypeEnum.INDEX_BUILD, JobTypeEnum.SUB_PARTITION_BUILD));
    }

    private List<AbstractExecutable> getPartialRunningExecutable(String project, String modelId) {
        KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
        ExecutableManager execManager = ExecutableManager.getInstance(kylinConfig, project);
        return execManager.listPartialExec(modelId, ExecutableState::isRunning, JobTypeEnum.INDEX_BUILD,
                JobTypeEnum.SUB_PARTITION_BUILD).stream().map(execManager::fromPO).collect(Collectors.toList());
    }

    public List<NDataSegmentResponse> getSegmentsResponse(String modelId, String project, String start, String end,
            String status, Collection<Long> withAllIndexes, Collection<Long> withoutAnyIndexes, boolean allToComplement,
            String sortBy, boolean reverse, List<String> statuses) {
        val executables = getPartialRunningExecutable(project, modelId);
        return getSegmentsResponse(modelId, project, start, end, status, withAllIndexes, withoutAnyIndexes, executables,
                allToComplement, sortBy, reverse, false, statuses);
    }

    public List<NDataSegmentResponse> getSegmentsResponse(String modelId, String project, String start, String end,
            String status, Collection<Long> withAllIndexes, Collection<Long> withoutAnyIndexes,
            List<AbstractExecutable> executables, boolean allToComplement, String sortBy, boolean reverse, boolean lite,
            List<String> statuses) {
        aclEvaluate.checkProjectReadPermission(project);
        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        NDataflow dataflow = dataflowManager.getDataflow(modelId);
        List<NDataSegmentResponse> segmentResponseList = getSegmentsResponseCore(modelId, project, start, end, status,
                withAllIndexes, withoutAnyIndexes, executables, allToComplement, dataflow);
        segmentResponseList = segmentResponseFilter(statuses, segmentResponseList);
        segmentsResponseListSort(sortBy, reverse, segmentResponseList);
        return segmentResponseList;
    }

    public List<NDataSegmentResponse> getSegmentsResponseByJob(String modelId, String project, AbstractExecutable job) {
        aclEvaluate.checkProjectReadPermission(project);
        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        NDataflow dataflow = dataflowManager.getDataflow(modelId);
        List<NDataSegmentResponse> segmentResponseList = getSegmentsResponseCoreByJob(job, dataflow);
        return segmentResponseList;
    }

    private List<NDataSegmentResponse> getSegmentsResponseCoreByJob(AbstractExecutable job, NDataflow dataflow) {
        if (CollectionUtils.isEmpty(job.getSegmentIds())) {
            return Lists.newArrayList();
        }
        val segmentIds = Sets.newHashSet(job.getSegmentIds());
        val segs = dataflow == null ? Segments.empty() : dataflow.getSegments(segmentIds);
        List<AbstractExecutable> runningJob = job.getStatus() == ExecutableState.RUNNING ? Lists.newArrayList(job)
                : Lists.newArrayList();
        return segs.stream().map(segment -> new NDataSegmentResponse(dataflow, segment, runningJob))
                .collect(Collectors.toList());
    }

    public void segmentsResponseListSort(String sortBy, boolean reverse,
            List<NDataSegmentResponse> segmentResponseList) {
        Comparator<NDataSegmentResponse> comparator = BasicService
                .propertyComparator(StringUtils.isEmpty(sortBy) ? "create_time_utc" : sortBy, reverse);
        segmentResponseList.sort(comparator);
    }

    public List<NDataSegmentResponse> getSegmentsResponseCore(String modelId, String project, String start, String end,
            String status, Collection<Long> withAllIndexes, Collection<Long> withoutAnyIndexes,
            List<AbstractExecutable> executables, boolean allToComplement, NDataflow dataflow) {
        List<NDataSegmentResponse> segmentResponseList;
        val segs = getSegmentsByRange(modelId, project, start, end);

        // filtering on index
        IndexPlan indexPlan = NIndexPlanManager.getInstance(KylinConfig.getInstanceFromEnv(), project)
                .getIndexPlan(modelId);
        Set<Long> allIndexes = indexPlan.getAllLayoutIds(true);
        if (CollectionUtils.isNotEmpty(withAllIndexes)) {
            for (Long idx : withAllIndexes) {
                if (!allIndexes.contains(idx)) {
                    throw new KylinException(INVALID_SEGMENT_PARAMETER, "Invalid index id " + idx);
                }
            }
        }
        if (CollectionUtils.isNotEmpty(withoutAnyIndexes)) {
            for (Long idx : withoutAnyIndexes) {
                if (!allIndexes.contains(idx)) {
                    throw new KylinException(INVALID_SEGMENT_PARAMETER, "Invalid index id " + idx);
                }
            }
        }
        segmentResponseList = segs.stream()
                .filter(segment -> filterSeg(withAllIndexes, withoutAnyIndexes, allToComplement,
                        indexPlan.getAllLayoutIds(false), segment))
                .filter(segment -> !StringUtils.isNotEmpty(status) || status.equalsIgnoreCase(
                        SegmentUtil.getSegmentStatusToDisplay(segs, segment, executables, null).toString()))
                .map(segment -> new NDataSegmentResponse(dataflow, segment, executables)).collect(Collectors.toList());
        return segmentResponseList;
    }

    public List<NDataSegmentResponse> segmentResponseFilter(List<String> statuses,
            List<NDataSegmentResponse> segmentResponseList) {
        if (CollectionUtils.isEmpty(statuses)) {
            return segmentResponseList;
        }
        val statusEnumSet = statuses.stream().map(SegmentStatusEnumToDisplay::getByName).filter(Objects::nonNull)
                .collect(Collectors.toSet());
        return segmentResponseList.stream().filter(segmentResponse -> CollectionUtils.isEmpty(statusEnumSet)
                || statusEnumSet.contains(segmentResponse.getStatusToDisplay())).collect(Collectors.toList());
    }

    private boolean filterSeg(Collection<Long> withAllIndexes, Collection<Long> withoutAnyIndexes,
            boolean allToComplement, Set<Long> allIndexWithoutTobeDel, NDataSegment segment) {
        if (allToComplement) {
            // find seg that does not have all indexes(don't include tobeDeleted)
            val segLayoutIds = segment.getSegDetails().getEffectiveLayouts().stream().map(NDataLayout::getLayoutId)
                    .collect(Collectors.toSet());
            return !Sets.difference(allIndexWithoutTobeDel, segLayoutIds).isEmpty();
        }
        if (CollectionUtils.isNotEmpty(withAllIndexes)) {
            // find seg that has all required indexes
            return segment.getLayoutIds().containsAll(withAllIndexes);
        }
        if (CollectionUtils.isNotEmpty(withoutAnyIndexes)) {
            // find seg that miss any of the indexes
            return !segment.getLayoutIds().containsAll(withoutAnyIndexes);
        }
        return true;
    }

    public Segments<NDataSegment> getSegmentsByRange(String modelId, String project, String start, String end) {
        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        val df = dataflowManager.getDataflow(modelId);
        if (df == null) {
            return Segments.empty();
        }
        val model = df.getModel();
        if (model == null || model.isBroken()) {
            return Segments.empty();
        }
        SegmentRange filterRange = getSegmentRangeByModel(project, modelId, start, end);
        return df.getSegmentsByRange(filterRange);
    }

    public IndicesResponse getAggIndices(String project, String modelId, Long indexId, String contentSeg,
            boolean isCaseSensitive, Integer pageOffset, Integer pageSize, String sortBy, Boolean reverse) {
        aclEvaluate.checkProjectReadPermission(project);

        logger.debug("find project={}, model={}, index={}, content={}, isCaseSensitive={}, sortBy={}, reverse={}",
                project, modelId, indexId, contentSeg, isCaseSensitive, sortBy, reverse);

        IndicesResponse result;
        if (Objects.nonNull(indexId)) {
            result = getIndicesById(project, modelId, indexId);
        } else {
            IndexPlan indexPlan = getIndexPlan(modelId, project);
            result = new IndicesResponse(indexPlan);
            indexPlan.getAllIndexes().stream().filter(e -> e.getId() < IndexEntity.TABLE_INDEX_START_ID)
                    .forEach(result::addIndexEntity);
        }
        List<IndicesResponse.Index> indices = result.getIndices();
        if (Objects.nonNull(contentSeg)) {
            indices = filterFuzzyMatchedIndices(indices, contentSeg, isCaseSensitive);
        }
        result.setSize(indices.size());
        result.setIndices(sortIndicesThenCutPage(indices, sortBy, reverse, pageOffset, pageSize));
        return result;
    }

    private List<IndicesResponse.Index> filterFuzzyMatchedIndices(List<IndicesResponse.Index> indices,
            String contentSeg, boolean isCaseSensitive) {
        if (StringUtils.isBlank(contentSeg)) {
            return indices;
        }
        return indices.stream().filter(index -> // weird rule
        fuzzyMatched(contentSeg, isCaseSensitive, String.valueOf(index.getId()))
                || index.getDimensions().stream().anyMatch(d -> fuzzyMatched(contentSeg, isCaseSensitive, d))
                || index.getMeasures().stream().anyMatch(m -> fuzzyMatched(contentSeg, isCaseSensitive, m)))
                .collect(Collectors.toList());
    }

    private List<IndicesResponse.Index> sortIndicesThenCutPage(List<IndicesResponse.Index> indices, String sortBy,
            boolean reverse, int pageOffset, int pageSize) {
        Comparator<IndicesResponse.Index> comparator = BasicService
                .propertyComparator(StringUtils.isEmpty(sortBy) ? IndicesResponse.LAST_MODIFY_TIME : sortBy, !reverse);
        indices.sort(comparator);
        return PagingUtil.cutPage(indices, pageOffset, pageSize);
    }

    private boolean fuzzyMatched(String contentSeg, boolean isCaseSensitive, String content) {
        if (isCaseSensitive) {
            return StringUtils.contains(content, contentSeg);
        }
        return StringUtils.containsIgnoreCase(content, contentSeg);
    }

    @VisibleForTesting
    public IndicesResponse getIndicesById(String project, String modelId, Long indexId) {
        aclEvaluate.checkProjectReadPermission(project);
        IndexPlan indexPlan = getIndexPlan(modelId, project);
        IndicesResponse result = new IndicesResponse(indexPlan);
        result.addIndexEntity(indexPlan.getIndexEntity(indexId));
        return result;
    }

    public IndicesResponse getTableIndices(String modelId, String project) {
        aclEvaluate.checkProjectReadPermission(project);
        IndexPlan indexPlan = getIndexPlan(modelId, project);
        IndicesResponse result = new IndicesResponse(indexPlan);
        indexPlan.getAllIndexes().stream().filter(e -> IndexEntity.isTableIndex(e.getId()))
                .forEach(result::addIndexEntity);
        return result;
    }

    @VisibleForTesting
    IndicesResponse getIndices(String modelId, String project) {
        IndexPlan indexPlan = getIndexPlan(modelId, project);
        IndicesResponse result = new IndicesResponse(indexPlan);
        indexPlan.getAllIndexes().forEach(result::addIndexEntity);
        return result;
    }

    public String getModelJson(String modelId, String project) throws JsonProcessingException {
        aclEvaluate.checkProjectReadPermission(project);
        NDataModel modelDesc = getManager(NDataModelManager.class, project).getDataModelDesc(modelId);
        return JsonUtil.writeValueAsIndentString(modelDesc);
    }

    public String getModelSql(String modelId, String project) {
        aclEvaluate.checkProjectReadPermission(project);
        try (QueryContext queryContext = QueryContext.current()) {
            queryContext.setAclInfo(AclPermissionUtil.createAclInfo(project, getCurrentUserGroups()));
            NDataModel model = getManager(NDataModelManager.class, project).getDataModelDesc(modelId);
            return PushDownUtil.generateFlatTableSql(model, false);
        }
    }

    private void checkAliasIsExceededLimit(String newAlias) {
        if (newAlias.length() > Constant.MODEL_ALIAS_LEN_LIMIT) {
            throw new KylinException(MODEL_NAME_TOO_LONG);
        }
    }

    private void checkAliasExist(String modelId, String newAlias, String project) {
        if (!checkModelAliasUniqueness(modelId, newAlias, project)) {
            throw new KylinException(MODEL_NAME_DUPLICATE, newAlias);
        }
    }

    public boolean checkModelAliasUniqueness(String modelId, String newAlias, String project) {
        aclEvaluate.checkProjectWritePermission(project);
        List<NDataModel> models = getManager(NDataflowManager.class, project).listUnderliningDataModels();
        for (NDataModel model : models) {
            if ((StringUtils.isNotEmpty(modelId) || !model.getUuid().equals(modelId))
                    && model.getAlias().equalsIgnoreCase(newAlias)) {
                return false;
            }
        }
        return true;
    }

    @Transaction(project = 1)
    public void dropModel(String modelId, String project) {
        aclEvaluate.checkProjectWritePermission(project);
        checkModelPermission(project, modelId);
        val model = getModelById(modelId, project);
        val modelName = model.getAlias();
        innerDropModel(modelId, project);
        EventBusFactory.getInstance().postSync(new ModelDropEvent(project, modelId, modelName));
    }

    void innerDropModel(String modelId, String project) {
        NDataModel dataModelDesc = getModelById(modelId, project);
        boolean isStreamingModel = false;
        if (dataModelDesc.isStreaming()) {
            isStreamingModel = true;
        } else if (dataModelDesc.getModelType() == NDataModel.ModelType.UNKNOWN) {
            val streamingJobMgr = StreamingJobManager.getInstance(getConfig(), project);
            isStreamingModel = streamingJobMgr.getStreamingJobByUuid(modelId + "_build") != null
                    || streamingJobMgr.getStreamingJobByUuid(modelId + "_merge") != null;
        }
        if (isStreamingModel) {
            EventBusFactory.getInstance().postSync(new StreamingJobKillEvent(project, modelId));
            EventBusFactory.getInstance().postSync(new StreamingJobDropEvent(project, modelId));
        }

        val dataflowManager = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
        val indexPlanManager = NIndexPlanManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
        val dataModelManager = NDataModelManager.getInstance(KylinConfig.getInstanceFromEnv(), project);

        dataflowManager.dropDataflow(modelId);
        indexPlanManager.dropIndexPlan(modelId);

        dataModelManager.dropModel(dataModelDesc);
    }

    @Transaction(project = 1)
    public void purgeModel(String modelId, String project) {
        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        val indexPlan = getIndexPlan(modelId, project);
        List<NDataSegment> segments = new ArrayList<>();
        if (indexPlan != null) {
            NDataflow dataflow = dataflowManager.getDataflow(indexPlan.getUuid());
            NDataflowUpdate nDataflowUpdate = new NDataflowUpdate(dataflow.getUuid());
            segments.addAll(dataflow.getSegments());
            NDataSegment[] segmentsArray = new NDataSegment[segments.size()];
            NDataSegment[] nDataSegments = segments.toArray(segmentsArray);
            nDataflowUpdate.setToRemoveSegs(nDataSegments);
            dataflowManager.updateDataflow(nDataflowUpdate);
        }
        offlineModelIfNecessary(dataflowManager, modelId);
    }

    @Transaction(project = 1)
    public void purgeModelManually(String modelId, String project) {
        aclEvaluate.checkProjectOperationPermission(project);
        NDataModel dataModelDesc = getModelById(modelId, project);
        if (ManagementType.TABLE_ORIENTED == dataModelDesc.getManagementType()) {
            throw new KylinException(PERMISSION_DENIED,
                    String.format(Locale.ROOT, MsgPicker.getMsg().getModelCanNotPurge(), dataModelDesc.getAlias()));
        }
        purgeModel(modelId, project);
    }

    public void cloneModel(String modelId, String newModelName, String project) {
        aclEvaluate.checkProjectWritePermission(project);
        checkAliasExist("", newModelName, project);
        checkAliasIsExceededLimit(newModelName);
        EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
            NDataModelManager dataModelManager = getManager(NDataModelManager.class, project);
            NDataModel dataModelDesc = getModelById(modelId, project);
            //copyForWrite nDataModel do init,but can not set new modelname
            NDataModel nDataModel = semanticUpdater.deepCopyModel(dataModelDesc);
            nDataModel.setUuid(RandomUtil.randomUUIDStr());
            nDataModel.setAlias(newModelName);
            nDataModel.setLastModified(System.currentTimeMillis());
            nDataModel.setRecommendationsCount(0);
            nDataModel.setMvcc(-1);
            nDataModel.setProject(project);
            nDataModel.setComputedColumnDescs(dataModelDesc.getComputedColumnDescs());
            changeModelOwner(nDataModel);
            val newModel = dataModelManager.createDataModelDesc(nDataModel, nDataModel.getOwner());
            cloneIndexPlan(modelId, project, nDataModel.getOwner(), newModel.getUuid(), RealizationStatusEnum.OFFLINE);
            return null;
        }, project);
    }

    private void cloneIndexPlan(String modelId, String project, String owner, String newModelId,
            RealizationStatusEnum realizationStatusEnum) {
        NIndexPlanManager indexPlanManager = getManager(NIndexPlanManager.class, project);
        IndexPlan indexPlan = indexPlanManager.getIndexPlan(modelId);
        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        IndexPlan copy = indexPlanManager.copy(indexPlan);
        copy.setUuid(newModelId);
        copy.setLastModified(System.currentTimeMillis());
        copy.setMvcc(-1);
        Set<Long> toBeDeletedLayouts = copy.getToBeDeletedIndexes().stream()
                .flatMap(indexEntity -> indexEntity.getLayouts().stream()).map(LayoutEntity::getId)
                .collect(Collectors.toSet());
        copy.removeLayouts(toBeDeletedLayouts, true, true);
        indexPlanManager.createIndexPlan(copy);

        dataflowManager.createDataflow(copy, owner, realizationStatusEnum);
    }

    @Transaction(project = 0)
    public void renameDataModel(String project, String modelId, String newAlias, String description) {
        aclEvaluate.checkProjectWritePermission(project);
        NDataModelManager modelManager = getManager(NDataModelManager.class, project);
        NDataModel nDataModel = getModelById(modelId, project);
        //rename

        if (description != null && nDataModel.getAlias().equalsIgnoreCase(newAlias)) {
            nDataModel.setDescription(description);
        } else {
            checkAliasExist(modelId, newAlias, project);
            checkAliasIsExceededLimit(newAlias);
            nDataModel.setAlias(newAlias);
            if (description != null) {
                nDataModel.setDescription(description);
            }
        }

        NDataModel modelUpdate = modelManager.copyForWrite(nDataModel);
        modelManager.updateDataModelDesc(modelUpdate);
    }

    @Transaction(project = 0)
    public void offlineAllModelsInProject(String project) {
        aclEvaluate.checkProjectWritePermission(project);
        Set<String> ids = listAllModelIdsInProject(project);
        for (String id : ids) {
            updateDataModelStatus(id, project, "OFFLINE");
        }
    }

    @Transaction(project = 0)
    public void offlineMultiPartitionModelsInProject(String project) {
        aclEvaluate.checkProjectWritePermission(project);
        List<String> multiPartitionModels = getMultiPartitionModelsByStatus(project, getModelNonOffOnlineStatus())
                .stream().map(NDataModel::getId).collect(Collectors.toList());
        if (CollectionUtils.isEmpty(multiPartitionModels)) {
            return;
        }
        offlineModelsInProjectById(project, new HashSet<>(multiPartitionModels));
    }

    @Transaction(project = 0)
    public void offlineModelsInProjectById(String project, Set<String> modelIds) {
        aclEvaluate.checkProjectWritePermission(project);
        for (String id : modelIds) {
            if (modelIds.contains(id)) {
                updateDataModelStatus(id, project, ModelStatusToDisplayEnum.OFFLINE.name());
            }
        }
    }

    @Transaction(project = 0)
    public void updateSCD2ModelStatusInProjectById(String project, ModelStatusToDisplayEnum status) {
        aclEvaluate.checkProjectWritePermission(project);
        List<String> scd2Models = getSCD2Models(project);
        if (CollectionUtils.isEmpty(scd2Models)) {
            return;
        }
        updateModelStatusInProjectById(project, new HashSet<>(scd2Models), status);
    }

    @Transaction(project = 0)
    public void updateModelStatusInProjectById(String project, Set<String> modelIds, ModelStatusToDisplayEnum status) {
        aclEvaluate.checkProjectWritePermission(project);
        for (String id : modelIds) {
            try {
                updateDataModelStatus(id, project, status.name());
            } catch (Exception e) {
                logger.warn("Failed update model {} status to {}, {}", id, status.name(), e.getMessage());
            }
        }
    }

    @Transaction(project = 0)
    public void onlineAllModelsInProject(String project) {
        aclEvaluate.checkProjectWritePermission(project);
        Set<String> ids = listAllModelIdsInProject(project);
        for (String id : ids) {
            updateDataModelStatus(id, project, "ONLINE");
        }
    }

    @Transaction(project = 1)
    public void updateDataModelStatus(String modelId, String project, String status) {
        NDataModel nDataModel = getModelById(modelId, project);
        if (nDataModel.isFusionModel()) {
            NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
            NDataflow dataflow = dataflowManager.getDataflow(nDataModel.getUuid());
            if (CollectionUtils.isNotEmpty(dataflow.getSegments())) {
                doUpdateDataModelStatus(nDataModel, project, status);
            }
            val fusionId = nDataModel.getFusionId();
            val fusionModelMgr = FusionModelManager.getInstance(getConfig(), project);
            val batchModel = fusionModelMgr.getFusionModel(fusionId).getBatchModel();
            if (batchModel != null) {
                dataflow = dataflowManager.getDataflow(batchModel.getUuid());
                if (CollectionUtils.isNotEmpty(dataflow.getSegments())) {
                    doUpdateDataModelStatus(batchModel, project, status);
                }
            }
        } else {
            doUpdateDataModelStatus(nDataModel, project, status);
        }
    }

    private void doUpdateDataModelStatus(NDataModel nDataModel, String project, String status) {
        String modelId = nDataModel.getUuid();
        aclEvaluate.checkProjectWritePermission(project);
        IndexPlan indexPlan = getIndexPlan(nDataModel.getUuid(), project);
        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        NDataflow dataflow = dataflowManager.getDataflow(indexPlan.getUuid());
        checkDataflowStatus(dataflow, modelId);
        boolean needChangeStatus = (status.equals(RealizationStatusEnum.OFFLINE.name())
                && RealizationStatusEnum.ONLINE == dataflow.getStatus())
                || (status.equals(RealizationStatusEnum.ONLINE.name())
                        && RealizationStatusEnum.OFFLINE == dataflow.getStatus());
        if (needChangeStatus) {
            NDataflowUpdate nDataflowUpdate = new NDataflowUpdate(dataflow.getUuid());
            if (status.equals(RealizationStatusEnum.OFFLINE.name())) {
                nDataflowUpdate.setStatus(RealizationStatusEnum.OFFLINE);
            } else if (status.equals(RealizationStatusEnum.ONLINE.name())) {
                if (SCD2CondChecker.INSTANCE.isScd2Model(dataflow.getModel())
                        && !projectService.getProjectConfig(project).isScd2Enabled()) {
                    throw new KylinException(MODEL_ONLINE_ABANDON,
                            MsgPicker.getMsg().getScd2ModelOnlineWithScd2ConfigOff());
                }
                if (dataflow.getSegments().isEmpty() && !KylinConfig.getInstanceFromEnv().isUTEnv()) {
                    throw new KylinException(MODEL_ONLINE_ABANDON, MsgPicker.getMsg().getModelOnlineWithEmptySeg());
                }
                if (dataflowManager.isOfflineModel(dataflow)) {
                    throw new KylinException(MODEL_ONLINE_ABANDON, MsgPicker.getMsg().getModelOnlineForbidden());
                }
                nDataflowUpdate.setStatus(RealizationStatusEnum.ONLINE);
            }
            dataflowManager.updateDataflow(nDataflowUpdate);
        }
    }

    private void checkDataflowStatus(NDataflow dataflow, String modelId) {
        if (RealizationStatusEnum.BROKEN == dataflow.getStatus()) {
            throw new KylinException(DUPLICATE_JOIN_CONDITION,
                    String.format(Locale.ROOT, MsgPicker.getMsg().getBrokenModelCannotOnoffline(), modelId));
        }
    }

    public SegmentRange getSegmentRangeByModel(String project, String modelId, String start, String end) {
        TableRef tableRef = getManager(NDataModelManager.class, project).getDataModelDesc(modelId).getRootFactTable();
        TableDesc tableDesc = getManager(NTableMetadataManager.class, project)
                .getTableDesc(tableRef.getTableIdentity());
        return SourceFactory.getSource(tableDesc).getSegmentRange(start, end);
    }

    public boolean isModelsUsingTable(String table, String project) {
        TableDesc tableDesc = getManager(NTableMetadataManager.class, project).getTableDesc(table);
        return CollectionUtils.isNotEmpty(getManager(NDataflowManager.class, project).getModelsUsingTable(tableDesc));
    }

    public List<NDataModel> getModelsUsingTable(String table, String project) {
        return getManager(NDataflowManager.class, project)
                .getModelsUsingTable(getManager(NTableMetadataManager.class, project).getTableDesc(table));
    }

    @VisibleForTesting
    public void checkFlatTableSql(NDataModel model) {
        if (skipCheckFlatTable(model)) {
            return;
        }

        long rangePartitionTableCount = model.getAllTableRefs().stream()
                .filter(p -> p.getTableDesc().isRangePartition()).count();
        if (rangePartitionTableCount > 0) {
            logger.info("Range partitioned tables do not support pushdown, so do not need to perform subsequent logic");
            return;
        }

        try (QueryContext queryContext = QueryContext.current()) {
            String project = model.getProject();
            ProjectInstance prjInstance = getManager(NProjectManager.class).getProject(project);
            queryContext.setAclInfo(AclPermissionUtil.createAclInfo(project, getCurrentUserGroups()));
            if (prjInstance.getSourceType() == ISourceAware.ID_SPARK
                    && model.getModelType() == NDataModel.ModelType.BATCH) {
                SparkSession ss = SparderEnv.getSparkSession();
                String flatTableSql = PushDownUtil.generateFlatTableSql(model, false);
                QueryParams queryParams = new QueryParams(project, flatTableSql, "default", false);
                queryParams.setKylinConfig(prjInstance.getConfig());
                queryParams.setAclInfo(AclPermissionUtil.createAclInfo(project, getCurrentUserGroups()));
                ss.sql(PushDownUtil.massagePushDownSql(queryParams));
            }
        } catch (Exception e) {
            buildExceptionMessage(model, e);
        }
    }

    private boolean skipCheckFlatTable(NDataModel model) {
        if (KylinConfig.getInstanceFromEnv().isUTEnv()) {
            return true;
        }
        IndexPlan indexPlan = getIndexPlan(model.getId(), model.getProject());
        KylinConfig config = indexPlan == null || indexPlan.getConfig() == null
                ? NProjectManager.getProjectConfig(model.getProject())
                : indexPlan.getConfig();
        return config.skipCheckFlatTable();
    }

    private void buildExceptionMessage(NDataModel dataModel, Exception e) {
        Pattern pattern = Pattern.compile("cannot resolve '(.*?)' given input columns");
        Matcher matcher = pattern.matcher(e.getMessage().replace("`", ""));
        if (matcher.find()) {
            String column = matcher.group(1);
            String table = column.contains(".") ? column.split("\\.")[0] : dataModel.getRootFactTableName();
            String error = String.format(Locale.ROOT, MsgPicker.getMsg().saveModelFail(), dataModel.getAlias(), column,
                    table);
            throw new KylinException(TABLE_NOT_EXIST, error);
        } else {
            String errorMsg = String.format(Locale.ROOT, "model [%s], %s", dataModel.getAlias(),
                    String.format(Locale.ROOT, MsgPicker.getMsg().getDefaultReason(),
                            null != e.getMessage() ? e.getMessage() : "null"));
            throw new KylinException(FAILED_EXECUTE_MODEL_SQL, errorMsg);
        }
    }

    private void validatePartitionDateColumn(ModelRequest modelRequest) {
        if (Objects.nonNull(modelRequest.getPartitionDesc())) {
            if (StringUtils.isNotEmpty(modelRequest.getPartitionDesc().getPartitionDateColumn())) {
                Preconditions.checkArgument(
                        StringUtils.isNotEmpty(modelRequest.getPartitionDesc().getPartitionDateFormat()),
                        "Partition column format can not be empty!");
            } else {
                modelRequest.getPartitionDesc().setPartitionDateFormat("");
            }
            validateFusionModelDimension(modelRequest);
        }
    }

    public void validateFusionModelDimension(ModelRequest modelRequest) {
        val rootFactTableName = modelRequest.getRootFactTableName();
        // fusion model check timestamp
        val modelType = modelRequest.getModelType();
        if (!StringUtils.isEmpty(rootFactTableName)
                && (modelType != NDataModel.ModelType.BATCH && modelType != NDataModel.ModelType.STREAMING)) {
            val mgr = NTableMetadataManager.getInstance(KylinConfig.getInstanceFromEnv(), modelRequest.getProject());
            val tableDesc = mgr.getTableDesc(rootFactTableName);
            if (tableDesc != null && tableDesc.isKafkaTable() && tableDesc.getKafkaConfig().hasBatchTable()) {
                val fullColumnName = modelRequest.getPartitionDesc().getPartitionDateColumn();
                val columnName = fullColumnName.substring(fullColumnName.indexOf(".") + 1);
                val hasPartitionColumn = modelRequest.getSimplifiedDimensions().stream()
                        .anyMatch(column -> column.getName().equalsIgnoreCase(columnName));
                if (!hasPartitionColumn && !modelRequest.getDimensionNameIdMap().containsKey(fullColumnName)) {
                    throw new KylinException(TIMESTAMP_COLUMN_NOT_EXIST,
                            MsgPicker.getMsg().getTimestampPartitionColumnNotExist());
                }
            }
        }
    }

    public void batchCreateModel(String project, List<ModelRequest> newModels, List<ModelRequest> reusedModels) {
        checkNewModels(project, newModels);
        EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
            saveNewModelsAndIndexes(project, newModels);
            updateReusedModelsAndIndexPlans(project, reusedModels);
            return null;
        }, project);
    }

    public void updateRecommendationsCount(String project, String modelId, int size) {
        EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
            NDataModelManager mgr = NDataModelManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
            NDataModel dataModel = mgr.getDataModelDesc(modelId);
            if (dataModel != null && !dataModel.isBroken() && dataModel.getRecommendationsCount() != size) {
                mgr.updateDataModel(modelId, copyForWrite -> copyForWrite.setRecommendationsCount(size));
            }
            return null;
        }, project);
    }

    public void mergeMetadataForSamplingOrSnapshot(String project, MergerInfo mergerInfo) {
        EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
            MetadataMerger merger = MetadataMerger.createMetadataMerger(project, mergerInfo.getHandlerType());

            List<MergerInfo.TaskMergeInfo> infoList = mergerInfo.getTaskMergeInfoList();
            Preconditions.checkArgument(infoList.size() == 1);

            merger.merge(infoList.get(0));
            return null;
        }, project);
    }

    public void mergeMetadataForLoadingInternalTable(String project, MergerInfo mergerInfo) {
        EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
            MetadataMerger merger = MetadataMerger.createMetadataMerger(project, mergerInfo.getHandlerType());

            List<MergerInfo.TaskMergeInfo> infoList = mergerInfo.getTaskMergeInfoList();
            Preconditions.checkArgument(infoList.size() == 1);

            merger.merge(infoList.get(0));
            return null;
        }, project);
    }

    public List<NDataLayout[]> mergeMetadata(String project, MergerInfo mergerInfo) {
        return EnhancedUnitOfWork
                .doInTransactionWithCheckAndRetry(UnitOfWorkParams.<List<NDataLayout[]>> builder().processor(() -> {
                    MetadataMerger merger = MetadataMerger.createMetadataMerger(project, mergerInfo.getHandlerType());

                    List<NDataLayout[]> mergedLayouts = new ArrayList<>();
                    mergerInfo.getTaskMergeInfoList().forEach(info -> mergedLayouts.add(merger.merge(info)));

                    if (mergerInfo.getHandlerType() == HandlerType.ADD_CUBOID) {
                        tryRemoveToBeDeletedLayouts(project, mergerInfo);
                    }
                    markDFStatus(project, mergerInfo.getModelId(), mergerInfo.getHandlerType(),
                            mergerInfo.getErrorOrPausedJobCount());
                    return mergedLayouts;
                }).retryMoreTimeForDeadLockException(true).unitName(project).build());
    }

    private void tryRemoveToBeDeletedLayouts(String project, MergerInfo mergerInfo) {
        AbstractExecutable executable = ExecutableManager.getInstance(getConfig(), project)
                .getJob(mergerInfo.getJobId());
        if (!(executable instanceof NSparkCubingJob)) {
            return;
        }
        NSparkCubingJob job = (NSparkCubingJob) executable;
        if (job.getSparkCubingStep().getStatus() != ExecutableState.SUCCEED) {
            return;
        }
        boolean layoutsDeletableAfterBuild = Boolean
                .parseBoolean(job.getParam(NBatchConstants.P_LAYOUTS_DELETABLE_AFTER_BUILD));
        if (!layoutsDeletableAfterBuild) {
            return;
        }

        // Notice: The following df & indexPlan have been updated in transaction
        NDataflow df = NDataflowManager.getInstance(getConfig(), project).getDataflow(job.getTargetModelId());
        IndexPlan indexPlan = NIndexPlanManager.getInstance(getConfig(), project).getIndexPlan(job.getTargetModelId());
        Set<Long> toBeDeletedLayoutIds = indexPlan.getAllToBeDeleteLayoutId();

        if (!toBeDeletedLayoutIds.isEmpty()) {
            Set<Long> processedLayouts = mergerInfo.getTaskMergeInfoList().stream()
                    .flatMap(taskMergeInfo -> taskMergeInfo.getLayoutIds().stream())
                    .collect(Collectors.toCollection(LinkedHashSet::new));
            List<NDataSegment> targetSegments = df.getSegments(Sets.newHashSet(job.getTargetSegments()));

            // Almost the final layouts which will be deleted for sure
            Set<Long> prunedToBeDeletedLayoutIds = IndexBuildJobUtil
                    .pruneTBDelLayouts(
                            toBeDeletedLayoutIds.stream().map(indexPlan::getLayoutEntity)
                                    .collect(Collectors.toCollection(LinkedHashSet::new)),
                            processedLayouts.stream().map(indexPlan::getLayoutEntity).collect(
                                    Collectors.toCollection(LinkedHashSet::new)),
                            df, indexPlan, targetSegments)
                    .stream().map(LayoutEntity::getId).collect(Collectors.toSet());

            log.info("The final toBeDeletedLayouts: {}", prunedToBeDeletedLayoutIds);
            if (!prunedToBeDeletedLayoutIds.isEmpty()) {
                updateIndex(project, -1, mergerInfo.getModelId(), prunedToBeDeletedLayoutIds, true, true);
            }
        }
    }

    @Transaction(project = 0)
    public void makeSegmentReady(String project, String modelId, String segmentId, int errorOrPausedJobCount) {
        val kylinConfig = KylinConfig.getInstanceFromEnv();

        NDataflowManager dfMgr = NDataflowManager.getInstance(kylinConfig, project);
        NDataflow df = dfMgr.getDataflow(modelId);

        //update target seg's status
        val dfUpdate = new NDataflowUpdate(modelId);
        val seg = df.copy().getSegment(segmentId);
        seg.setStatus(SegmentStatusEnum.READY);
        dfUpdate.setToUpdateSegs(seg);
        dfMgr.updateDataflow(dfUpdate);
        markDFStatus(project, modelId, HandlerType.ADD_SEGMENT, errorOrPausedJobCount);
    }

    public void markDFStatus(String project, String modelId, HandlerType handlerType, int errorOrPausedJobCount) {
        NDataflowManager dfManager = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
        NDataflow df = dfManager.getDataflow(modelId);
        boolean isOffline = dfManager.isOfflineModel(df);
        RealizationStatusEnum status = df.getStatus();
        if (RealizationStatusEnum.ONLINE == status && isOffline) {
            dfManager.updateDataflowStatus(df.getId(), RealizationStatusEnum.OFFLINE);
        } else if (RealizationStatusEnum.OFFLINE == status && !isOffline) {
            updateDataflowStatus(project, df.getId(), RealizationStatusEnum.ONLINE);
        }
    }

    public void checkAndAutoMergeSegments(String project, String modelId, String owner) {
        try {
            SegmentAutoMergeUtil.autoMergeSegments(project, modelId, owner);
        } catch (Exception e) {
            log.error("Auto merge failed on project {} model {}", project, modelId, e);
        }
    }

    public void checkNewModels(String project, List<ModelRequest> newModels) {
        aclEvaluate.checkProjectWritePermission(project);
        checkDuplicateAliasInModelRequests(newModels);
        for (ModelRequest modelRequest : newModels) {
            validatePartitionDateColumn(modelRequest);
            modelRequest.setProject(project);
            doCheckBeforeModelSave(project, modelRequest);
        }
    }

    public void saveNewModelsAndIndexes(String project, List<ModelRequest> newModels) {
        saveNewModelsAndIndexes(project, null, newModels);
    }

    public void saveNewModelsAndIndexes(String project, String saveIndexesStrategy, List<ModelRequest> newModels) {
        if (CollectionUtils.isEmpty(newModels)) {
            return;
        }

        KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
        NDataModelManager dataModelManager = NDataModelManager.getInstance(kylinConfig, project);
        NIndexPlanManager indexPlanManager = NIndexPlanManager.getInstance(kylinConfig, project);
        NDataflowManager dataflowManager = NDataflowManager.getInstance(kylinConfig, project);
        for (ModelRequest modelRequest : newModels) {
            if (modelRequest.getIndexPlan() == null) {
                continue;
            }
            // create model
            NDataModel model = JsonUtil.deepCopyQuietly(modelRequest, NDataModel.class);
            model.setProject(project);
            model.setComputedColumnDescs(modelRequest.getComputedColumnDescs());
            IndexPlan indexPlan = modelRequest.getIndexPlan();
            indexPlan.setProject(project);

            NDataModel saved;
            if (dataModelManager.getDataModelDesc(model.getUuid()) != null) {
                saved = dataModelManager.updateDataModelDesc(model);
            } else {
                saved = dataModelManager.createDataModelDesc(model, model.getOwner());
            }

            // expand measures
            semanticUpdater.deleteExpandableMeasureInternalMeasures(saved);
            semanticUpdater.expandExpandableMeasure(saved);
            preProcessBeforeModelSave(saved, project);
            NDataModel expanded = getManager(NDataModelManager.class, project).updateDataModelDesc(saved);

            // create IndexPlan
            IndexPlan emptyIndex = new IndexPlan();
            emptyIndex.setUuid(expanded.getUuid());
            indexPlanManager.createIndexPlan(emptyIndex);
            indexPlanService.expandIndexPlanRequest(indexPlan, expanded);
            if (SAVE_INDEXES_STRATEGY.equalsIgnoreCase(saveIndexesStrategy)) {
                indexPlan.setBaseAggIndexReduceHighCardinalityDim(true);
            }
            addBaseIndex(modelRequest, expanded, indexPlan);

            // create DataFlow
            val df = dataflowManager.createDataflow(emptyIndex, expanded.getOwner());
            if (modelRequest.isWithEmptySegment() && !modelRequest.isStreaming()) {
                dataflowManager.appendSegment(df, SegmentRange.TimePartitionedSegmentRange.createInfinite(),
                        SegmentStatusEnum.READY);
            }
            if (modelRequest.isWithModelOnline()) {
                dataflowManager.updateDataflowStatus(df.getId(), RealizationStatusEnum.ONLINE);
            }

            createStreamingJob(project, expanded, modelRequest);
            updateIndexPlan(project, indexPlan, expanded, saveIndexesStrategy);
            UnitOfWorkContext context = UnitOfWork.get();
            context.doAfterUnit(() -> EventBusFactory.getInstance()
                    .postSync(new ModelAddEvent(project, expanded.getId(), expanded.getAlias())));
        }
    }

    private void updateReusedModelsAndIndexPlans(String project, List<ModelRequest> modelRequestList) {
        if (CollectionUtils.isEmpty(modelRequestList)) {
            return;
        }

        if (modelRequestList.stream()
                .anyMatch(modelRequest -> !FusionIndexService.checkUpdateIndexEnabled(project, modelRequest.getId()))) {
            throw new KylinException(STREAMING_INDEX_UPDATE_DISABLE, MsgPicker.getMsg().getStreamingIndexesConvert());
        }

        for (ModelRequest modelRequest : modelRequestList) {
            modelRequest.setProject(project);
            semanticUpdater.expandModelRequest(modelRequest);

            KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();
            NDataModelManager modelManager = NDataModelManager.getInstance(kylinConfig, project);
            NIndexPlanManager indexPlanManager = NIndexPlanManager.getInstance(kylinConfig, project);

            Map<String, NDataModel.NamedColumn> columnMap = Maps.newHashMap();
            modelRequest.getAllNamedColumns().forEach(column -> {
                Preconditions.checkArgument(!columnMap.containsKey(column.getAliasDotColumn()));
                columnMap.put(column.getAliasDotColumn(), column);
            });

            BaseIndexUpdateHelper baseIndexUpdater = new BaseIndexUpdateHelper(
                    modelManager.getDataModelDesc(modelRequest.getId()), false);
            // update model
            List<LayoutRecDetailResponse> recItems = modelRequest.getRecItems();
            NDataModel updated = modelManager.updateDataModel(modelRequest.getId(), copyForWrite -> {
                copyForWrite.setJoinTables(modelRequest.getJoinTables());
                List<NDataModel.NamedColumn> allNamedColumns = copyForWrite.getAllNamedColumns();
                Map<Integer, NDataModel.NamedColumn> namedColumnMap = Maps.newHashMap();
                allNamedColumns.forEach(col -> namedColumnMap.put(col.getId(), col));
                Map<String, ComputedColumnDesc> newCCMap = Maps.newLinkedHashMap();
                Map<String, NDataModel.NamedColumn> newDimMap = Maps.newLinkedHashMap();
                Map<String, NDataModel.Measure> newMeasureMap = Maps.newLinkedHashMap();
                recItems.forEach(recItem -> {
                    recItem.getComputedColumns().stream() //
                            .filter(LayoutRecDetailResponse.RecComputedColumn::isNew) //
                            .forEach(recCC -> {
                                ComputedColumnDesc cc = recCC.getCc();
                                newCCMap.putIfAbsent(cc.getFullName(), cc);
                            });
                    recItem.getDimensions().stream() //
                            .filter(LayoutRecDetailResponse.RecDimension::isNew) //
                            .forEach(recDim -> {
                                NDataModel.NamedColumn dim = recDim.getDimension();
                                newDimMap.putIfAbsent(dim.getAliasDotColumn(), dim);
                            });
                    recItem.getMeasures().stream() //
                            .filter(LayoutRecDetailResponse.RecMeasure::isNew) //
                            .forEach(recMeasure -> {
                                NDataModel.Measure measure = recMeasure.getMeasure();
                                newMeasureMap.putIfAbsent(measure.getName(), measure);
                            });
                });
                newCCMap.forEach((ccName, cc) -> {
                    copyForWrite.getComputedColumnDescs().add(cc);
                    NDataModel.NamedColumn column = columnMap.get(cc.getFullName());
                    allNamedColumns.add(column);
                    namedColumnMap.putIfAbsent(column.getId(), column);
                });
                newDimMap.forEach((colName, dim) -> {
                    if (namedColumnMap.containsKey(dim.getId())) {
                        namedColumnMap.get(dim.getId()).setStatus(NDataModel.ColumnStatus.DIMENSION);
                    } else {
                        allNamedColumns.add(dim);
                    }
                });
                Set<Integer> namedColumnIds = allNamedColumns.stream().map(NDataModel.NamedColumn::getId)
                        .collect(Collectors.toSet());
                modelRequest.getAllNamedColumns().forEach(col -> {
                    if (!namedColumnIds.contains(col.getId())) {
                        allNamedColumns.add(col);
                        namedColumnIds.add(col.getId());
                    }
                });
                newMeasureMap.forEach((measureName, measure) -> copyForWrite.getAllMeasures().add(measure));
                // keep order of all columns and measures
                copyForWrite.keepColumnOrder();
                copyForWrite.keepMeasureOrder();
            });

            // expand
            semanticUpdater.deleteExpandableMeasureInternalMeasures(updated);
            semanticUpdater.expandExpandableMeasure(updated);
            preProcessBeforeModelSave(updated, project);
            NDataModel expanded = getManager(NDataModelManager.class, project).updateDataModelDesc(updated);

            // update IndexPlan
            IndexPlan indexPlan = modelRequest.getIndexPlan();
            indexPlanService.expandIndexPlanRequest(indexPlan, expanded);
            Map<Long, LayoutEntity> layoutMap = Maps.newHashMap();
            indexPlan.getAllLayouts().forEach(layout -> layoutMap.putIfAbsent(layout.getId(), layout));
            indexPlanManager.updateIndexPlan(modelRequest.getId(), copyForWrite -> {
                IndexPlan.IndexPlanUpdateHandler updateHandler = copyForWrite.createUpdateHandler();
                for (LayoutRecDetailResponse recItem : recItems) {
                    long layoutId = recItem.getIndexId();
                    LayoutEntity layout = layoutMap.get(layoutId);
                    updateHandler.add(layout, IndexEntity.isAggIndex(recItem.getIndexId()));
                }
                updateHandler.complete();
            });
            modelChangeSupporters.forEach(listener -> listener.onUpdateSingle(project, modelRequest.getUuid()));
            baseIndexUpdater.update(indexPlanService);
        }
    }

    public NDataModel createModel(String project, ModelRequest modelRequest) {
        checkNewModels(project, Lists.newArrayList(modelRequest));
        UnitOfWorkParams<NDataModel> params = UnitOfWorkParams.<NDataModel> builder().unitName(project)
                .processor(() -> {
                    NDataModel model = saveModel(project, modelRequest);
                    modelRequest.setUuid(model.getUuid());
                    updateExcludedCheckerResult(project, modelRequest);
                    return getManager(NDataModelManager.class, project).getDataModelDesc(model.getUuid());
                }).build();
        return EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(params);
    }

    private NDataModel doCheckBeforeModelSave(String project, ModelRequest modelRequest) {
        checkAliasExist(modelRequest.getUuid(), modelRequest.getAlias(), project);
        checkAliasIsExceededLimit(modelRequest.getAlias());
        modelRequest.setOwner(AclPermissionUtil.getCurrentUsername());
        modelRequest.setLastModified(modelRequest.getCreateTime());
        checkModelRequest(modelRequest);

        //remove some attributes in modelResponse to fit NDataModel
        val dataModel = semanticUpdater.convertToDataModel(modelRequest);
        if (ManagementType.TABLE_ORIENTED == dataModel.getManagementType()) {
            throw new KylinException(FAILED_CREATE_MODEL, MsgPicker.getMsg().getInvalidCreateModel());
        }

        preProcessBeforeModelSave(dataModel, project);
        checkFlatTableSql(dataModel);
        return dataModel;
    }

    private NDataModel saveModel(String project, ModelRequest modelRequest) {
        validatePartitionDateColumn(modelRequest);

        val dataModel = semanticUpdater.convertToDataModel(modelRequest);
        preProcessBeforeModelSave(dataModel, project);
        createStreamingJob(project, dataModel, modelRequest);
        var created = getManager(NDataModelManager.class, project).createDataModelDesc(dataModel, dataModel.getOwner());

        semanticUpdater.expandExpandableMeasure(created);
        preProcessBeforeModelSave(created, project);
        KylinConfig config = KylinConfig.getInstanceFromEnv();
        val model = getManager(NDataModelManager.class, project).updateDataModelDesc(created);

        val indexPlanManager = NIndexPlanManager.getInstance(config, model.getProject());
        val dataflowManager = NDataflowManager.getInstance(config, model.getProject());
        val indexPlan = new IndexPlan();
        indexPlan.setUuid(model.getUuid());
        indexPlan.setLastModified(System.currentTimeMillis());
        addBaseIndex(modelRequest, model, indexPlan);
        indexPlanManager.createIndexPlan(indexPlan);
        val df = dataflowManager.createDataflow(indexPlan, model.getOwner(), RealizationStatusEnum.OFFLINE);
        SegmentRange range = null;
        if (PartitionDesc.isEmptyPartitionDesc(model.getPartitionDesc())) {
            range = SegmentRange.TimePartitionedSegmentRange.createInfinite();
        } else if (StringUtils.isNotEmpty(modelRequest.getStart()) && StringUtils.isNotEmpty(modelRequest.getEnd())) {
            range = getSegmentRangeByModel(project, model.getUuid(), modelRequest.getStart(), modelRequest.getEnd());
        }
        if (range != null) {
            dataflowManager.fillDfManually(df, Lists.newArrayList(range));
        }
        UnitOfWorkContext context = UnitOfWork.get();
        context.doAfterUnit(() -> EventBusFactory.getInstance()
                .postSync(new ModelAddEvent(project, model.getId(), model.getAlias())));
        return getManager(NDataModelManager.class, project).getDataModelDesc(model.getUuid());
    }

    public void addBaseIndex(ModelRequest modelRequest, NDataModel model, IndexPlan indexPlan) {
        if (NDataModel.ModelType.BATCH == model.getModelType()) {
            List<IndexEntity.Source> sources = needHandleBaseIndexType(modelRequest);
            indexPlan.createAndAddBaseIndex(model, sources);
        }
    }

    private List<IndexEntity.Source> needHandleBaseIndexType(ModelRequest modelRequest) {
        List<IndexEntity.Source> sources = Lists.newArrayList();
        Set<IndexEntity.Source> requestSource = modelRequest.getBaseIndexType();
        if (requestSource != null) {
            if (requestSource.contains(IndexEntity.Source.BASE_AGG_INDEX)) {
                sources.add(IndexEntity.Source.BASE_AGG_INDEX);
            }
            if (requestSource.contains(IndexEntity.Source.BASE_TABLE_INDEX)) {
                sources.add(IndexEntity.Source.BASE_TABLE_INDEX);
            }
        } else if (modelRequest.isWithBaseIndex()) {
            sources.add(IndexEntity.Source.BASE_AGG_INDEX);
            sources.add(IndexEntity.Source.BASE_TABLE_INDEX);
        }
        return sources;
    }

    // for streaming & fusion model
    private void createStreamingJob(String project, NDataModel model, ModelRequest request) {
        if (NDataModel.ModelType.BATCH != model.getModelType()) {
            val jobManager = StreamingJobManager.getInstance(KylinConfig.getInstanceFromEnv(), model.getProject());
            jobManager.createStreamingJob(model);
            createBatchModelInFusion(project, model, request);
        }
    }

    // only for fusion model
    // create batch side model
    private void createBatchModelInFusion(String project, NDataModel model, ModelRequest request) {
        KafkaConfig kafkaConfig = model.getRootFactTableRef().getTableDesc().getKafkaConfig();
        if (kafkaConfig.hasBatchTable()) {
            String tableName = kafkaConfig.getBatchTable();

            ModelRequest copy = JsonUtil.deepCopyQuietly(request, ModelRequest.class);
            copy.setAlias(FusionModel.getBatchName(request.getAlias(), model.getUuid()));
            copy.setRootFactTableName(tableName);
            copy.setFusionId(model.getUuid());
            copy.setProject(project);
            model.setFusionId(model.getUuid());

            String tableAlias = kafkaConfig.getBatchTableAlias();
            String oldAliasName = model.getRootFactTableRef().getTableName();
            convertToDataModelResponse(copy, tableAlias, oldAliasName);
            NDataModel copyModel = saveModel(project, copy);
            createFusionModel(project, model, copyModel);
        }
    }

    private void createFusionModel(String project, NDataModel model, NDataModel copyModel) {
        FusionModel fusionModel = new FusionModel(model, copyModel);
        FusionModelManager fusionModelManager = FusionModelManager.getInstance(KylinConfig.getInstanceFromEnv(),
                project);
        fusionModelManager.createModel(fusionModel);
    }

    private void convertToDataModelResponse(ModelRequest copy, String tableName, String oldAliasName) {
        copy.getSimplifiedJoinTableDescs()
                .forEach(x -> x.getSimplifiedJoinDesc().changeFKTableAlias(oldAliasName, tableName));
        copy.getSimplifiedDimensions().forEach(x -> x.changeTableAlias(oldAliasName, tableName));
        copy.getSimplifiedMeasures().forEach(x -> x.changeTableAlias(oldAliasName, tableName));
        copy.getPartitionDesc().changeTableAlias(oldAliasName, tableName);
    }

    void updateIndexPlan(String project, IndexPlan indexPlan, NDataModel model, String saveIndexesStrategy) {
        NIndexPlanManager indexPlanManager = NIndexPlanManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
        indexPlanManager.updateIndexPlan(indexPlan.getId(), copyForWrite -> {
            if (SAVE_INDEXES_STRATEGY.equalsIgnoreCase(saveIndexesStrategy)) {
                copyForWrite.setBaseAggIndexReduceHighCardinalityDim(true);
                splitIndexesIntoSingleDimIndexes(model, indexPlan);
            }
            if (indexPlan.getAggShardByColumns() != null) {
                copyForWrite.setAggShardByColumns(indexPlan.getAggShardByColumns());
            }
            if (CollectionUtils.isNotEmpty(indexPlan.getIndexes())) {
                copyForWrite.setIndexes(indexPlan.getIndexes());
            }
            copyForWrite.setEngineType(indexPlan.getEngineType());
            copyForWrite.setIndexPlanOverrideIndexes(indexPlan.getIndexPlanOverrideIndexes());
            copyForWrite.setLastModified(System.currentTimeMillis());
        });
    }

    private void checkModelRequest(ModelRequest request) {
        checkModelOwner(request);
        checkModelDimensions(request);
        checkModelMeasures(request);
        checkModelJoinConditions(request);
    }

    private void checkModelOwner(ModelRequest request) {
        if (StringUtils.isBlank(request.getOwner())) {
            throw new KylinException(INVALID_PARAMETER, "Invalid parameter, model owner is empty.");
        }
    }

    @VisibleForTesting
    public void checkModelDimensions(ModelRequest request) {
        Set<String> dimensionNames = new HashSet<>();

        KylinConfig kylinConfig = getManager(NProjectManager.class).getProject(request.getProject()).getConfig();
        int maxModelDimensionMeasureNameLength = kylinConfig.getMaxModelDimensionMeasureNameLength();

        for (NDataModel.NamedColumn dimension : request.getSimplifiedDimensions()) {
            dimension.setName(StringUtils.trim(dimension.getName()));
            // check if the dimension name is valid
            if (StringUtils.length(dimension.getName()) > maxModelDimensionMeasureNameLength
                    || !VALID_NAME_FOR_DIMENSION.matcher(dimension.getName()).matches())
                throw new KylinException(INVALID_NAME,
                        String.format(Locale.ROOT, MsgPicker.getMsg().getInvalidDimensionName(), dimension.getName(),
                                maxModelDimensionMeasureNameLength));

            // check duplicate dimension names
            if (dimensionNames.contains(dimension.getName()))
                throw new KylinException(DUPLICATE_DIMENSION_NAME, String.format(Locale.ROOT,
                        MsgPicker.getMsg().getDuplicateDimensionName(), dimension.getName()));

            dimensionNames.add(dimension.getName());
        }
    }

    private void checkMeasureNameValid(ModelRequest request) {
        int maxLen = NProjectManager.getProjectConfig(request.getProject()).getMaxModelDimensionMeasureNameLength();
        String invalidPattern = MsgPicker.getMsg().getInvalidMeasureName();
        for (SimplifiedMeasure measure : request.getSimplifiedMeasures()) {
            String name = measure.getName();
            if (StringUtils.length(name) <= maxLen && checkIsValidMeasureName(name)) {
                continue;
            }
            throw new KylinException(INVALID_NAME, String.format(Locale.ROOT, invalidPattern, name, maxLen));
        }
    }

    private void checkMeasureNameDuplicate(ModelRequest modelRequest) {
        Set<String> measureNames = Sets.newHashSet();
        for (SimplifiedMeasure measure : modelRequest.getSimplifiedMeasures()) {
            if (measureNames.contains(measure.getName())) {
                throw new KylinException(DUPLICATE_MEASURE_NAME,
                        String.format(Locale.ROOT, MsgPicker.getMsg().getDuplicateMeasureName(), measure.getName()));
            } else {
                measureNames.add(measure.getName());
            }
        }
    }

    @VisibleForTesting
    public void checkModelMeasures(ModelRequest request) {
        Set<SimplifiedMeasure> measures = new HashSet<>();
        request.getSimplifiedMeasures().forEach(measure -> measure.setName(StringUtils.trim(measure.getName())));
        checkMeasureNameValid(request);
        checkMeasureNameDuplicate(request);

        for (SimplifiedMeasure measure : request.getSimplifiedMeasures()) {
            // check duplicate measure definitions
            SimplifiedMeasure dupMeasure = null;
            for (SimplifiedMeasure m : measures) {
                if (isDupMeasure(measure, m)) {
                    dupMeasure = m;
                    break;
                }
            }

            if (dupMeasure != null) {
                dupMeasure = dupMeasure.getId() != 0 ? dupMeasure : measure;
                if (request.getId() != null) {
                    NDataModel.Measure existingMeasure = getModelById(request.getId(), request.getProject())
                            .getEffectiveMeasures().get(dupMeasure.getId());
                    if (existingMeasure != null && existingMeasure.getType() == NDataModel.MeasureType.INTERNAL) {
                        throw new KylinException(DUPLICATE_MEASURE_EXPRESSION, String.format(Locale.ROOT,
                                MsgPicker.getMsg().getDuplicateInternalMeasureDefinition(), measure.getName()));
                    }
                }
                throw new KylinException(DUPLICATE_MEASURE_EXPRESSION, String.format(Locale.ROOT,
                        MsgPicker.getMsg().getDuplicateMeasureDefinition(), measure.getName()));
            }

            measures.add(measure);
        }
    }

    private boolean checkIsValidMeasureName(String measureName) {
        if (!KylinConfig.getInstanceFromEnv().isMeasureNameCheckEnabled()) {
            return true;
        }
        return VALID_NAME_FOR_MEASURE.matcher(measureName).matches();
    }

    private boolean isDupMeasure(SimplifiedMeasure measure, SimplifiedMeasure measure1) {
        if (measure.getExpression().equalsIgnoreCase(measure1.getExpression())
                && Objects.equals(measure.getParameterValue(), measure1.getParameterValue())
                && Objects.equals(measure.getConfiguration(), measure1.getConfiguration())) {

            // if both has a return type then compare the return type
            if (!Strings.isNullOrEmpty(measure1.getReturnType()) && !Strings.isNullOrEmpty(measure.getReturnType())) {
                return Objects.equals(measure1.getReturnType(), measure.getReturnType());
            } else {
                // otherwise if return type is null on any side, then just compare expr and params
                // one measure may be a newly added one and has not been assigned a return type yet
                return true;
            }
        }
        return false;
    }

    private void checkModelJoinConditions(ModelRequest request) {
        for (JoinTableDesc joinTableDesc : request.getJoinTables()) {
            Set<Pair<String, String>> joinKeys = new HashSet<>();
            JoinDesc joinDesc = joinTableDesc.getJoin();
            int size = joinDesc.getPrimaryKey().length;
            String[] primaryKeys = joinDesc.getPrimaryKey();
            String[] foreignKey = joinDesc.getForeignKey();

            for (int i = 0; i < size; i++) {
                if (joinKeys.contains(Pair.newPair(primaryKeys[i], foreignKey[i])))
                    throw new KylinException(DUPLICATE_JOIN_CONDITION, String.format(Locale.ROOT,
                            MsgPicker.getMsg().getDuplicateJoinConditions(), primaryKeys[i], foreignKey[i]));

                joinKeys.add(Pair.newPair(primaryKeys[i], foreignKey[i]));
            }
        }
    }

    public String probeDateFormatIfNotExist(String project, NDataModel model) throws SQLException {
        PartitionDesc partitionDesc = model.getPartitionDesc();
        if (PartitionDesc.isEmptyPartitionDesc(partitionDesc)
                || StringUtils.isNotEmpty(partitionDesc.getPartitionDateFormat()))
            return "";

        if (StringUtils.isNotEmpty(partitionDesc.getPartitionDateColumn())
                && StringUtils.isNotEmpty(partitionDesc.getPartitionDateFormat())) {
            return partitionDesc.getPartitionDateColumn();
        }

        String partitionColumn = model.getPartitionDesc().getPartitionDateColumnRef().getBackTickExp();
        String date = PushDownUtil.probeColFormat(model.getRootFactTableName(), partitionColumn, project);
        return DateFormat.proposeDateFormat(date);
    }

    public void saveDateFormatIfNotExist(String project, String modelId, String format) {
        if (StringUtils.isEmpty(format)) {
            return;
        }
        getManager(NDataModelManager.class, project).updateDataModel(modelId,
                model -> model.getPartitionDesc().setPartitionDateFormat(format));

    }

    public NDataSegment appendSegment(AddSegmentRequest request) {
        String project = request.getProject();
        return EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
            val df = getManager(NDataflowManager.class, project).getDataflow(request.getModelId());
            return getManager(NDataflowManager.class, project).appendSegment(df, request.getSegRange(),
                    request.getStatus(), request.getMultiPartitionValues());
        }, project);
    }

    private Pair<String, String> getPartitionColMinMaxValue(String project, String table, PartitionDesc desc)
            throws Exception {
        Preconditions.checkNotNull(desc);
        String partitionColumn = desc.getPartitionDateColumn();
        String dateFormat = desc.getPartitionDateFormat();
        Preconditions.checkArgument(StringUtils.isNotEmpty(dateFormat) && StringUtils.isNotEmpty(partitionColumn));

        val minAndMaxTime = PushDownUtil.probeMinMaxTsWithTimeout(partitionColumn, table, project);

        return new Pair<>(DateFormat.getFormattedDate(minAndMaxTime.getFirst(), dateFormat),
                DateFormat.getFormattedDate(minAndMaxTime.getSecond(), dateFormat));
    }

    public SegmentCheckResponse checkSegHoleExistIfNewRangeBuild(String project, String modelId, String start,
            String end, boolean isBuildAllIndexes, List<Long> batchIndexIds) {
        aclEvaluate.checkProjectOperationPermission(project);
        Preconditions.checkArgument(!PushDownUtil.needPushdown(start, end), "Load data must set start and end date");
        NDataModel dataModelDesc = getManager(NDataModelManager.class, project).getDataModelDesc(modelId);
        TableDesc table = getManager(NTableMetadataManager.class, project)
                .getTableDesc(dataModelDesc.getRootFactTableName());
        SegmentRange segmentRangeToBuild = SourceFactory.getSource(table).getSegmentRange(start, end);
        List<NDataSegment> segmentGaps = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), project)
                .checkHoleIfNewSegBuild(modelId, segmentRangeToBuild);
        List<NDataSegment> overlapSegments = checkSegmentToBuildOverlapsBuilt(project, dataModelDesc,
                segmentRangeToBuild, isBuildAllIndexes, batchIndexIds);
        val overlapSegmentResponses = overlapSegments.stream().map(
                segment -> new SegmentRangeResponse(segment.getTSRange().getStart(), segment.getTSRange().getEnd()))
                .collect(Collectors.toList());
        SegmentCheckResponse segmentCheckResponse = new SegmentCheckResponse();
        val segHoles = segmentGaps.stream()
                .map(seg -> new SegmentRangeResponse(seg.getTSRange().getStart(), seg.getTSRange().getEnd()))
                .collect(Collectors.toList());
        segmentCheckResponse.setSegmentHoles(segHoles);
        segmentCheckResponse.setOverlapSegments(overlapSegmentResponses);
        return segmentCheckResponse;
    }

    public SegmentCheckResponse checkSegHoleIfSegDeleted(String model, String project, String[] ids) {
        aclEvaluate.checkProjectOperationPermission(project);
        NDataModel dataModel = getManager(NDataModelManager.class, project).getDataModelDesc(model);
        if (ManagementType.TABLE_ORIENTED == dataModel.getManagementType()) {
            throw new KylinException(PERMISSION_DENIED,
                    String.format(Locale.ROOT, MsgPicker.getMsg().getModelSegmentCanNotRemove(), dataModel.getAlias()));
        }
        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        checkSegmentsExistById(model, project, ids);
        checkSegmentsStatus(model, project, ids, SegmentStatusEnumToDisplay.LOCKED);
        NDataflow dataflow = dataflowManager.getDataflow(model);
        val toDeletedSeg = dataflow.getSegments().stream().filter(seg -> Arrays.asList(ids).contains(seg.getId()))
                .collect(Collectors.toList());
        val remainSegs = dataflow.getSegments().stream().filter(seg -> !Arrays.asList(ids).contains(seg.getId()))
                .collect(Collectors.toList());
        val segHoles = dataflowManager.calculateHoles(model, remainSegs).stream()
                .filter(seg -> toDeletedSeg.stream()
                        .anyMatch(deletedSeg -> deletedSeg.getSegRange().overlaps(seg.getSegRange())))
                .map(seg -> new SegmentRangeResponse(seg.getTSRange().getStart(), seg.getTSRange().getEnd()))
                .collect(Collectors.toList());
        SegmentCheckResponse response = new SegmentCheckResponse();
        response.setSegmentHoles(segHoles);
        return response;
    }

    public JobInfoResponse fixSegmentHoles(String project, String modelId, List<SegmentTimeRequest> segmentHoles,
            Set<String> ignoredSnapshotTables) throws SQLException {
        aclEvaluate.checkProjectOperationPermission(project);
        NDataModel modelDesc = getManager(NDataModelManager.class, project).getDataModelDesc(modelId);
        checkModelAndIndexManually(project, modelId);
        String format = probeDateFormatIfNotExist(project, modelDesc);

        List<JobInfoResponse.JobInfo> jobIds = EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
            List<JobInfoResponse.JobInfo> jobInfos = Lists.newArrayList();
            List<String[]> allPartitions = null;
            if (modelDesc.isMultiPartitionModel()) {
                allPartitions = modelDesc.getMultiPartitionDesc().getPartitions().stream()
                        .map(MultiPartitionDesc.PartitionInfo::getValues).collect(Collectors.toList());
            }
            for (SegmentTimeRequest hole : segmentHoles) {
                jobInfos.add(modelBuildService.constructIncrementBuild(
                        new IncrementBuildSegmentParams(project, modelId, hole.getStart(), hole.getEnd(), format, true,
                                allPartitions).withIgnoredSnapshotTables(ignoredSnapshotTables)));
            }
            return jobInfos;
        }, project);

        JobInfoResponse jobInfoResponse = new JobInfoResponse();
        jobInfoResponse.setJobs(jobIds);
        return jobInfoResponse;
    }

    @Transaction(project = 0)
    public JobInfoResponse optimizeLayoutData(String project, String modelId, OptimizeLayoutDataRequest request)
            throws Exception {
        aclEvaluate.checkProjectWritePermission(project);
        checkModelPermission(project, modelId);

        Set<Long> targetLayout = updateOptimizeSettings(project, modelId, request);

        JobParam jobParam = new JobParam(modelId, BasicService.getUsername()).withProject(project)
                .withJobTypeEnum(JobTypeEnum.LAYOUT_DATA_OPTIMIZE).withPriority(request.getPriority())
                .withYarnQueue(request.getYarnQueue()).withTargetLayouts(targetLayout);

        String jobId = JobManager.getInstance(getConfig(), project).addJob(jobParam);
        JobInfoResponse.JobInfo jobInfo = new JobInfoResponse.JobInfo(JobTypeEnum.LAYOUT_DATA_OPTIMIZE.toString(),
                jobId);
        JobInfoResponse jobInfoResponse = new JobInfoResponse();
        jobInfoResponse.setJobs(Lists.newArrayList(jobInfo));
        return jobInfoResponse;
    }

    public Set<Long> updateOptimizeSettings(String project, String modelId, OptimizeLayoutDataRequest request) {
        Set<Long> targetLayout = Sets.newHashSet();

        Set<Long> toOptimizeModelLayouts = updateModelOptimizeSettings(project, modelId,
                request.getModelOptimizationSetting());
        targetLayout.addAll(toOptimizeModelLayouts);

        Set<Long> toOptimizeLayouts = updateLayoutOptimizeSettings(project, modelId,
                request.getLayoutDataOptimizationSettingList());
        targetLayout.addAll(toOptimizeLayouts);
        return targetLayout;
    }

    private Set<Long> updateModelOptimizeSettings(String project, String modelId,
            OptimizeLayoutDataRequest.DataOptimizationSetting modelSettings) {
        NIndexPlanManager indexPlanManager = NIndexPlanManager.getInstance(getConfig(), project);
        HashSet<Long> toOptimizeLayouts = Sets.newHashSet();
        AtomicReference<Boolean> modelConfigChange = new AtomicReference<>(false);
        if (modelSettings != null) {
            indexPlanManager.updateIndexPlan(modelId, indexPlan -> {
                LinkedHashMap<String, String> oldProps = indexPlan.getOverrideProps();
                List<String> partitionByCols = modelSettings.getRepartitionByColumns();
                List<String> zorderByCols = modelSettings.getZorderByColumns();
                long maxFileSize = modelSettings.getMaxCompactionFileSize();
                long minFileSize = modelSettings.getMinCompactionFileSize();
                if (partitionByCols != null) {
                    oldProps.put(IndexPlan.STORAGE_V3_MODEL_DEFAULT_PARTITION_BY_CONF_KEY,
                            String.join(IndexPlan.STORAGE_V3_CONFIG_COLUMN_SEPARATOR, partitionByCols));
                    modelConfigChange.set(true);
                }
                if (zorderByCols != null) {
                    oldProps.put(IndexPlan.STORAGE_V3_MODEL_DEFAULT_ZORDER_BY_CONF_KEY,
                            String.join(IndexPlan.STORAGE_V3_CONFIG_COLUMN_SEPARATOR, zorderByCols));
                    modelConfigChange.set(true);
                }
                if (maxFileSize > 0) {
                    oldProps.put(IndexPlan.STORAGE_V3_MODEL_DEFAULT_MAX_FILE_SIZE_CONF_KEY, Long.toString(maxFileSize));
                    modelConfigChange.set(true);
                }
                if (minFileSize > 0) {
                    oldProps.put(IndexPlan.STORAGE_V3_MODEL_DEFAULT_MIN_FILE_SIZE_CONF_KEY, Long.toString(minFileSize));
                    modelConfigChange.set(true);
                }
            });
        }

        if (modelConfigChange.get()) {
            toOptimizeLayouts.addAll(indexPlanManager.getIndexPlan(modelId).getAllLayoutIds(false));
        }

        return toOptimizeLayouts;
    }

    private Set<Long> updateLayoutOptimizeSettings(String project, String modelId,
            List<OptimizeLayoutDataRequest.LayoutDataOptimizationSetting> layoutSettings) {
        NDataLayoutDetailsManager layoutDetailsManager = NDataLayoutDetailsManager.getInstance(getConfig(), project);
        HashSet<Long> toOptimizeLayouts = Sets.newHashSet();
        if (layoutSettings == null) {
            return toOptimizeLayouts;
        }
        layoutSettings.forEach(optimizeRequest -> {
            optimizeRequest.getLayoutIdList().forEach(layoutId -> {
                if (optimizeRequest.getSetting() != null) {
                    updateLayoutOptimizeSettings(optimizeRequest.getSetting(), modelId, layoutId, layoutDetailsManager,
                            toOptimizeLayouts);
                }
            });
        });
        return toOptimizeLayouts;
    }

    private void updateLayoutOptimizeSettings(OptimizeLayoutDataRequest.DataOptimizationSetting layoutSetting,
                                              String modelId, Long layoutId,
                                              NDataLayoutDetailsManager layoutDetailsManager,
                                              HashSet<Long> toOptimizeLayouts) {
        layoutDetailsManager.updateLayoutDetails(modelId, layoutId, (copy) -> {
            if (layoutSetting.getMinCompactionFileSize() > 0) {
                copy.setMinCompactionFileSizeInBytes(layoutSetting.getMinCompactionFileSize());
            }
            if (layoutSetting.getMaxCompactionFileSize() > 0) {
                copy.setMaxCompactionFileSizeInBytes(layoutSetting.getMaxCompactionFileSize());
            }
            if (layoutSetting.getZorderByColumns() != null) {
                copy.setZorderByColumns(layoutSetting.getZorderByColumns());
            }
            if (layoutSetting.getRepartitionByColumns() != null) {
                copy.setPartitionColumns(layoutSetting.getRepartitionByColumns());
            }
            copy.setCompactionAfterUpdate(layoutSetting.isCompaction());
            toOptimizeLayouts.add(layoutId);
        });
    }

    @Transaction(project = 0)
    public void setStorageType(String project, String modelId, int storageType) {
        aclEvaluate.checkProjectWritePermission(project);
        checkModelPermission(project, modelId);
        NDataModelManager manager = NDataModelManager.getInstance(getConfig(), project);
        NDataflowManager dataflowManager = NDataflowManager.getInstance(getConfig(), project);
        NDataflow dataflow = dataflowManager.getDataflow(modelId);
        if (dataflow.getSegments().isEmpty()) {
            manager.updateDataModel(modelId, model -> model.setStorageType(storageType));
        } else {
            throw new KylinException(MODEL_STORAGE_UPDATE_FAILED, MsgPicker.getMsg().getModelStorageUpdateFailed());
        }
    }

    public NDataLayoutDetails getLayoutDetail(String project, String modelId, long layoutId) {
        aclEvaluate.checkProjectReadPermission(project);
        return NDataLayoutDetailsManager.getInstance(getConfig(), project).getNDataLayoutDetails(modelId, layoutId);
    }

    public void removeIndexesFromSegments(String project, String modelId, List<String> segmentIds,
            List<Long> indexIds) {
        aclEvaluate.checkProjectOperationPermission(project);
        checkModelPermission(project, modelId);
        EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
            val dfManger = getManager(NDataflowManager.class, project);
            NDataLayoutDetailsManager layoutDetailsManger = NDataLayoutDetailsManager.getInstance(getConfig(), project);
            NDataflow dataflow = dfManger.getDataflow(modelId);
            for (String segmentId : segmentIds) {
                NDataSegment seg = dataflow.getSegment(segmentId);
                dfManger.updateDataflowDetailsLayouts(seg, indexIds, Collections.emptyList());
                for (Long toRemoveIndex : indexIds) {
                    layoutDetailsManger.updateLayoutDetails(modelId, toRemoveIndex,
                            (layoutDetail) -> layoutDetail.getFragmentRangeSet().remove(seg.getRange()));
                }
            }
            getManager(NIndexPlanManager.class, project).updateIndexPlan(dataflow.getUuid(),
                    IndexPlan::removeTobeDeleteIndexIfNecessary);
            return null;
        }, project);
    }

    @Transaction(project = 0)
    public NDataSegment appendPartitions(String project, String dfId, String segId, List<String[]> partitionValues) {
        return NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), project).appendPartitions(dfId, segId,
                partitionValues);
    }

    @Transaction(project = 0)
    public NDataSegment mergeSegments(String project, MergeSegmentRequest mergeSegmentRequest) {
        NDataflow df = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), project)
                .getDataflow(mergeSegmentRequest.getIndexPlanUuid());
        return NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), project).mergeSegments(df,
                mergeSegmentRequest.getSegRange(), mergeSegmentRequest.isForce(), mergeSegmentRequest.getFileLayer(),
                mergeSegmentRequest.getNewSegId());
    }

    public ModelRequest convertToRequest(NDataModel modelDesc) throws IOException {
        val request = new ModelRequest(JsonUtil.deepCopy(modelDesc, NDataModel.class));
        request.setSimplifiedMeasures(modelDesc.getEffectiveMeasures().values().stream()
                .map(SimplifiedMeasure::fromMeasure).collect(Collectors.toList()));
        request.setSimplifiedDimensions(modelDesc.getAllNamedColumns().stream()
                .filter(NDataModel.NamedColumn::isDimension).collect(Collectors.toList()));
        request.setComputedColumnDescs(modelDesc.getComputedColumnDescs());

        return request;
    }

    @Transaction(project = 0)
    public void updatePartitionColumn(String project, String modelId, PartitionDesc partitionDesc,
            MultiPartitionDesc multiPartitionDesc) throws IOException {
        aclEvaluate.checkProjectWritePermission(project);
        checkModelPermission(project, modelId);
        val dataflowManager = getManager(NDataflowManager.class, project);
        val df = dataflowManager.getDataflow(modelId);
        val model = df.getModel();
        ModelUtils.checkPartitionColumn(model, partitionDesc, MsgPicker.getMsg().getPartitionColumnSaveError());

        if (PartitionDesc.isEmptyPartitionDesc(model.getPartitionDesc()) && df.getFirstSegment() == null
                && !model.isMultiPartitionModel()) {
            dataflowManager.fillDfManually(df,
                    Lists.newArrayList(SegmentRange.TimePartitionedSegmentRange.createInfinite()));
        }
        if (!Objects.equals(partitionDesc, model.getPartitionDesc())
                || !ModelSemanticHelper.isMultiPartitionDescSame(model.getMultiPartitionDesc(), multiPartitionDesc)) {
            val request = convertToRequest(model);
            request.setProject(project);
            request.setPartitionDesc(partitionDesc);
            request.setSaveOnly(true);
            request.setMultiPartitionDesc(multiPartitionDesc);

            updateDataModelSemantic(project, request);
        }
    }

    public List<MultiPartitionValueResponse> getMultiPartitionValues(String project, String modelId) {
        val model = getModelById(modelId, project);
        val multiPartitionDesc = model.getMultiPartitionDesc();
        val segments = getManager(NDataflowManager.class, project).getDataflow(modelId).getSegments();
        val totalSegCount = segments.size();
        val responses = Lists.<MultiPartitionValueResponse> newArrayList();

        if (multiPartitionDesc == null || CollectionUtils.isEmpty(multiPartitionDesc.getPartitions())) {
            return responses;
        }

        for (MultiPartitionDesc.PartitionInfo partition : multiPartitionDesc.getPartitions()) {
            val builtSegmentCount = (int) segments.stream()
                    .filter(segment -> segment.getMultiPartitionIds().contains(partition.getId())).count();
            responses.add(new MultiPartitionValueResponse(partition.getId(), partition.getValues(), builtSegmentCount,
                    totalSegCount));
        }

        return responses;
    }

    /**
     * batch update multiple partition values
     *
     * @param project
     * @param modelId
     * @param partitionValues
     * @return
     */
    @Transaction(project = 0)
    public NDataModel batchUpdateMultiPartition(String project, String modelId, List<String[]> partitionValues) {
        NDataModelManager modelManager = getManager(NDataModelManager.class, project);
        NDataModel dataModel = modelManager.getDataModelDesc(modelId);
        if (dataModel == null) {
            throw new KylinException(MODEL_ID_NOT_EXIST, modelId);
        }

        MultiPartitionDesc multiPartitionDesc = dataModel.getMultiPartitionDesc();

        Set<Long> tobeDeletedPartitions = multiPartitionDesc.getPartitions().stream()
                .filter(partitionInfo -> partitionValues.stream()
                        .noneMatch(pv -> Objects.deepEquals(pv, partitionInfo.getValues())))
                .map(MultiPartitionDesc.PartitionInfo::getId).collect(Collectors.toSet());

        if (!tobeDeletedPartitions.isEmpty()) {
            logger.debug("Import model {} delete partitions {}", dataModel.getAlias(), tobeDeletedPartitions);
            deletePartitions(dataModel.getProject(), null, dataModel.getUuid(), tobeDeletedPartitions);
        }

        dataModel = modelManager.getDataModelDesc(modelId);

        modelManager.addPartitionsIfAbsent(dataModel, partitionValues);

        return modelManager.getDataModelDesc(modelId);
    }

    @Transaction(project = 0)
    public void addMultiPartitionValues(String project, String modelId, List<String[]> subPartitionValues) {
        aclEvaluate.checkProjectOperationPermission(project);
        val modelManager = NDataModelManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
        val model = modelManager.getDataModelDesc(modelId);
        modelManager.addPartitionsIfAbsent(model, subPartitionValues);
    }

    private void checkModelAndIndexManually(String project, String modelId) {
        checkModelAndIndexManually(new FullBuildSegmentParams(project, modelId, true));
    }

    public void checkModelAndIndexManually(FullBuildSegmentParams params) {
        if (params.isNeedBuild()) {
            val indexPlan = getIndexPlan(params.getModelId(), params.getProject());
            if (indexPlan == null || indexPlan.getAllLayouts().isEmpty()) {
                throw new KylinException(PERMISSION_DENIED, MsgPicker.getMsg().getCanNotBuildSegment());
            }
        }
    }

    public List<NDataSegment> checkSegmentToBuildOverlapsBuilt(String project, NDataModel model,
            SegmentRange<Long> segmentRangeToBuild, boolean isBuildAllIndexes, List<Long> batchIndexIds) {
        boolean isOverlap;
        Segments<NDataSegment> segments = getSegmentsByRange(model.getId(), project, "0", "" + Long.MAX_VALUE);
        List<NDataSegment> overlapsBuiltSegment = Lists.newArrayListWithCapacity(segments.size());

        if (CollectionUtils.isEmpty(segments)) {
            return overlapsBuiltSegment;
        }

        boolean buildSegmentOverlapEnable = getIndexPlan(model.getId(), project).getConfig()
                .isBuildSegmentOverlapEnabled();
        boolean isBuildAllIndexesFinally = CollectionUtils.isEmpty(batchIndexIds)
                || batchIndexIds.size() == getIndexPlan(model.getId(), project).getAllIndexes().size();

        for (NDataSegment existedSegment : segments) {
            if (buildSegmentOverlapEnable && NDataModel.ModelType.BATCH == model.getModelType()
                    && !model.isMultiPartitionModel() && isBuildAllIndexes && isBuildAllIndexesFinally) {
                isOverlap = existedSegment.getSegRange().overlaps(segmentRangeToBuild)
                        && !segmentRangeToBuild.contains(existedSegment.getSegRange());
            } else {
                isOverlap = existedSegment.getSegRange().overlaps(segmentRangeToBuild);
            }
            if (isOverlap) {
                overlapsBuiltSegment.add(existedSegment);
            }
        }

        return overlapsBuiltSegment;
    }

    public ComputedColumnUsageResponse getComputedColumnUsages(String project) {
        aclEvaluate.checkProjectWritePermission(project);
        ComputedColumnUsageResponse ret = new ComputedColumnUsageResponse();
        List<NDataModel> models = getManager(NDataflowManager.class, project).listUnderliningDataModels();
        for (NDataModel model : models) {
            for (ComputedColumnDesc computedColumnDesc : model.getComputedColumnDescs()) {
                ret.addUsage(computedColumnDesc, model.getUuid());
            }
        }
        return ret;
    }

    /**
     * check if the computed column expressions are valid (in hive)
     * <p>
     * ccInCheck is optional, if provided, other cc in the model will skip hive check
     */
    public ComputedColumnCheckResponse checkComputedColumn(NDataModel model, String project, String ccInCheck) {
        aclEvaluate.checkProjectWritePermission(project);
        if (model.getUuid() == null) {
            model.updateRandomUuid();
        }

        model.init(getConfig(), project, getManager(NDataModelManager.class, project).getCCRelatedModels(model));
        model.getComputedColumnDescs().forEach(cc -> {
            String innerExp = PushDownUtil.massageComputedColumn(model, project, cc, null);
            cc.setInnerExpression(innerExp);
        });

        if (model.isSeekingCCAdvice()) {
            // if it's seeking for advice, it should have thrown exceptions by far
            throw new IllegalStateException("No advice could be provided");
        }

        checkCCNameAmbiguity(model);
        ComputedColumnDesc checkedCC = null;

        QueryContext.AclInfo aclInfo = AclPermissionUtil.createAclInfo(project, getCurrentUserGroups());
        AntiFlatChecker checker = new AntiFlatChecker(model.getJoinTables(), model);
        for (ComputedColumnDesc cc : model.getComputedColumnDescs()) {
            checkCCName(cc.getColumnName());

            if (!StringUtils.isEmpty(ccInCheck) && !StringUtils.equalsIgnoreCase(cc.getFullName(), ccInCheck)) {
                checkCascadeErrorOfNestedCC(cc, ccInCheck);
            } else {
                //replace computed columns with basic columns
                String antiFlattenLookup = checker.detectAntiFlattenLookup(cc);
                if (antiFlattenLookup != null) {
                    throw new KylinException(COMPUTED_COLUMN_DEPENDS_ANTI_FLATTEN_LOOKUP, String.format(Locale.ROOT,
                            MsgPicker.getMsg().getccOnAntiFlattenLookup(), antiFlattenLookup));
                }
                ComputedColumnDesc.simpleParserCheck(cc.getExpression(), model.getAliasMap().keySet());
                String innerExpression = PushDownUtil.massageComputedColumn(model, project, cc, aclInfo);
                cc.setInnerExpression(innerExpression);

                //check by data source, this could be slow
                long ts = System.currentTimeMillis();
                ComputedColumnEvalUtil.evaluateExprAndType(model, cc);
                logger.debug("Spent {} ms to visit data source to validate computed column expression: {}",
                        (System.currentTimeMillis() - ts), cc.getExpression());
                checkedCC = cc;
            }
        }

        Preconditions.checkState(checkedCC != null, "No computed column match: {}", ccInCheck);
        // check invalid measure removed due to cc data type change
        val modelManager = getManager(NDataModelManager.class, model.getProject());
        val oldDataModel = modelManager.getDataModelDesc(model.getUuid());

        // brand new model, no measure to remove
        if (oldDataModel == null)
            return getComputedColumnCheckResponse(checkedCC, new ArrayList<>());

        val copyModel = modelManager.copyForWrite(oldDataModel);
        val request = new ModelRequest(model);
        request.setProject(model.getProject());
        request.setMeasures(model.getAllMeasures());
        UpdateImpact updateImpact = semanticUpdater.updateModelColumns(copyModel, request);
        // get invalid measure names in original model
        val removedMeasures = updateImpact.getInvalidMeasures();
        val measureNames = oldDataModel.getAllMeasures().stream().filter(m -> removedMeasures.contains(m.getId()))
                .map(NDataModel.Measure::getName).collect(Collectors.toList());
        // get invalid measure names in request
        val removedRequestMeasures = updateImpact.getInvalidRequestMeasures();
        val requestMeasureNames = request.getAllMeasures().stream()
                .filter(m -> removedRequestMeasures.contains(m.getId())).map(NDataModel.Measure::getName)
                .collect(Collectors.toList());
        measureNames.addAll(requestMeasureNames);
        return getComputedColumnCheckResponse(checkedCC, measureNames);
    }

    static void checkCCName(String name) {
        if (PushDownConverterKeyWords.CALCITE.contains(name.toUpperCase(Locale.ROOT))
                || PushDownConverterKeyWords.HIVE.contains(name.toUpperCase(Locale.ROOT))) {
            throw new KylinException(INVALID_NAME,
                    String.format(Locale.ROOT, MsgPicker.getMsg().getInvalidComputerColumnNameWithKeyword(), name));
        }
        if (!Pattern.compile("^[a-zA-Z]+\\w*$").matcher(name).matches()) {
            throw new KylinException(INVALID_NAME,
                    String.format(Locale.ROOT, MsgPicker.getMsg().getInvalidComputerColumnName(), name));
        }
    }

    public void checkCCNameAmbiguity(NDataModel model) {
        Set<String> ambiguousCCNameSet = Sets.newHashSet();

        Set<String> ccColumnNames = Sets.newHashSet();
        for (ComputedColumnDesc cc : model.getComputedColumnDescs()) {
            if (ccColumnNames.contains(cc.getColumnName())) {
                ambiguousCCNameSet.add(cc.getColumnName());
            } else {
                ccColumnNames.add(cc.getColumnName());
            }
        }
        if (CollectionUtils.isEmpty(ccColumnNames)) {
            return;
        }

        for (TableRef table : model.getFactTables()) {
            for (TblColRef tblColRef : table.getColumns()) {
                if (tblColRef.getColumnDesc().isComputedColumn()) {
                    continue;
                }

                if (ccColumnNames.contains(tblColRef.getName())) {
                    ambiguousCCNameSet.add(tblColRef.getName());
                }
            }
        }

        if (CollectionUtils.isNotEmpty(ambiguousCCNameSet)) {
            buildDuplicateCCException(ambiguousCCNameSet);
        }
    }

    private void buildDuplicateCCException(Set<String> ambiguousCCNameSet) {
        StringBuilder error = new StringBuilder();
        ambiguousCCNameSet.forEach(name -> {
            error.append(String.format(Locale.ROOT, MsgPicker.getMsg().getCheckCcAmbiguity(), name));
            error.append("\r\n");
        });
        throw new KylinException(DUPLICATE_COMPUTED_COLUMN_NAME, error.toString());
    }

    void preProcessBeforeModelSave(NDataModel model, String project) {
        NDataModelManager modelManager = getManager(NDataModelManager.class, project);
        model.init(getConfig(), project, modelManager.getCCRelatedModels(model), true);
        massageModelFilterCondition(model);

        checkCCNameAmbiguity(model);

        try {
            NProjectLoader.updateCache(project);
            // Update CC expression from query transformers
            QueryContext.AclInfo aclInfo = AclPermissionUtil.createAclInfo(project, getCurrentUserGroups());
            for (ComputedColumnDesc ccDesc : model.getComputedColumnDescs()) {
                String ccExpression = PushDownUtil.massageComputedColumn(model, project, ccDesc, aclInfo);
                ccDesc.setInnerExpression(ccExpression);
                TblColRef tblColRef = model.findColumn(ccDesc.getTableAlias(), ccDesc.getColumnName());
                tblColRef.getColumnDesc().setComputedColumnExpr(ccExpression);
            }

            ComputedColumnEvalUtil.evalDataTypeOfCCInBatch(model, model.getComputedColumnDescs());
        } finally {
            NProjectLoader.removeCache();
        }
    }

    @Transaction(project = 0)
    public void updateModelDataCheckDesc(String project, String modelId, long checkOptions, long faultThreshold,
            long faultActions) {
        aclEvaluate.checkProjectWritePermission(project);
        final NDataModel dataModel = getManager(NDataModelManager.class, project).getDataModelDesc(modelId);
        if (dataModel == null) {
            throw new KylinException(MODEL_ID_NOT_EXIST, modelId);
        }

        dataModel.setDataCheckDesc(DataCheckDesc.valueOf(checkOptions, faultThreshold, faultActions));
        getManager(NDataModelManager.class, project).updateDataModelDesc(dataModel);
    }

    @Transaction(project = 1)
    public void deleteSegmentById(String model, String project, String[] ids, boolean force) {
        aclEvaluate.checkProjectOperationPermission(project);
        NDataModel dataModel = getManager(NDataModelManager.class, project).getDataModelDesc(model);
        if (ManagementType.TABLE_ORIENTED == dataModel.getManagementType()) {
            throw new KylinException(PERMISSION_DENIED,
                    String.format(Locale.ROOT, MsgPicker.getMsg().getModelSegmentCanNotRemove(), dataModel.getAlias()));
        }
        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        checkSegmentsExistById(model, project, ids);
        checkSegmentsStatus(model, project, ids, SegmentStatusEnumToDisplay.LOCKED);
        val indexPlan = getIndexPlan(model, project);
        NDataflow dataflow = dataflowManager.getDataflow(indexPlan.getUuid());
        Set<String> idsToDelete = Sets.newHashSet();
        for (String id : ids) {
            if (dataflow.getSegment(id) != null) {
                idsToDelete.add(id);
            } else {
                throw new IllegalArgumentException(
                        String.format(Locale.ROOT, MsgPicker.getMsg().getSegNotFound(), id, dataflow.getModelAlias()));
            }
        }
        removeSegment(project, dataflow.getUuid(), idsToDelete);
        offlineModelIfNecessary(dataflowManager, model);
    }

    public ModelFileSegments getModelFileSegments(String project, String modelAlias) {
        aclEvaluate.checkProjectOperationPermission(project);
        NDataflowManager dfManager = getManager(NDataflowManager.class, project);
        NDataflow df = dfManager.getDataflowByModelAlias(modelAlias);
        if (df == null || df.isBroken() || df.getModel().isBroken())
            return ModelFileSegments.broken(project, modelAlias);

        return FileSegments.getModelFileSegments(project, df.getModel().getId(), true);
    }

    @Transaction(project = 0)
    public void forceFileSegments(String project, String modelId, String storageLocation,
            Optional<List<String>> fileHashs, SegmentStatusEnum initStatus) {
        aclEvaluate.checkProjectOperationPermission(project);
        NDataflowManager dfManager = getManager(NDataflowManager.class, project);
        FileSegments.forceFileSegments(project, modelId, storageLocation, fileHashs, (fileSegRangeToCreate) -> {
            NDataSegment newSeg;
            if (initStatus == SegmentStatusEnum.NEW) {
                // normal new segment and kickoff build immediately
                IncrementBuildSegmentParams incrParams = new IncrementBuildSegmentParams(project, modelId,
                        fileSegRangeToCreate, true);
                modelBuildService.constructIncrementBuild(incrParams);
                newSeg = dfManager.getDataflow(modelId).getSegmentByName(fileSegRangeToCreate.proposeSegmentName());
            } else {
                // a new READY segment without immediate build
                newSeg = dfManager.appendSegment(dfManager.getDataflow(modelId), fileSegRangeToCreate, initStatus,
                        null);
            }
            return newSeg;
        }, (segmentIdToDelete) -> {
            deleteSegmentById(modelId, project, new String[] { segmentIdToDelete }, true);
            return true;
        });
    }

    public void removeSegment(String project, String dataflowId, Set<String> tobeRemoveSegmentIds) {
        KylinConfig kylinConfig = KylinConfig.getInstanceFromEnv();

        NDataflowManager dfMgr = NDataflowManager.getInstance(kylinConfig, project);
        NDataflow df = dfMgr.getDataflow(dataflowId);
        if (CollectionUtils.isEmpty(tobeRemoveSegmentIds)) {
            return;
        }

        List<NDataSegment> dataSegments = Lists.newArrayList();
        for (String tobeRemoveSegmentId : tobeRemoveSegmentIds) {
            NDataSegment dataSegment = df.getSegment(tobeRemoveSegmentId);
            if (dataSegment == null) {
                continue;
            }
            dataSegments.add(dataSegment);
        }

        if (CollectionUtils.isNotEmpty(dataSegments)) {
            NDataflowUpdate update = new NDataflowUpdate(df.getUuid());
            update.setToRemoveSegs(dataSegments.toArray(new NDataSegment[0]));
            dfMgr.updateDataflow(update);
        }
    }

    @Transaction(project = 0)
    public NDataSegment refreshSegment(String project, String indexPlanUuid, String segmentId) {
        NDataflowManager dfManager = NDataflowManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
        NDataflow df = dfManager.getDataflow(indexPlanUuid);
        NDataSegment segment = df.getSegment(segmentId);
        return dfManager.refreshSegment(df, segment.getSegRange());
    }

    private void offlineModelIfNecessary(NDataflowManager dfManager, String modelId) {
        NDataflow df = dfManager.getDataflow(modelId);
        if (df.getSegments().isEmpty() && RealizationStatusEnum.ONLINE == df.getStatus()) {
            dfManager.updateDataflowStatus(df.getId(), RealizationStatusEnum.OFFLINE);
        }
    }

    public void checkSegmentsExistById(String modelId, String project, String[] ids) {
        checkSegmentsExistById(modelId, project, ids, true);
    }

    public boolean checkSegmentsExistById(String modelId, String project, String[] ids, boolean shouldThrown) {
        Preconditions.checkNotNull(modelId);
        Preconditions.checkNotNull(project);
        Preconditions.checkNotNull(ids);
        aclEvaluate.checkProjectOperationPermission(project);

        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        IndexPlan indexPlan = getIndexPlan(modelId, project);
        NDataflow dataflow = dataflowManager.getDataflow(indexPlan.getUuid());

        List<String> notExistIds = Stream.of(ids).filter(segmentId -> null == dataflow.getSegment(segmentId))
                .filter(Objects::nonNull).collect(Collectors.toList());
        if (shouldThrown && !CollectionUtils.isEmpty(notExistIds)) {
            throw new KylinException(SEGMENT_NOT_EXIST_ID, StringUtils.join(notExistIds, ","));
        }
        return CollectionUtils.isEmpty(notExistIds);
    }

    private void checkSegmentsExistByName(String model, String project, String[] names) {
        checkSegmentsExistByName(model, project, names, true);
    }

    public boolean checkSegmentsExistByName(String model, String project, String[] names, boolean shouldThrow) {
        Preconditions.checkNotNull(model);
        Preconditions.checkNotNull(project);
        Preconditions.checkNotNull(names);
        aclEvaluate.checkProjectOperationPermission(project);

        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        IndexPlan indexPlan = getIndexPlan(model, project);
        NDataflow dataflow = dataflowManager.getDataflow(indexPlan.getUuid());

        List<String> notExistNames = Stream.of(names)
                .filter(segmentName -> null == dataflow.getSegmentByName(segmentName)).filter(Objects::nonNull)
                .collect(Collectors.toList());
        if (shouldThrow && !CollectionUtils.isEmpty(notExistNames)) {
            throw new KylinException(SEGMENT_NOT_EXIST_NAME, StringUtils.join(notExistNames, ","));
        }
        return CollectionUtils.isEmpty(notExistNames);
    }

    public void checkSegmentsStatus(String model, String project, String[] ids,
            SegmentStatusEnumToDisplay... statuses) {
        for (SegmentStatusEnumToDisplay status : statuses) {
            checkSegmentsStatus(model, project, ids, status);
        }
    }

    public void checkSegmentsStatus(String model, String project, String[] ids, SegmentStatusEnumToDisplay status) {
        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        IndexPlan indexPlan = getIndexPlan(model, project);
        NDataflow dataflow = dataflowManager.getDataflow(indexPlan.getUuid());
        Segments<NDataSegment> segments = dataflow.getSegments();
        ErrorCodeServer serverErrorCode = SegmentStatusEnumToDisplay.LOCKED == status ? SEGMENT_LOCKED : SEGMENT_STATUS;
        Set<String> allIndexJobRunningSegments = SegmentUtil.getAllIndexJobRunningSegments(dataflow.getModel());
        for (String id : ids) {
            val segment = dataflow.getSegment(id);
            if (SegmentUtil.getSegmentStatusToDisplay(segments, segment, null, allIndexJobRunningSegments) == status) {
                throw new KylinException(serverErrorCode, segment.displayIdName(), status);
            }
        }
    }

    private void checkSegmentsContinuous(String modelId, String project, String[] ids) {
        val dfManager = getManager(NDataflowManager.class, project);
        val indexPlan = getIndexPlan(modelId, project);
        val df = dfManager.getDataflow(indexPlan.getUuid());
        List<NDataSegment> segmentList = Arrays.stream(ids).map(df::getSegment).sorted(NDataSegment::compareTo)
                .collect(Collectors.toList());

        for (int i = 0; i < segmentList.size() - 1; i++) {
            if (!segmentList.get(i).getSegRange().connects(segmentList.get(i + 1).getSegRange())) {
                throw new KylinException(SEGMENT_MERGE_CONTAINS_GAPS);
            }
        }
    }

    public Pair<Long, Long> checkMergeSegments(MergeSegmentParams params) {
        String project = params.getProject();
        String modelId = params.getModelId();
        String[] ids = params.getSegmentIds();

        aclEvaluate.checkProjectOperationPermission(project);

        val dfManager = getManager(NDataflowManager.class, project);
        val df = dfManager.getDataflow(modelId);

        checkSegmentsExistById(modelId, project, ids);
        checkSegmentsStatus(modelId, project, ids, SegmentStatusEnumToDisplay.LOADING,
                SegmentStatusEnumToDisplay.REFRESHING, SegmentStatusEnumToDisplay.MERGING,
                SegmentStatusEnumToDisplay.LOCKED);
        checkSegmentsContinuous(modelId, project, ids);

        long start = Long.MAX_VALUE;
        long end = -1;

        for (String id : ids) {
            val segment = df.getSegment(id);
            if (segment == null) {
                throw new IllegalArgumentException(
                        String.format(Locale.ROOT, MsgPicker.getMsg().getSegNotFound(), id, df.getModelAlias()));
            }

            if (SegmentStatusEnum.READY != segment.getStatus() && SegmentStatusEnum.WARNING != segment.getStatus()) {
                throw new KylinException(PERMISSION_DENIED, MsgPicker.getMsg().getInvalidMergeSegment());
            }
            val segmentStart = segment.getTSRange().getStart();
            val segmentEnd = segment.getTSRange().getEnd();

            if (segmentStart < start)
                start = segmentStart;

            if (segmentEnd > end)
                end = segmentEnd;
        }

        return Pair.newPair(start, end);
    }

    public BuildBaseIndexResponse updateDataModelSemantic(String project, ModelRequest request) {
        return updateDataModelSemantic(project, request, true);
    }

    public BuildBaseIndexResponse updateDataModelSemantic(String project, ModelRequest request, boolean isCheckFlat) {
        try {
            UnitOfWorkParams<BuildBaseIndexResponse> params = UnitOfWorkParams.<BuildBaseIndexResponse> builder()
                    .unitName(project).processor(() -> {
                        aclEvaluate.checkProjectWritePermission(project);
                        semanticUpdater.expandModelRequest(request);
                        checkModelRequest(request);
                        checkModelPermission(project, request.getUuid());
                        validatePartitionDateColumn(request);

                        val modelId = request.getUuid();
                        val modelManager = getManager(NDataModelManager.class, project);
                        val originModel = modelManager.getDataModelDesc(modelId);

                        val copyModel = modelManager.copyForWrite(originModel);
                        UpdateImpact updateImpact = semanticUpdater.updateModelColumns(copyModel, request, true);
                        copyModel.init(modelManager.getConfig(), project, modelManager.getCCRelatedModels(copyModel));

                        BaseIndexUpdateHelper baseIndexUpdater = new BaseIndexUpdateHelper(originModel,
                                needHandleBaseIndexType(request));

                        preProcessBeforeModelSave(copyModel, project);
                        val updated = modelManager.updateDataModelDesc(copyModel);

                        // 1. delete old internal measure
                        // 2. expand new measures
                        // the ordering of the two steps is important as tomb internal measure
                        // should not be used to construct new expandable measures
                        semanticUpdater.deleteExpandableMeasureInternalMeasures(updated);
                        semanticUpdater.expandExpandableMeasure(updated);
                        preProcessBeforeModelSave(updated, project);
                        getManager(NDataModelManager.class, project).updateDataModelDesc(updated);
                        indexPlanService.checkPartitionDimensionForV3Storage(project, modelId, getConfig());

                        indexPlanService.updateForMeasureChange(project, modelId, updateImpact.getInvalidMeasures(),
                                updateImpact.getReplacedMeasures());
                        Set<Integer> affectedSet = updateImpact.getAffectedIds();
                        val affectedLayoutSet = getAffectedLayouts(project, modelId, affectedSet);
                        if (affectedLayoutSet.size() > 0)
                            indexPlanService.reloadLayouts(project, modelId, affectedLayoutSet);
                        indexPlanService.clearShardColIfNotDim(project, modelId);

                        var newModel = modelManager.getDataModelDesc(modelId);

                        checkIndexColumnExist(project, modelId, originModel);

                        if (isCheckFlat) {
                            checkFlatTableSql(newModel);
                        }

                        val needBuild = semanticUpdater.doHandleSemanticUpdate(project, modelId, originModel,
                                request.getStart(), request.getEnd());

                        updateExcludedCheckerResult(project, request);
                        BuildBaseIndexResponse baseIndexResponse = baseIndexUpdater.update(indexPlanService);
                        if (!request.isSaveOnly() && (needBuild || baseIndexResponse.hasIndexChange())) {
                            val targetSegments = SegmentUtil.getValidSegments(modelId, project).stream()
                                    .map(NDataSegment::getId).collect(Collectors.toSet());
                            semanticUpdater.buildForModelSegments(project, modelId, targetSegments);
                        }
                        modelChangeSupporters.forEach(listener -> listener.onUpdate(project, modelId));
                        return baseIndexResponse;
                    }).build();
            return EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(params);
        } catch (TransactionException te) {
            Throwable root = te.getCause();
            if (root instanceof RuntimeException) {
                throw (RuntimeException) root;
            }
            throw te;
        }
    }

    public void updateExcludedCheckerResult(String project, ModelRequest request) {
        NDataModelManager modelManager = getManager(NDataModelManager.class, project);
        NIndexPlanManager indexPlanManager = getManager(NIndexPlanManager.class, project);

        String uuid = request.getUuid();
        NDataModel convertedModel = modelManager.getDataModelDesc(uuid);
        List<JoinTableDesc> joinTables = convertedModel.getJoinTables();

        IndexPlan indexPlan = indexPlanManager.getIndexPlan(uuid);
        AntiFlatChecker checker = new AntiFlatChecker(joinTables, convertedModel);
        List<ComputedColumnDesc> invalidCCList = checker.getInvalidComputedColumns(convertedModel);
        Set<Integer> invalidDimensions = checker.getInvalidDimensions(convertedModel);
        Set<Integer> invalidMeasures = checker.getInvalidMeasures(convertedModel);
        Set<Integer> invalidScope = Sets.newHashSet();
        invalidScope.addAll(invalidDimensions);
        invalidScope.addAll(invalidMeasures);
        Set<Long> invalidIndexes = checker.getInvalidIndexes(indexPlan, invalidScope);

        Map<String, ComputedColumnDesc> invalidCCMap = Maps.newHashMap();
        invalidCCList.forEach(cc -> invalidCCMap.put(cc.getColumnName(), cc));
        if (!invalidIndexes.isEmpty()) {
            indexPlanService.removeIndexes(project, uuid, invalidIndexes, invalidDimensions, invalidMeasures);
        }

        modelManager.updateDataModel(uuid, copyForWrite -> {
            copyForWrite.getComputedColumnDescs().removeIf(cc -> invalidCCMap.containsKey(cc.getColumnName()));
            copyForWrite.getAllMeasures().forEach(measure -> {
                if (invalidMeasures.contains(measure.getId())) {
                    measure.setTomb(true);
                }
            });
            copyForWrite.getAllNamedColumns().forEach(column -> {
                if (!column.isExist()) {
                    return;
                }
                if (invalidDimensions.contains(column.getId())) {
                    column.setStatus(NDataModel.ColumnStatus.EXIST);
                }

                if (invalidCCMap.containsKey(column.getName())) {
                    String colName = column.getAliasDotColumn();
                    final String fullName = invalidCCMap.get(column.getName()).getFullName();
                    if (fullName.equalsIgnoreCase(colName)) {
                        column.setStatus(NDataModel.ColumnStatus.TOMB);
                    }
                }
            });
        });
    }

    public String[] convertSegmentIdWithName(String modelId, String project, String[] segIds, String[] segNames) {
        String[] ids = segIds;

        if (ArrayUtils.isEmpty(segNames)) {
            return ids;
        }

        aclEvaluate.checkProjectOperationPermission(project);
        checkSegmentsExistByName(modelId, project, segNames);

        NDataflow dataflow = getManager(NDataflowManager.class, project).getDataflow(modelId);

        ids = Stream.of(segNames).map(segmentName -> {
            val segmentByName = dataflow.getSegmentByName(segmentName);
            return Objects.isNull(segmentByName) ? null : segmentByName.getId();
        }).toArray(String[]::new);

        return ids;
    }

    private Set<Long> getAffectedLayouts(String project, String modelId, Set<Integer> modifiedSet) {
        var affectedLayoutSet = new HashSet<Long>();
        val indePlanManager = NIndexPlanManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
        val indexPlan = indePlanManager.getIndexPlan(modelId);
        for (LayoutEntity layoutEntity : indexPlan.getAllLayouts()) {
            if (layoutEntity.getColOrder().stream().anyMatch(modifiedSet::contains)) {
                affectedLayoutSet.add(layoutEntity.getId());
            }
        }
        return affectedLayoutSet;
    }

    private void checkIndexColumnExist(String project, String modelId, NDataModel originModel) {
        val indexPlanManager = NIndexPlanManager.getInstance(KylinConfig.getInstanceFromEnv(), project);
        val indexPlan = indexPlanManager.getIndexPlan(modelId);
        var newModel = getManager(NDataModelManager.class, project).getDataModelDesc(modelId);

        // check agg group contains removed dimensions
        val rule = indexPlan.getRuleBasedIndex();
        if (rule != null) {
            if (!newModel.getEffectiveDimensions().keySet().containsAll(rule.getDimensions())) {
                val allDimensions = rule.getDimensions();
                val dimensionNames = allDimensions.stream()
                        .filter(id -> !newModel.getEffectiveDimensions().containsKey(id))
                        .map(originModel::getColumnNameByColumnId).collect(Collectors.toList());
                throw new KylinException(FAILED_UPDATE_MODEL, String.format(Locale.ROOT,
                        MsgPicker.getMsg().getDimensionNotfound(), StringUtils.join(dimensionNames, ",")));
            }

            for (NAggregationGroup agg : rule.getAggregationGroups()) {
                if (!newModel.getEffectiveMeasures().keySet().containsAll(Sets.newHashSet(agg.getMeasures()))) {
                    val measureNames = Arrays.stream(agg.getMeasures())
                            .filter(measureId -> !newModel.getEffectiveMeasures().containsKey(measureId))
                            .map(originModel::getMeasureNameByMeasureId).collect(Collectors.toList());
                    throw new KylinException(FAILED_UPDATE_MODEL, String.format(Locale.ROOT,
                            MsgPicker.getMsg().getMeasureNotfound(), StringUtils.join(measureNames, ",")));
                }
            }
        }

        //check table index contains removed columns
        val tableIndexColumns = indexPlan.getIndexes().stream().filter(IndexEntity::isTableIndex)
                .map(IndexEntity::getDimensions).flatMap(List::stream).collect(Collectors.toSet());
        val allSelectedColumns = newModel.getAllSelectedColumns().stream().map(NDataModel.NamedColumn::getId)
                .collect(Collectors.toList());
        if (!allSelectedColumns.containsAll(tableIndexColumns)) {
            val columnNames = tableIndexColumns.stream().filter(x -> !allSelectedColumns.contains(x))
                    .map(originModel::getColumnNameByColumnId).collect(Collectors.toList());
            throw new KylinException(FAILED_UPDATE_MODEL, String.format(Locale.ROOT,
                    MsgPicker.getMsg().getDimensionNotfound(), StringUtils.join(columnNames, ",")));
        }

        //check recommend agg index contains removed columns
        val recommendAggIndex = indexPlan.getIndexes().stream().filter(x -> !x.isTableIndex())
                .collect(Collectors.toList());
        for (val aggIndex : recommendAggIndex) {
            if (!newModel.getEffectiveDimensions().keySet().containsAll(aggIndex.getDimensions())) {
                val allDimensions = aggIndex.getDimensions();
                val dimensionNames = allDimensions.stream()
                        .filter(id -> !newModel.getEffectiveDimensions().containsKey(id))
                        .map(originModel::getColumnNameByColumnId).collect(Collectors.toList());
                throw new KylinException(FAILED_UPDATE_MODEL, String.format(Locale.ROOT,
                        MsgPicker.getMsg().getDimensionNotfound(), StringUtils.join(dimensionNames, ",")));
            }

            if (!newModel.getEffectiveMeasures().keySet().containsAll(aggIndex.getMeasures())) {
                val measureNames = aggIndex.getMeasures().stream()
                        .filter(measureId -> !newModel.getEffectiveMeasures().containsKey(measureId))
                        .map(originModel::getMeasureNameByMeasureId).collect(Collectors.toList());
                throw new KylinException(FAILED_UPDATE_MODEL, String.format(Locale.ROOT,
                        MsgPicker.getMsg().getMeasureNotfound(), StringUtils.join(measureNames, ",")));

            }
        }
    }

    public void updateBrokenModel(String project, ModelRequest modelRequest, Set<Integer> columnIds) {
        val modelManager = getManager(NDataModelManager.class, project);
        val origin = modelManager.getDataModelDesc(modelRequest.getUuid());
        val copyModel = modelManager.copyForWrite(origin);
        try {
            semanticUpdater.updateModelColumns(copyModel, modelRequest);
        } catch (Exception e) {
            log.warn("Update existing model failed.{}", ThreadUtil.getKylinStackTrace());
        }
        copyModel.setBrokenReason(NDataModel.BrokenReason.SCHEMA);
        copyModel.getAllNamedColumns().forEach(namedColumn -> {
            if (columnIds.contains(namedColumn.getId())) {
                namedColumn.setStatus(NDataModel.ColumnStatus.TOMB);
            }
        });
        modelManager.updateDataModelDesc(copyModel);
    }

    public NDataModel repairBrokenModel(String project, ModelRequest modelRequest) throws Exception {
        aclEvaluate.checkProjectWritePermission(project);
        semanticUpdater.expandModelRequest(modelRequest);
        val modelManager = getManager(NDataModelManager.class, project);
        val origin = modelManager.getDataModelDesc(modelRequest.getId());
        val broken = modelQuerySupporter.getBrokenModel(project, origin.getId());
        if (ManagementType.TABLE_ORIENTED == broken.getManagementType()) {
            throw new KylinException(FAILED_UPDATE_MODEL, "Can not repair model manually smart mode!");
        }
        broken.setPartitionDesc(modelRequest.getPartitionDesc());
        broken.setMultiPartitionDesc(modelRequest.getMultiPartitionDesc());
        broken.setMultiPartitionKeyMapping(modelRequest.getMultiPartitionKeyMapping());
        broken.setFilterCondition(modelRequest.getFilterCondition());
        broken.setJoinTables(modelRequest.getJoinTables());
        discardInvalidColumnAndMeasure(broken, modelRequest);
        broken.init(getConfig(), project, getManager(NDataflowManager.class, project).listUnderliningDataModels());
        broken.setBrokenReason(NDataModel.BrokenReason.NULL);
        String format = probeDateFormatIfNotExist(project, broken);
        return EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
            NDataModelManager modelMgr = getManager(NDataModelManager.class, project);
            semanticUpdater.updateModelColumns(broken, modelRequest);
            NDataModel copy = modelManager.copyForWrite(broken);
            NDataModel model = modelMgr.updateDataModelDesc(copy);
            modelMgr.updateDataModel(model.getUuid(), copyForWrite -> {
                modelChangeSupporters.forEach(listener -> listener.onUpdateSingle(project, model.getUuid()));
                saveDateFormatIfNotExist(project, model.getUuid(), format);
            });
            getManager(NDataflowManager.class, project).updateDataflowStatus(broken.getId(),
                    RealizationStatusEnum.ONLINE);
            return modelMgr.getDataModelDesc(model.getUuid());
        }, project);
    }

    public void discardInvalidColumnAndMeasure(NDataModel brokenModel, ModelRequest modelRequest) {
        NDataModel newModel = convertAndInitDataModel(modelRequest, brokenModel.getProject());
        Set<String> aliasDotColumns = newModel.getAllNamedColumns().stream().filter(NDataModel.NamedColumn::isExist)
                .map(NDataModel.NamedColumn::getAliasDotColumn).collect(Collectors.toSet());
        for (NDataModel.NamedColumn column : brokenModel.getAllNamedColumns()) {
            if (!aliasDotColumns.contains(column.getAliasDotColumn())) {
                column.setStatus(NDataModel.ColumnStatus.TOMB);
            }
        }
        for (NDataModel.Measure measure : brokenModel.getAllMeasures()) {
            if (measure.isTomb()) {
                continue;
            }
            FunctionDesc function = measure.getFunction();
            for (ParameterDesc param : function.getParameters()) {
                if (!param.isConstant() && !aliasDotColumns.contains(param.getValue())) {
                    measure.setTomb(true);
                    break;
                }
            }
        }
    }

    public NDataModel convertToDataModel(ModelRequest modelDesc) {
        return semanticUpdater.convertToDataModel(modelDesc);
    }

    public AffectedModelsResponse getAffectedModelsByDeletingTable(String tableName, String project) {
        aclEvaluate.checkProjectReadPermission(project);
        val dataflowManager = getManager(NDataflowManager.class, project);
        val table = getManager(NTableMetadataManager.class, project).getTableDesc(tableName);
        val response = new AffectedModelsResponse();
        val models = dataflowManager.getModelsUsingTable(table).stream().map(RootPersistentEntity::getUuid)
                .collect(Collectors.toList());
        var size = 0;
        response.setModels(models);
        for (val model : models) {
            size += dataflowManager.getDataflowStorageSize(model);
        }
        response.setByteSize(size);
        return response;
    }

    public List<ModelConfigResponse> getModelConfig(String project, String modelName) {
        aclEvaluate.checkProjectReadPermission(project);
        val responseList = Lists.<ModelConfigResponse> newArrayList();
        boolean streamingEnabled = getConfig().isStreamingEnabled();
        getManager(NDataflowManager.class, project).listUnderliningDataModels().stream()
                .filter(model -> (StringUtils.isEmpty(modelName) || model.getAlias().contains(modelName)))
                .filter(model -> model.isAccessible(streamingEnabled) && !model.fusionModelBatchPart())
                .forEach(dataModel -> {
                    val response = new ModelConfigResponse();
                    response.setModel(dataModel.getUuid());
                    response.setAlias(dataModel.getAlias());
                    val segmentConfig = dataModel.getSegmentConfig();
                    response.setAutoMergeEnabled(segmentConfig.getAutoMergeEnabled());
                    response.setAutoMergeTimeRanges(segmentConfig.getAutoMergeTimeRanges());
                    response.setVolatileRange(segmentConfig.getVolatileRange());
                    response.setRetentionRange(segmentConfig.getRetentionRange());
                    response.setConfigLastModified(dataModel.getConfigLastModified());
                    response.setConfigLastModifier(dataModel.getConfigLastModifier());
                    val indexPlan = getIndexPlan(dataModel.getUuid(), project);
                    if (indexPlan != null) {
                        val overrideProps = indexPlan.getOverrideProps();
                        response.getOverrideProps().putAll(overrideProps);
                        MODEL_CONFIG_BLOCK_LIST.forEach(item -> response.getOverrideProps().remove(item));
                    }
                    responseList.add(response);
                });
        return responseList;
    }

    @Transaction(project = 0)
    public void updateModelConfig(String project, String modelId, ModelConfigRequest request) {
        aclEvaluate.checkProjectWritePermission(project);
        checkModelConfigParameters(request);
        val dataModelManager = getManager(NDataModelManager.class, project);
        dataModelManager.updateDataModel(modelId, copyForWrite -> {
            val segmentConfig = copyForWrite.getSegmentConfig();
            segmentConfig.setAutoMergeEnabled(request.getAutoMergeEnabled());
            segmentConfig.setAutoMergeTimeRanges(request.getAutoMergeTimeRanges());
            segmentConfig.setVolatileRange(request.getVolatileRange());
            segmentConfig.setRetentionRange(request.getRetentionRange());

            copyForWrite.setConfigLastModified(System.currentTimeMillis());
            copyForWrite.setConfigLastModifier(BasicService.getUsername());
        });
        val indexPlan = getIndexPlan(modelId, project);
        val indexPlanManager = getManager(NIndexPlanManager.class, project);
        val overrideProps = request.getOverrideProps();
        indexPlanManager.updateIndexPlan(indexPlan.getUuid(), copyForWrite -> {
            // affected by "kylin.cube.aggrgroup.is-base-cuboid-always-valid" config
            String affectProp = "kylin.cube.aggrgroup.is-base-cuboid-always-valid";
            String oldProp = copyForWrite.getOverrideProps().get(affectProp);

            copyForWrite.setOverrideProps(overrideProps);
            String newProp = copyForWrite.getOverrideProps().get(affectProp);

            boolean affectedByProp = !StringUtils.equals(oldProp, newProp);
            if (affectedByProp && copyForWrite.getRuleBasedIndex() != null) {
                val newRule = JsonUtil.deepCopyQuietly(copyForWrite.getRuleBasedIndex(), RuleBasedIndex.class);
                newRule.setLastModifiedTime(System.currentTimeMillis());
                newRule.setLayoutIdMapping(Lists.newArrayList());
                copyForWrite.setRuleBasedIndex(newRule);
            }
        });
    }

    private void checkPropParameter(ModelConfigRequest request) {
        val props = request.getOverrideProps();
        if (props == null) {
            throw new KylinException(INVALID_PARAMETER,
                    String.format(Locale.ROOT, MsgPicker.getMsg().getInvalidNullValue(), "override_props"));
        }
        for (val pair : props.entrySet()) {
            if (Objects.isNull(pair.getValue())) {
                throw new KylinException(INVALID_PARAMETER,
                        String.format(Locale.ROOT, MsgPicker.getMsg().getInvalidNullValue(), pair.getKey()));
            }
        }
        String cores = props.get("kylin.engine.spark-conf.spark.executor.cores");
        String instances = props.get("kylin.engine.spark-conf.spark.executor.instances");
        String pattitions = props.get("kylin.engine.spark-conf.spark.sql.shuffle.partitions");
        String memory = props.get("kylin.engine.spark-conf.spark.executor.memory");
        String baseCuboidAllowed = props.get("kylin.cube.aggrgroup.is-base-cuboid-always-valid");
        if (null != cores && !StringHelper.validateNumber(cores)) {
            throw new KylinException(INVALID_PARAMETER,
                    String.format(Locale.ROOT, MsgPicker.getMsg().getInvalidIntegerFormat(), "spark.executor.cores"));
        }
        if (null != instances && !StringHelper.validateNumber(instances)) {
            throw new KylinException(INVALID_PARAMETER, String.format(Locale.ROOT,
                    MsgPicker.getMsg().getInvalidIntegerFormat(), "spark.executor.instances"));
        }
        if (null != pattitions && !StringHelper.validateNumber(pattitions)) {
            throw new KylinException(INVALID_PARAMETER, String.format(Locale.ROOT,
                    MsgPicker.getMsg().getInvalidIntegerFormat(), "spark.sql.shuffle.partitions"));
        }
        if (null != memory
                && (!memory.endsWith("g") || !StringHelper.validateNumber(memory.substring(0, memory.length() - 1)))) {
            throw new KylinException(INVALID_PARAMETER,
                    String.format(Locale.ROOT, MsgPicker.getMsg().getInvalidMemorySize(), "spark.executor.memory"));
        }
        if (null != baseCuboidAllowed && !StringHelper.validateBoolean(baseCuboidAllowed)) {
            throw new KylinException(INVALID_PARAMETER, String.format(Locale.ROOT,
                    MsgPicker.getMsg().getInvalidBooleanFormat(), "is-base-cuboid-always-valid"));
        }
    }

    @VisibleForTesting
    public void checkModelConfigParameters(ModelConfigRequest request) {
        Boolean autoMergeEnabled = request.getAutoMergeEnabled();
        List<AutoMergeTimeEnum> timeRanges = request.getAutoMergeTimeRanges();
        VolatileRange volatileRange = request.getVolatileRange();
        RetentionRange retentionRange = request.getRetentionRange();

        if (Boolean.TRUE.equals(autoMergeEnabled) && (null == timeRanges || timeRanges.isEmpty())) {
            throw new KylinException(INVALID_PARAMETER, MsgPicker.getMsg().getInvalidAutoMergeConfig());
        }
        if (null != volatileRange && volatileRange.isVolatileRangeEnabled()
                && (volatileRange.getVolatileRangeNumber() < 0 || null == volatileRange.getVolatileRangeType())) {
            throw new KylinException(INVALID_PARAMETER, MsgPicker.getMsg().getInvalidVolatileRangeConfig());
        }
        if (null != retentionRange && retentionRange.isRetentionRangeEnabled()
                && retentionRange.getRetentionRangeNumber() < 0) {
            throw new KylinException(INVALID_PARAMETER, MsgPicker.getMsg().getInvalidRetentionRangeConfig());
        }
        checkPropParameter(request);
    }

    public ExistedDataRangeResponse getLatestDataRange(String project, String modelId, PartitionDesc desc) {
        aclEvaluate.checkProjectReadPermission(project);
        Preconditions.checkNotNull(modelId);
        try {
            val df = getManager(NDataflowManager.class, project).getDataflow(modelId);
            val model = df.getModel();
            val table = model.getRootFactTableName();
            if (Objects.nonNull(desc) && !desc.equals(model.getPartitionDesc())) {
                Pair<String, String> pushdownResult = getPartitionColMinMaxValue(project, table, desc);
                return new ExistedDataRangeResponse(pushdownResult.getFirst(), pushdownResult.getSecond());
            }
            Pair<String, String> pushdownResult = getPartitionColMinMaxValue(project, table, model.getPartitionDesc());
            pushdownResult.setFirst(PushDownUtil.calcStart(pushdownResult.getFirst(), df.getCoveredRange()));
            if (Long.parseLong(pushdownResult.getFirst()) > Long.parseLong(pushdownResult.getSecond())) {
                pushdownResult.setSecond(pushdownResult.getFirst());
            }
            return new ExistedDataRangeResponse(pushdownResult.getFirst(), pushdownResult.getSecond());
        } catch (Exception e) {
            if (e instanceof KylinTimeoutException) {
                throw new KylinException(FAILED_DETECT_DATA_RANGE, MsgPicker.getMsg().getpushdownDatarangeTimeout());
            }
            Throwable cause = e.getCause();
            if (cause instanceof KylinException) {
                ServerErrorCode code = VIEW_PARTITION_DATE_FORMAT_DETECTION_FORBIDDEN;
                if (((KylinException) cause).getErrorCode().equals(code.toErrorCode())) {
                    throw new KylinException(code, MsgPicker.getMsg().getViewDateFormatDetectionError());
                }
            }
            throw new KylinException(INVALID_RANGE, MsgPicker.getMsg().getPushdownDatarangeError());
        }
    }

    public PurgeModelAffectedResponse getPurgeModelAffectedResponse(String project, String model) {
        aclEvaluate.checkProjectWritePermission(project);
        val response = new PurgeModelAffectedResponse();
        val byteSize = getManager(NDataflowManager.class, project).getDataflowStorageSize(model);
        response.setByteSize(byteSize);
        long jobSize = getManager(ExecutableManager.class, project).countByModelAndStatus(model,
                ExecutableState::isProgressing);
        response.setRelatedJobSize(jobSize);
        return response;
    }

    public ComputedColumnCheckResponse getComputedColumnCheckResponse(ComputedColumnDesc ccDesc,
            List<String> removedMeasures) {
        val response = new ComputedColumnCheckResponse();
        response.setComputedColumnDesc(ccDesc);
        response.setRemovedMeasures(removedMeasures);
        return response;
    }

    public ModelSaveCheckResponse checkBeforeModelSave(ModelRequest modelRequest) {
        aclEvaluate.checkProjectWritePermission(modelRequest.getProject());
        semanticUpdater.expandModelRequest(modelRequest);

        checkModelDimensions(modelRequest);
        checkModelMeasures(modelRequest);
        checkModelJoinConditions(modelRequest);

        validateFusionModelDimension(modelRequest);
        NDataModel model = semanticUpdater.convertToDataModel(modelRequest);

        for (NDataModel.Measure measure : model.getAllMeasures()) {
            measure.getFunction().init(model);
        }

        if (modelRequest.getPartitionDesc() != null
                && !KylinConfig.getInstanceFromEnv().isUseBigIntAsTimestampForPartitionColumn()) {
            PartitionDesc partitionDesc = modelRequest.getPartitionDesc();
            partitionDesc.init(model);
            if (!partitionDesc.checkIntTypeDateFormat()) {
                throw new KylinException(JobErrorCode.JOB_INT_DATE_FORMAT_NOT_MATCH_ERROR,
                        "int/bigint data type only support yyyymm/yyyymmdd format");
            }
        }

        NDataModel oldDataModel = getManager(NDataModelManager.class, model.getProject())
                .getDataModelDesc(model.getUuid());
        Set<Long> affectedLayouts = Sets.newHashSet();
        if (oldDataModel != null && !oldDataModel.isBroken()) {
            model = getManager(NDataModelManager.class, model.getProject()).copyForWrite(oldDataModel);
            UpdateImpact updateImpact = semanticUpdater.updateModelColumns(model, modelRequest);
            Set<Integer> modifiedSet = updateImpact.getAffectedIds();
            affectedLayouts = getAffectedLayouts(oldDataModel.getProject(), oldDataModel.getId(), modifiedSet);
        }
        try {
            massageModelFilterCondition(model);
        } catch (Exception e) {
            throw new KylinException(INVALID_FILTER_CONDITION, e);
        }
        if (!affectedLayouts.isEmpty())
            return new ModelSaveCheckResponse(true);
        return new ModelSaveCheckResponse();
    }

    private void checkCascadeErrorOfNestedCC(ComputedColumnDesc cc, String ccInCheck) {
        String upperCcInCheck = ccInCheck.toUpperCase(Locale.ROOT);
        try {
            SqlVisitor<Void> sqlVisitor = new SqlBasicVisitor<Void>() {
                @Override
                public Void visit(SqlIdentifier id) {
                    if (id.toString().equals(upperCcInCheck))
                        throw new Util.FoundOne(id);
                    return null;
                }
            };
            CalciteParser.getExpNode(cc.getExpression()).accept(sqlVisitor);
        } catch (Util.FoundOne e) {
            throw new KylinException(COMPUTED_COLUMN_CASCADE_ERROR, String.format(Locale.ROOT,
                    MsgPicker.getMsg().getNestedCcCascadeError(), ccInCheck, cc.getFullName()));
        }
    }

    /**
     * Convert model filter condition:
     * 1. transform special function and backtick
     * 2. use Calcite to add table name
     * 3. massage for push-down
     * 4. update the filter condition
     */
    void massageModelFilterCondition(final NDataModel model) {
        String filterCondition = model.getFilterCondition();
        if (StringUtils.isBlank(filterCondition)) {
            return;
        }

        filterCondition = QueryUtil.adaptCalciteSyntax(filterCondition);
        String newFilterCondition = addTableNameIfNotExist(filterCondition, model);
        QueryContext.AclInfo aclInfo = AclPermissionUtil.createAclInfo(model.getProject(), getCurrentUserGroups());
        String massaged = PushDownUtil.massageExpression(model, model.getProject(), newFilterCondition, aclInfo);
        model.setFilterCondition(massaged);
    }

    @VisibleForTesting
    String addTableNameIfNotExist(final String expr, final NDataModel model) {
        SqlNode sqlNode = CalciteParser.getExpNode(expr);
        SqlVisitor<Void> sqlVisitor = new SqlIdentifierFormatterVisitor(expr, model.getAllNamedColumns());
        sqlNode.accept(sqlVisitor);

        if (!NProjectManager.getProjectConfig(model.getProject()).isBuildExcludedTableEnabled()) {
            AntiFlatChecker checker = new AntiFlatChecker(model.getJoinTables(), model);
            String antiFlatLookup = checker.detectFilterCondition(sqlNode.toString());
            if (antiFlatLookup != null) {
                throw new KylinException(FILTER_CONDITION_DEPENDS_ANTI_FLATTEN_LOOKUP, String.format(Locale.ROOT,
                        MsgPicker.getMsg().getFilterConditionOnAntiFlattenLookup(), antiFlatLookup));
            }
        }
        String exp = sqlNode.toSqlString(CalciteSqlDialect.DEFAULT, true).toString();
        return CalciteParser.normalize(exp);
    }

    public NModelDescResponse getModelDesc(String modelAlias, String project) {
        if (getManager(NProjectManager.class).getProject(project) == null) {
            throw new KylinException(PROJECT_NOT_EXIST, project);
        }
        NDataModel dataModel = getManager(NDataModelManager.class, project).getDataModelDescByAlias(modelAlias);
        if (dataModel == null) {
            throw new KylinException(MODEL_NAME_NOT_EXIST, modelAlias);
        }
        NDataModelResponse model = new NDataModelResponse(dataModel);
        NModelDescResponse response = new NModelDescResponse();
        response.setUuid(model.getUuid());
        response.setLastModified(model.getLastModified());
        response.setCreateTime(model.getCreateTime());
        response.setVersion(model.getVersion());
        response.setName(model.getAlias());
        response.setProject(model.getProject());
        response.setDescription(model.getDescription());
        response.setComputedColumnDescs(model.getComputedColumnDescs());
        response.setFactTable(model.getRootFactTableName());
        response.setJoinTables(model.getJoinTables());

        response.setMeasures(model.getMeasures());
        List<NModelDescResponse.Dimension> dims = model.getNamedColumns().stream() //
                .map(namedColumn -> new NModelDescResponse.Dimension(namedColumn, false)) //
                .collect(Collectors.toList());
        response.setDimensions(dims);

        IndexPlan indexPlan = getIndexPlan(response.getUuid(), project);
        if (indexPlan.getRuleBasedIndex() != null) {
            List<AggGroupResponse> aggGroupResponses = indexPlan.getRuleBasedIndex().getAggregationGroups().stream() //
                    .map(x -> new AggGroupResponse(dataModel, x)) //
                    .collect(Collectors.toList());
            response.setAggregationGroups(aggGroupResponses);
        } else {
            response.setAggregationGroups(new ArrayList<>());
        }
        return response;
    }

    public void updateModelPartitionColumn(String project, String modelAlias,
            ModelParatitionDescRequest modelParatitionDescRequest) {
        aclEvaluate.checkProjectWritePermission(project);
        if (getManager(NProjectManager.class).getProject(project) == null) {
            throw new KylinException(PROJECT_NOT_EXIST, project);
        }
        NDataModel oldDataModel = getManager(NDataModelManager.class, project).getDataModelDescByAlias(modelAlias);
        if (oldDataModel == null) {
            throw new KylinException(MODEL_NAME_NOT_EXIST, modelAlias);
        }
        EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
            NDataModel model = getManager(NDataModelManager.class, project).getDataModelDesc(oldDataModel.getUuid());
            if (model == null) {
                throw new KylinException(MODEL_NAME_NOT_EXIST, modelAlias);
            }
            checkModelPermission(project, oldDataModel.getId());
            PartitionDesc partitionColumn = modelParatitionDescRequest.getPartitionDesc();
            if (partitionColumn != null) {
                String rootFactTable = oldDataModel.getRootFactTableName().split("\\.")[1];
                if (!StringUtils.startsWithIgnoreCase(partitionColumn.getPartitionDateColumn(), rootFactTable + ".")) {
                    throw new KylinException(INVALID_PARTITION_COLUMN, MsgPicker.getMsg().getInvalidPartitionColumn());
                }
            }
            getManager(NDataModelManager.class, project).updateDataModel(oldDataModel.getUuid(),
                    copyForWrite -> copyForWrite.setPartitionDesc(modelParatitionDescRequest.getPartitionDesc()));
            semanticUpdater.handleSemanticUpdate(project, oldDataModel.getUuid(), oldDataModel,
                    modelParatitionDescRequest.getStart(), modelParatitionDescRequest.getEnd());
            return null;

        }, project);
    }

    @Transaction(project = 0)
    public void updateModelOwner(String project, String modelId, OwnerChangeRequest ownerChangeRequest) {
        try {
            aclEvaluate.checkProjectAdminPermission(project);
            checkTargetOwnerPermission(project, modelId, ownerChangeRequest.getOwner());
        } catch (AccessDeniedException e) {
            throw new KylinException(PERMISSION_DENIED, MsgPicker.getMsg().getModelChangePermission());
        } catch (IOException e) {
            throw new KylinException(PERMISSION_DENIED, MsgPicker.getMsg().getOwnerChangeError());
        }

        getManager(NDataModelManager.class, project).updateDataModel(modelId,
                copyForWrite -> copyForWrite.setOwner(ownerChangeRequest.getOwner()));
    }

    private void checkTargetOwnerPermission(String project, String modelId, String owner) throws IOException {
        Set<String> projectManagementUsers = accessService.getProjectManagementUsers(project);
        val model = getManager(NDataModelManager.class, project).getDataModelDesc(modelId);
        if (Objects.isNull(model) || model.isBroken()) {
            throw new KylinException(MODEL_ID_NOT_EXIST, modelId);
        }
        projectManagementUsers.remove(model.getOwner());

        if (CollectionUtils.isEmpty(projectManagementUsers) || !projectManagementUsers.contains(owner)) {
            Message msg = MsgPicker.getMsg();
            throw new KylinException(PERMISSION_DENIED, msg.getModelOwnerChangeInvalidUser());
        }
    }

    @Transaction(project = 0)
    public void updateMultiPartitionMapping(String project, String modelId, MultiPartitionMappingRequest request) {
        aclEvaluate.checkProjectWritePermission(project);
        val model = checkModelIsMLP(modelId, project);
        checkModelPermission(project, modelId);
        val multiPartitionDesc = model.getMultiPartitionDesc();
        val msg = MsgPicker.getMsg();
        if (CollectionUtils.isNotEmpty(request.getPartitionCols())) {
            // check size
            Preconditions.checkArgument(request.getAliasCols().size() == multiPartitionDesc.getColumns().size(),
                    new KylinException(INVALID_MULTI_PARTITION_MAPPING_REQUEST,
                            msg.getMultiPartitionMappingReqeustNotValid()));

            for (int i = 0; i < request.getPartitionCols().size(); i++) {
                val partitionCol = request.getPartitionCols().get(i);
                val columnRef = model.findColumn(partitionCol);
                if (!columnRef.equals(multiPartitionDesc.getColumnRefs().get(i))) {
                    throw new KylinException(INVALID_MULTI_PARTITION_MAPPING_REQUEST,
                            msg.getMultiPartitionMappingReqeustNotValid());
                }
            }

            val partitionValues = request.getValueMapping().stream()
                    .map(pair -> pair.getOrigin().toArray(new String[0])).collect(Collectors.toList());
            for (MultiPartitionDesc.PartitionInfo partitionInfo : multiPartitionDesc.getPartitions()) {
                if (partitionValues.stream().noneMatch(value -> Arrays.equals(value, partitionInfo.getValues()))) {
                    throw new KylinException(INVALID_MULTI_PARTITION_MAPPING_REQUEST,
                            msg.getMultiPartitionMappingReqeustNotValid());
                }
            }
        }

        getManager(NDataModelManager.class, project).updateDataModel(modelId,
                copyForWrite -> copyForWrite.setMultiPartitionKeyMapping(request.convertToMultiPartitionMapping()));
    }

    private void changeModelOwner(NDataModel model) {
        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
        if (Objects.nonNull(auth) && Objects.nonNull(auth.getPrincipal())) {
            if (auth.getPrincipal() instanceof UserDetails) {
                model.setOwner(((UserDetails) auth.getPrincipal()).getUsername());
            } else {
                model.setOwner(auth.getPrincipal().toString());
            }
        }
    }

    public List<NDataModel> updateResponseAcl(List<NDataModel> models, String project) {
        Set<String> groups = getCurrentUserGroups();
        if (AclPermissionUtil.hasProjectAdminPermission(project, groups)) {
            models.forEach(model -> {
                NDataModelAclParams aclParams = new NDataModelAclParams();
                aclParams.setUnauthorizedTables(Sets.newHashSet());
                aclParams.setUnauthorizedColumns(Sets.newHashSet());
                if (model instanceof NDataModelResponse) {
                    ((NDataModelResponse) model).setAclParams(aclParams);
                } else if (model instanceof RelatedModelResponse) {
                    ((RelatedModelResponse) model).setAclParams(aclParams);
                }
            });
            return models;
        }
        String username = AclPermissionUtil.getCurrentUsername();
        Set<String> allAuthTables = Sets.newHashSet();
        Set<String> allAuthColumns = Sets.newHashSet();
        var auths = getManager(AclTCRManager.class, project).getAuthTablesAndColumns(project, username, true);
        allAuthTables.addAll(auths.getTables());
        allAuthColumns.addAll(auths.getColumns());

        for (val group : groups) {
            auths = getManager(AclTCRManager.class, project).getAuthTablesAndColumns(project, group, false);
            allAuthTables.addAll(auths.getTables());
            allAuthColumns.addAll(auths.getColumns());
        }
        List<NDataModel> normalModel = new ArrayList<>();
        List<NDataModel> noVisibleModel = new ArrayList<>();
        models.forEach(model -> {
            Set<String> tablesNoAcl = Sets.newHashSet();
            Set<String> columnsNoAcl = Sets.newHashSet();
            Set<String> tables = Sets.newHashSet();
            model.getJoinTables().forEach(table -> tables.add(table.getTable()));
            tables.add(model.getRootFactTableName());
            tables.stream().filter(table -> !allAuthTables.contains(table)).forEach(tablesNoAcl::add);
            tables.stream().filter(allAuthTables::contains).forEach(table -> {
                ColumnDesc[] columnDescs = NTableMetadataManager.getInstance(getConfig(), project).getTableDesc(table)
                        .getColumns();
                Arrays.stream(columnDescs).map(column -> table + "." + column.getName())
                        .filter(column -> !allAuthColumns.contains(column)).forEach(columnsNoAcl::add);
            });
            NDataModelAclParams aclParams = new NDataModelAclParams();
            aclParams.setUnauthorizedTables(tablesNoAcl);
            aclParams.setUnauthorizedColumns(columnsNoAcl);
            if (model instanceof NDataModelResponse) {
                ((NDataModelResponse) model).setAclParams(aclParams);
            } else if (model instanceof RelatedModelResponse) {
                ((RelatedModelResponse) model).setAclParams(aclParams);
            }
            (aclParams.isVisible() ? normalModel : noVisibleModel).add(model);
        });
        List<NDataModel> result = new ArrayList<>(normalModel);
        result.addAll(noVisibleModel);
        return result;
    }

    public NDataModel updateResponseAcl(NDataModel model, String project) {
        List<NDataModel> models = updateResponseAcl(Collections.singletonList(model), project);
        return models.get(0);
    }

    public void checkDuplicateAliasInModelRequests(Collection<ModelRequest> modelRequests) {
        Map<String, List<ModelRequest>> aliasModelRequestMap = modelRequests.stream().collect(
                groupingBy(modelRequest -> modelRequest.getAlias().toLowerCase(Locale.ROOT), Collectors.toList()));
        aliasModelRequestMap.forEach((alias, requests) -> {
            if (requests.size() > 1) {
                throw new KylinException(INVALID_NAME,
                        String.format(Locale.ROOT, MsgPicker.getMsg().getModelAliasDuplicated(),
                                requests.stream().map(ModelRequest::getAlias).collect(Collectors.joining(", "))));
            }
        });
    }

    /**
     * Validate computed column type and throw errors to report wrongly typed computed columns.
     * Models migrated from 3x may have wrongly typed computed columns. see KE-11862
     *
     * @param modelId
     * @param project
     */
    public void validateCCType(String modelId, String project) {
        StringBuilder errorSb = new StringBuilder();
        NDataModel model = getManager(NDataModelManager.class, project).getDataModelDesc(modelId);
        try {
            List<ComputedColumnDesc> ccListCopy = Lists.newArrayList();
            for (ComputedColumnDesc computedColumnDesc : model.getComputedColumnDescs()) {
                // evaluate on a computedcolumn copy to avoid changing the metadata
                ccListCopy.add(JsonUtil.deepCopy(computedColumnDesc, ComputedColumnDesc.class));
            }
            ComputedColumnEvalUtil.evalDataTypeOfCCInBatch(model, ccListCopy);
            for (int n = 0; n < ccListCopy.size(); n++) {
                ComputedColumnDesc cc = model.getComputedColumnDescs().get(n);
                ComputedColumnDesc ccCopy = ccListCopy.get(n);

                if (!matchCCDataType(cc.getDatatype(), ccCopy.getDatatype())) {
                    errorSb.append(new MessageFormat(MsgPicker.getMsg().getCheckCCType(), Locale.ROOT)
                            .format(new String[] { cc.getFullName(), ccCopy.getDatatype(), cc.getDatatype() }));
                }
            }

        } catch (Exception e) {
            logger.error("Error validating computed column ", e);
        }

        if (errorSb.length() > 0) {
            throw new KylinException(INVALID_COMPUTED_COLUMN_EXPRESSION, errorSb.toString());
        }
    }

    private boolean matchCCDataType(String actual, String expected) {
        if (actual == null) {
            return false;
        }

        actual = actual.toUpperCase(Locale.ROOT);
        expected = expected.toUpperCase(Locale.ROOT);
        // char, varchar, string are of the same
        if (STRING_TYPE_SET.contains(actual)) {
            return STRING_TYPE_SET.contains(expected);
        }

        // if actual type is of decimal type with no prec, scala
        // match it with any decimal types
        if (actual.equals("DECIMAL")) {
            return expected.startsWith("DECIMAL");
        }

        return actual.equalsIgnoreCase(expected);
    }

    public CheckSegmentResponse checkSegments(String project, String modelAlias, String start, String end) {
        NDataModel model = getManager(NDataModelManager.class, project).getDataModelDescByAlias(modelAlias);
        if (model == null) {
            throw new KylinException(MODEL_NAME_NOT_EXIST, modelAlias);
        }
        if (model.isBroken()) {
            throw new KylinException(MODEL_BROKEN,
                    "Failed to get segment information as " + modelAlias + " is broken. Please fix it and try again.");
        }
        Segments<NDataSegment> segments = getSegmentsByRange(model.getUuid(), project, start, end);
        CheckSegmentResponse checkSegmentResponse = new CheckSegmentResponse();
        segments.forEach(seg -> checkSegmentResponse.getSegmentsOverlap()
                .add(new CheckSegmentResponse.SegmentInfo(seg.getId(), seg.getName())));
        return checkSegmentResponse;
    }

    public String getPartitionColumnFormatById(String project, String modelId) {
        NDataModel dataModelDesc = getManager(NDataModelManager.class, project).getDataModelDesc(modelId);
        if (dataModelDesc.isBroken())
            return null;
        return dataModelDesc.getPartitionDesc() == null ? null
                : dataModelDesc.getPartitionDesc().getPartitionDateFormat();
    }

    public String getPartitionColumnFormatByAlias(String project, String modelAlias) {
        NDataModel dataModelDesc = getManager(NDataModelManager.class, project).getDataModelDescByAlias(modelAlias);
        if (dataModelDesc.isBroken())
            return null;
        return dataModelDesc.getPartitionDesc() == null ? null
                : dataModelDesc.getPartitionDesc().getPartitionDateFormat();
    }

    public List<SegmentPartitionResponse> getSegmentPartitions(String project, String modelId, String segmentId,
            List<String> status, String sortBy, boolean reverse) {
        aclEvaluate.checkProjectReadPermission(project);
        checkModelPermission(project, modelId);
        val model = getModelById(modelId, project);
        val partitionDesc = model.getMultiPartitionDesc();
        val dataflow = getManager(NDataflowManager.class, project).getDataflow(modelId);
        val segment = dataflow.getSegment(segmentId);
        if (segment == null) {
            throw new KylinException(SEGMENT_NOT_EXIST_ID, segmentId);
        }

        Comparator<SegmentPartitionResponse> comparator = BasicService
                .propertyComparator(StringUtils.isEmpty(sortBy) ? "last_modified_time" : sortBy, !reverse);

        return segment.getMultiPartitions().stream() //
                .map(partition -> {
                    val partitionInfo = partitionDesc.getPartitionInfo(partition.getPartitionId());
                    val lastModifiedTime = partition.getLastBuildTime() != 0 ? partition.getLastBuildTime()
                            : partition.getCreateTimeUTC();
                    return new SegmentPartitionResponse(partitionInfo.getId(), partitionInfo.getValues(),
                            partition.getStatus(), lastModifiedTime, partition.getSourceCount(),
                            partition.getStorageSize());
                })
                .filter(partitionResponse -> CollectionUtils.isEmpty(status)
                        || status.contains(partitionResponse.getStatus().name()))
                .sorted(comparator).collect(Collectors.toList());
    }

    @Transaction(project = 0)
    public void deletePartitionsByValues(String project, String segmentId, String modelId,
            List<String[]> subPartitionValues) {
        NDataModel model = checkModelIsMLP(modelId, project);
        checkPartitionValues(model, subPartitionValues);
        Set<Long> ids = model.getMultiPartitionDesc().getPartitionIdsByValues(subPartitionValues);
        deletePartitions(project, segmentId, model.getId(), ids);
    }

    private void checkPartitionValues(NDataModel model, List<String[]> subPartitionValues) {
        List<String[]> modelPartitionValues = model.getMultiPartitionDesc().getPartitions().stream()
                .map(MultiPartitionDesc.PartitionInfo::getValues).collect(Collectors.toList());
        List<String[]> absentValues = MultiPartitionUtil.findAbsentValues(modelPartitionValues, subPartitionValues);
        if (!absentValues.isEmpty()) {
            throw new KylinException(INVALID_PARTITION_VALUES, String.format(Locale.ROOT,
                    MsgPicker.getMsg().getInvalidPartitionValue(),
                    absentValues.stream().map(arr -> String.join(", ", arr)).collect(Collectors.joining(", "))));
        }
    }

    @Transaction(project = 0)
    public void deletePartitions(String project, String segmentId, String modelId, Set<Long> partitions) {
        aclEvaluate.checkProjectOperationPermission(project);
        checkModelPermission(project, modelId);
        checkSegmentsExistById(modelId, project, new String[] { segmentId });
        checkModelIsMLP(modelId, project);
        if (CollectionUtils.isEmpty(partitions)) {
            return;
        }
        NDataflowManager dataflowManager = getManager(NDataflowManager.class, project);
        if (StringUtils.isNotEmpty(segmentId)) {
            // remove partition in target segment
            dataflowManager.removeLayoutPartition(modelId, partitions, Sets.newHashSet(segmentId));
            // remove partition in target segment
            dataflowManager.removeSegmentPartition(modelId, partitions, Sets.newHashSet(segmentId));
        } else {
            // remove partition in all layouts
            dataflowManager.removeLayoutPartition(modelId, Sets.newHashSet(partitions), null);
            // remove partition in all  segments
            dataflowManager.removeSegmentPartition(modelId, Sets.newHashSet(partitions), null);
            // remove partition in model
            getManager(NDataModelManager.class, project).updateDataModel(modelId, copyForWrite -> {
                val multiPartitionDesc = copyForWrite.getMultiPartitionDesc();
                multiPartitionDesc.removePartitionValue(Lists.newArrayList(partitions));
            });
        }
    }

    public NDataModel checkModelIsMLP(String modelId, String project) {
        NDataModel model = getModelById(modelId, project);
        if (!model.isMultiPartitionModel()) {
            throw new KylinException(INVALID_MODEL_TYPE,
                    String.format(Locale.ROOT, MsgPicker.getMsg().getModelIsNotMlp(), model.getAlias()));
        }
        return model;
    }

    public InvalidIndexesResponse detectInvalidIndexes(ModelRequest request) {
        String project = request.getProject();
        aclEvaluate.checkProjectReadPermission(project);

        String modelId = request.getId();
        EnhancedUnitOfWork.doInTransactionWithCheckAndRetry(() -> {
            if (modelId != null && getModelById(modelId, project).isBroken()) {
                discardInvalidDataForBrokenModel(modelId, project);
            }
            return null;
        }, project);

        request.setPartitionDesc(null);
        request.setMultiPartitionDesc(null);
        request.setMultiPartitionKeyMapping(null);
        NDataModel model = convertAndInitDataModel(request, project);

        String uuid = model.getUuid();
        List<JoinTableDesc> joinTables = model.getJoinTables();
        IndexPlan indexPlan = getManager(NIndexPlanManager.class, project).getIndexPlan(uuid);
        AntiFlatChecker checker = new AntiFlatChecker(joinTables, model);
        List<ComputedColumnDesc> invalidCCList = checker.getInvalidComputedColumns(model);
        Set<Integer> invalidDimensions = checker.getInvalidDimensions(model);
        Set<Integer> invalidMeasures = checker.getInvalidMeasures(model);
        Set<Integer> invalidScope = Sets.newHashSet();
        invalidScope.addAll(invalidDimensions);
        invalidScope.addAll(invalidMeasures);
        Set<Long> invalidIndexes = checker.getInvalidIndexes(indexPlan, invalidScope);
        AtomicInteger aggIndexCount = new AtomicInteger();
        AtomicInteger tableIndexCount = new AtomicInteger();
        invalidIndexes.forEach(layoutId -> {
            if (layoutId > IndexEntity.TABLE_INDEX_START_ID) {
                tableIndexCount.getAndIncrement();
            } else {
                aggIndexCount.getAndIncrement();
            }
        });

        List<String> invalidDimensionNames = model.getAllNamedColumns().stream()
                .filter(col -> invalidDimensions.contains(col.getId())).map(NDataModel.NamedColumn::getAliasDotColumn)
                .collect(Collectors.toList());
        List<String> invalidMeasureNames = model.getAllMeasures().stream()
                .filter(measure -> invalidMeasures.contains(measure.getId())).map(MeasureDesc::getName)
                .collect(Collectors.toList());

        InvalidIndexesResponse response = new InvalidIndexesResponse();
        response.setCcList(invalidCCList);
        response.setDimensions(invalidDimensionNames);
        response.setMeasures(invalidMeasureNames);
        response.setIndexes(Lists.newArrayList(invalidIndexes));
        response.setInvalidAggIndexCount(aggIndexCount.get());
        response.setInvalidTableIndexCount(tableIndexCount.get());
        response.setAntiFlattenLookups(Lists.newArrayList(checker.getAntiFlattenLookups()));
        return response;
    }

    public void validatePartitionDesc(PartitionDesc partitionDesc) {
        if (partitionDesc != null) {
            if (partitionDesc.isEmpty()) {
                throw new KylinException(INVALID_PARTITION_COLUMN, MsgPicker.getMsg().getPartitionColumnNotExist());
            }
            if (!isSupportFormats(partitionDesc)) {
                throw new KylinException(DATETIME_FORMAT_PARSE_ERROR, partitionDesc.getPartitionDateFormat());
            }
            if (partitionDesc.getPartitionDateFormat() != null && !partitionDesc.partitionColumnIsTimestamp()) {
                validateDateTimeFormatPattern(partitionDesc.getPartitionDateFormat());
            }
        }
    }

    public void validateDateTimeFormatPattern(String pattern) {
        if (pattern.isEmpty()) {
            throw new KylinException(DATETIME_FORMAT_EMPTY);
        }
        try {
            new SimpleDateFormat(pattern, Locale.getDefault(Locale.Category.FORMAT));
        } catch (IllegalArgumentException e) {
            throw new KylinException(DATETIME_FORMAT_PARSE_ERROR, e, pattern);
        }
    }

    private boolean isSupportFormats(PartitionDesc partitionDesc) {
        if (partitionDesc.partitionColumnIsTimestamp()) {
            return false;
        }
        String dateFormat = partitionDesc.getPartitionDateFormat();
        Matcher matcher = QUOTE_PATTERN.matcher(dateFormat);
        while (matcher.find()) {
            dateFormat = dateFormat.replaceAll(matcher.group(), "");
        }
        for (String frontEndFormat : SUPPORTED_FORMATS) {
            dateFormat = dateFormat.replaceAll(frontEndFormat, "");
        }
        int length = dateFormat.length();
        for (int i = 0; i < length; i++) {
            char c = dateFormat.charAt(i);
            if (c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z') {
                return false;
            }
        }
        return true;
    }

    private void discardInvalidDataForBrokenModel(String modelId, String project) {
        NDataModelManager modelManager = getManager(NDataModelManager.class, project);
        NDataModel brokenModel = modelQuerySupporter.getBrokenModel(project, modelId);
        NDataModel copyForWrite = modelManager.copyForWrite(brokenModel);

        semanticUpdater.discardInvalidColsAndMeasForBrokenModel(project, copyForWrite);

        modelManager.updateDataBrokenModelDesc(copyForWrite);
    }

    private NDataModel convertAndInitDataModel(ModelRequest request, String project) {
        NDataModel model = convertToDataModel(request);
        semanticUpdater.discardInvalidColsAndMeasForBrokenModel(project, model);
        model.init(KylinConfig.getInstanceFromEnv(), project,
                getManager(NDataflowManager.class, project).listUnderliningDataModels());
        for (ComputedColumnDesc cc : model.getComputedColumnDescs()) {
            String innerExp = cc.getInnerExpression();
            if (cc.getExpression().equalsIgnoreCase(innerExp)) {
                innerExp = PushDownUtil.massageComputedColumn(model, project, cc, null);
            }
            cc.setInnerExpression(innerExp);
        }
        return model;
    }

    private void setRequest(ModelRequest request, NDataModel model, AffectedModelContext removeAffectedModel,
            AffectedModelContext changeTypeAffectedModel, String projectName) {
        request.setSimplifiedMeasures(model.getEffectiveMeasures().values().stream()
                .filter(m -> !removeAffectedModel.getMeasures().contains(m.getId())
                        && !changeTypeAffectedModel.getMeasures().contains(m.getId()))
                .map(SimplifiedMeasure::fromMeasure).collect(Collectors.toList()));

        request.setSimplifiedMeasures(request.getSimplifiedMeasures().stream().map(simplifiedMeasure -> {
            int measureId = simplifiedMeasure.getId();
            NDataModel.Measure updateMeasure = changeTypeAffectedModel.getUpdateMeasureMap().get(measureId);
            if (updateMeasure != null) {
                simplifiedMeasure = SimplifiedMeasure.fromMeasure(updateMeasure);
            }
            return simplifiedMeasure;
        }).collect(Collectors.toList()));

        request.setSimplifiedDimensions(model.getAllNamedColumns().stream()
                .filter(col -> !removeAffectedModel.getDimensions().contains(col.getId()) && col.isDimension())
                .collect(Collectors.toList()));
        request.setComputedColumnDescs(model.getComputedColumnDescs().stream()
                .filter(cc -> !removeAffectedModel.getComputedColumns().contains(cc.getColumnName()))
                .collect(Collectors.toList()));

        // clean
        request.getAllNamedColumns().stream() //
                .filter(col -> removeAffectedModel.getColumns().contains(col.getId())) //
                .forEach(col -> col.setStatus(NDataModel.ColumnStatus.TOMB));
        String factTableAlias = model.getRootFactTable().getAlias();
        Set<String> ccFullNameSet = removeAffectedModel.getComputedColumns().stream() //
                .map(ccName -> factTableAlias + "." + ccName) //
                .collect(Collectors.toSet());
        request.getAllNamedColumns().stream() //
                .filter(col -> ccFullNameSet.contains(col.getAliasDotColumn())) //
                .forEach(col -> col.setStatus(NDataModel.ColumnStatus.TOMB));

        request.setProject(projectName);
    }

    public void checkCCEmpty(ModelRequest modelRequest) {
        List<ComputedColumnDesc> ccList = modelRequest.getComputedColumnDescs();
        if (CollectionUtils.isEmpty(ccList)) {
            return;
        }
        boolean matchEmpty = ccList.stream()
                .anyMatch(cc -> StringUtils.isEmpty(cc.getColumnName()) || StringUtils.isEmpty(cc.getExpression()));
        if (matchEmpty) {
            throw new KylinException(COMPUTED_COLUMN_NAME_OR_EXPR_EMPTY);
        }
    }

    public void checkSegmentOverlap(NDataModelResponse model, List<NDataSegmentResponse> segments) {
        if (ModelStatusToDisplayEnum.BROKEN != model.getStatus()) {
            boolean hasAnyOverlapSegment = segments.stream()
                    .anyMatch(seg -> SegmentStatusEnumToDisplay.OVERLAP == seg.getStatusToDisplay());
            if (hasAnyOverlapSegment) {
                logger.warn("model {} segments overlap found, mark as broken!", model.getId());
                model.setBroken(true);
                model.setBrokenReason(NDataModel.BrokenReason.SEGMENT_OVERLAP);
                model.setStatus(ModelStatusToDisplayEnum.BROKEN);
            }
        }
    }

    public Pair<ModelRequest, ComputedColumnConflictResponse> checkCCConflict(ModelRequest modelRequest) {
        String project = modelRequest.getProject();
        validatePartitionDateColumn(modelRequest);

        val dataModel = semanticUpdater.convertToDataModel(modelRequest);
        val modelManager = getManager(NDataModelManager.class, project);
        val ccRelatedModels = modelManager.getCCRelatedModels(dataModel);
        // check cc conflict and return ccConflictInfo
        val ccConflictInfo = dataModel.checkCCFailAtEnd(getConfig(), project, ccRelatedModels, true);
        boolean autoAdjust = modelRequest.isComputedColumnNameAutoAdjust();

        if (ccConflictInfo.noneConflict()) {
            // No conflict, return
            return Pair.newPair(modelRequest, new ComputedColumnConflictResponse());
        }
        if (ccConflictInfo.hasSameNameConflict()) {
            // have sameNameDiffExpr Conflict, all conflict messages need to be thrown
            val response = handleOnConflictResponse(ccConflictInfo.getAllConflictException());
            throw new KylinException(COMPUTED_COLUMN_CONFLICT).withData(response);
        }
        // have sameExprDiffExprConflict Conflict
        if (!autoAdjust) {
            // AutoAdjust = false
            // SameExpr conflict messages need to be thrown
            val response = handleOnConflictResponse(ccConflictInfo.getSameExprConflictException());
            throw new KylinException(COMPUTED_COLUMN_CONFLICT).withData(response);
        }
        // AutoAdjust = true
        List<ComputedColumnDesc> inputCCDescList = Lists.newArrayList(modelRequest.getComputedColumnDescs());
        // deal with conflicts
        val pair = ccConflictInfo.getAdjustedCCList(inputCCDescList);
        val adjustExceptions = pair.getSecond().stream() //
                .map(ComputedColumnUtil.CCConflictDetail::getAdjustKylinException).collect(Collectors.toList());
        ModelRequest resultModelRequest = adjustModelRequestCCName(modelRequest, pair);

        return Pair.newPair(resultModelRequest, handleOnConflictResponse(adjustExceptions));
    }

    public ModelRequest adjustModelRequestCCName(ModelRequest modelRequest,
            Pair<List<ComputedColumnDesc>, List<ComputedColumnUtil.CCConflictDetail>> pair) {
        val adjustDetails = pair.getSecond();
        // adjust cc name
        modelRequest.setComputedColumnDescs(pair.getFirst());

        val dimensions = modelRequest.getSimplifiedDimensions();
        val measures = modelRequest.getSimplifiedMeasures();
        for (val detail : adjustDetails) {
            String newCCFullName = detail.getNewCC().getFullName();
            String existingCCFullName = detail.getExistingCC().getFullName();

            // adjust dimensions
            dimensions.stream() //
                    .filter(NDataModel.NamedColumn::isExist) //
                    // column equals
                    .filter(d -> StringUtils.equalsIgnoreCase(d.getAliasDotColumn(), newCCFullName))
                    .forEach(d -> d.setAliasDotColumn(existingCCFullName));

            // adjust measures
            measures.forEach(m -> m.getParameterValue().stream() //
                    // type = column
                    .filter(pr -> StringUtils.equalsIgnoreCase(pr.getType(), PARAMETER_TYPE_COLUMN))
                    // value equals
                    .filter(pr -> StringUtils.equalsIgnoreCase(pr.getValue(), newCCFullName))
                    .forEach(pr -> pr.setValue(existingCCFullName)));
        }

        // adjust filter condition
        String filterCondition = modelRequest.getFilterCondition();
        if (StringUtils.isEmpty(filterCondition)) {
            return modelRequest;
        }
        for (val detail : adjustDetails) {
            String newCCFullName = detail.getNewCC().getFullName();
            String existingCCFullName = detail.getExistingCC().getFullName();
            if (StringUtils.containsIgnoreCase(filterCondition, newCCFullName)) {
                filterCondition = replaceAllIgnoreCase(filterCondition, newCCFullName, existingCCFullName);
            }
        }
        modelRequest.setFilterCondition(filterCondition);
        return modelRequest;
    }

    public String replaceAllIgnoreCase(String input, String regex, String replacement) {
        return Pattern.compile(regex, Pattern.CASE_INSENSITIVE).matcher(input).replaceAll(replacement);
    }

    public ComputedColumnConflictResponse handleOnConflictResponse(List<KylinException> exceptionList) {
        val response = new ComputedColumnConflictResponse();
        exceptionList.stream() //
                .filter(Objects::nonNull) //
                .forEach(e -> {
                    val producer = e.getErrorCodeProducer();
                    val code = producer.getErrorCode().getCode();
                    val msg = producer.getMsg(e.getArgs());
                    response.addConflictDetail(code, msg);
                });
        return response;
    }

    @Override
    public void onUpdateBrokenModel(NDataModel model, AffectedModelContext removeAffectedModel,
            AffectedModelContext changeTypeAffectedModel, String projectName) throws Exception {
        val request = new ModelRequest(JsonUtil.deepCopy(model, NDataModel.class));
        setRequest(request, model, removeAffectedModel, changeTypeAffectedModel, projectName);
        updateBrokenModel(projectName, request, removeAffectedModel.getColumns());
    }

    @Override
    public void onUpdateDataModel(NDataModel model, AffectedModelContext removeAffectedModel,
            AffectedModelContext changeTypeAffectedModel, String projectName, TableDesc tableDesc) throws Exception {
        val request = new ModelRequest(JsonUtil.deepCopy(model, NDataModel.class));
        setRequest(request, model, removeAffectedModel, changeTypeAffectedModel, projectName);
        request.setColumnsFetcher((tableRef, isFilterCC) -> TableRef.filterColumns(
                tableRef.getIdentity().equals(tableDesc.getIdentity()) ? tableDesc : tableRef, isFilterCC));
        updateDataModelSemantic(projectName, request, false);
    }

    @Override
    public NDataModel onGetModelByAlias(String modelAlias, String project) {
        return getModelByAlias(modelAlias, project);
    }

    @Override
    public NDataModel onGetModelById(String modelId, String project) {
        return getModelById(modelId, project);
    }

    @Override
    public void onModelUpdate(String project, String modelId) {
        modelChangeSupporters.forEach(listener -> listener.onUpdate(project, modelId));
    }

    @Override
    public void onUpdateSCD2ModelStatus(String project, RealizationStatusEnum status) {
        updateSCD2ModelStatusInProjectById(project, (ModelStatusToDisplayEnum.convert(status)));
    }

    @Override
    public void onOfflineMultiPartitionModels(String project) {
        offlineMultiPartitionModelsInProject(project);
    }

}
