package com.purui.prims.headless.server.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.purui.prims.auth.api.authentication.pojo.User;
import com.purui.prims.common.jsqlparser.SqlReplaceHelper;
import com.purui.prims.common.jsqlparser.SqlSelectHelper;
import com.purui.prims.common.pojo.Aggregator;
import com.purui.prims.common.pojo.Castor;
import com.purui.prims.common.pojo.Constants;
import com.purui.prims.common.pojo.DataEvent;
import com.purui.prims.common.pojo.DataItem;
import com.purui.prims.common.pojo.DateConf;
import com.purui.prims.common.pojo.Filter;
import com.purui.prims.common.pojo.PersistenceConf;
import com.purui.prims.common.pojo.enums.AggOperatorEnum;
import com.purui.prims.common.pojo.enums.AuthType;
import com.purui.prims.common.pojo.enums.DataFormatTypeEnum;
import com.purui.prims.common.pojo.enums.EventType;
import com.purui.prims.common.pojo.enums.PublishEnum;
import com.purui.prims.common.pojo.enums.StatusEnum;
import com.purui.prims.common.pojo.enums.TypeEnums;
import com.purui.prims.common.util.BeanMapper;
import com.purui.prims.common.util.ContextUtils;
import com.purui.prims.common.util.DateModeUtils;
import com.purui.prims.common.util.DateUtils;
import com.purui.prims.dscheduler.model.TaskInstance;
import com.purui.prims.dscheduler.model.TaskLog;
import com.purui.prims.headless.api.pojo.Collect;
import com.purui.prims.headless.api.pojo.DrillDownDimension;
import com.purui.prims.headless.api.pojo.MeasureParam;
import com.purui.prims.headless.api.pojo.MetricParam;
import com.purui.prims.headless.api.pojo.MetricQueryDefaultConfig;
import com.purui.prims.headless.api.pojo.ModelDetail;
import com.purui.prims.headless.api.pojo.SchemaElementMatch;
import com.purui.prims.headless.api.pojo.SchemaElementType;
import com.purui.prims.headless.api.pojo.SchemaItem;
import com.purui.prims.headless.api.pojo.enums.MapModeEnum;
import com.purui.prims.headless.api.pojo.enums.MetricDefineType;
import com.purui.prims.headless.api.pojo.request.MetaBatchReq;
import com.purui.prims.headless.api.pojo.request.MetricBaseReq;
import com.purui.prims.headless.api.pojo.request.MetricReq;
import com.purui.prims.headless.api.pojo.request.PageMetricReq;
import com.purui.prims.headless.api.pojo.request.QueryMapReq;
import com.purui.prims.headless.api.pojo.request.QueryMetricReq;
import com.purui.prims.headless.api.pojo.request.QuerySqlReq;
import com.purui.prims.headless.api.pojo.request.QueryStructReq;
import com.purui.prims.headless.api.pojo.request.SemanticQueryReq;
import com.purui.prims.headless.api.pojo.response.DataSetMapInfo;
import com.purui.prims.headless.api.pojo.response.DataSetResp;
import com.purui.prims.headless.api.pojo.response.DatabaseResp;
import com.purui.prims.headless.api.pojo.response.DimensionResp;
import com.purui.prims.headless.api.pojo.response.MapInfoResp;
import com.purui.prims.headless.api.pojo.response.MetricPersistenceConfigResp;
import com.purui.prims.headless.api.pojo.response.MetricResp;
import com.purui.prims.headless.api.pojo.response.ModelResp;
import com.purui.prims.headless.api.pojo.response.RoleAuthResp;
import com.purui.prims.headless.core.pojo.QueryStatement;
import com.purui.prims.headless.core.translator.SemanticTranslator;
import com.purui.prims.headless.server.manager.MetricYamlManager;
import com.purui.prims.headless.server.persistence.dataobject.MetricDO;
import com.purui.prims.headless.server.persistence.dataobject.MetricQueryDefaultConfigDO;
import com.purui.prims.headless.server.persistence.mapper.MetricDOMapper;
import com.purui.prims.headless.server.persistence.repository.MetricRepository;
import com.purui.prims.headless.server.pojo.DimensionsFilter;
import com.purui.prims.headless.server.pojo.MetaFilter;
import com.purui.prims.headless.server.pojo.MetricFilter;
import com.purui.prims.headless.server.pojo.MetricLineage;
import com.purui.prims.headless.server.pojo.MetricVersion;
import com.purui.prims.headless.server.pojo.MetricsFilter;
import com.purui.prims.headless.server.pojo.ModelCluster;
import com.purui.prims.headless.server.pojo.ModelFilter;
import com.purui.prims.headless.server.pojo.yaml.MetricYamlTpl;
import com.purui.prims.headless.server.service.ChatLayerService;
import com.purui.prims.headless.server.service.CollectService;
import com.purui.prims.headless.server.service.DataSetService;
import com.purui.prims.headless.server.service.DimensionService;
import com.purui.prims.headless.server.service.MetricLineageService;
import com.purui.prims.headless.server.service.MetricPersistenceConfigService;
import com.purui.prims.headless.server.service.MetricService;
import com.purui.prims.headless.server.service.MetricVersionService;
import com.purui.prims.headless.server.service.ModelService;
import com.purui.prims.headless.server.service.RoleAuthService;
import com.purui.prims.headless.server.service.SchedulerService;
import com.purui.prims.headless.server.utils.DorisUtils;
import com.purui.prims.headless.server.utils.MetricCheckUtils;
import com.purui.prims.headless.server.utils.MetricConverter;
import com.purui.prims.headless.server.utils.ModelClusterBuilder;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.expression.Expression;
import net.sf.jsqlparser.expression.Function;
import net.sf.jsqlparser.expression.LongValue;
import net.sf.jsqlparser.expression.StringValue;
import net.sf.jsqlparser.expression.operators.relational.ExpressionList;
import net.sf.jsqlparser.statement.select.FromItem;
import net.sf.jsqlparser.statement.select.ParenthesedSelect;
import net.sf.jsqlparser.statement.select.PlainSelect;
import net.sf.jsqlparser.statement.select.SelectItem;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

@Service
@Slf4j
public class MetricServiceImpl extends ServiceImpl<MetricDOMapper, MetricDO> implements MetricService {

    private final MetricRepository metricRepository;

    private final ModelService modelService;

    private final DimensionService dimensionService;

    private final CollectService collectService;

    private final DataSetService dataSetService;

    private final ApplicationEventPublisher eventPublisher;

    private final ChatLayerService chatQueryService;

    private final MetricVersionService metricVersionService;

    public MetricServiceImpl(MetricRepository metricRepository, ModelService modelService,
                             CollectService collectService, DataSetService dataSetService, ApplicationEventPublisher eventPublisher,
                             DimensionService dimensionService, MetricVersionService metricVersionService,
                             @Lazy ChatLayerService chatQueryService) {
        this.metricRepository = metricRepository;
        this.modelService = modelService;
        this.eventPublisher = eventPublisher;
        this.collectService = collectService;
        this.dataSetService = dataSetService;
        this.dimensionService = dimensionService;
        this.chatQueryService = chatQueryService;
        this.metricVersionService = metricVersionService;
    }

    @Override
    public MetricResp createMetric(MetricReq metricReq, User user) {
        checkExist(Lists.newArrayList(metricReq));
        MetricCheckUtils.checkParam(metricReq);
        metricReq.createdBy(user.getUserName());
        MetricDO metricDO = MetricConverter.convert2MetricDO(metricReq);
        metricRepository.createMetric(metricDO);
        sendEventBatch(Lists.newArrayList(metricDO), EventType.ADD);
        return MetricConverter.convert2MetricResp(metricDO);
    }

    @Override
    public void createMetricBatch(List<MetricReq> metricReqs, User user) {
        if (CollectionUtils.isEmpty(metricReqs)) {
            return;
        }
        Long modelId = metricReqs.get(0).getModelId();
        List<MetricResp> metricResps = getMetrics(new MetaFilter(Lists.newArrayList(modelId)));
        Map<String, MetricResp> bizNameMap = metricResps.stream().collect(Collectors.toMap(MetricResp::getBizName, a -> a, (k1, k2) -> k1));
        Map<String, MetricResp> nameMap = metricResps.stream().collect(Collectors.toMap(MetricResp::getName, a -> a, (k1, k2) -> k1));
        List<MetricReq> metricToInsert =
            metricReqs.stream().filter(metric -> !bizNameMap.containsKey(metric.getBizName()) && !nameMap.containsKey(metric.getName()))
                      .collect(Collectors.toList());
        if (CollectionUtils.isEmpty(metricToInsert)) {
            return;
        }
        List<MetricDO> metricDOS =
            metricToInsert.stream().peek(metric -> metric.createdBy(user.getUserName())).map(MetricConverter::convert2MetricDO)
                          .collect(Collectors.toList());
        metricRepository.createMetricBatch(metricDOS);
        sendEventBatch(metricDOS, EventType.ADD);
    }

    @Override
    public MetricResp updateMetric(MetricReq metricReq, User user) {
        MetricCheckUtils.checkParam(metricReq);
        checkExist(Lists.newArrayList(metricReq));
        metricReq.updatedBy(user.getUserName());
        MetricDO metricDO = metricRepository.getMetricById(metricReq.getId());
        String oldName = metricDO.getName();
        Integer oldPublishStatus = metricDO.getIsPublish();
        MetricConverter.convert(metricDO, metricReq);
        metricDO.setIsPublish(PublishEnum.UN_PUBLISHED.getCode());
        metricRepository.updateMetric(metricDO);
        // 如果修改前是发布状态，则创建新版本
        if (oldPublishStatus == PublishEnum.PUBLISHED.getCode()) {
            // new version
            metricVersionService.createNewVersion(metricDO.getId());
        }

        if (!oldName.equals(metricDO.getName())) {
            DataItem dataItem = getDataItem(metricDO);
            dataItem.setName(oldName);
            dataItem.setNewName(metricDO.getName());
            sendEvent(dataItem, EventType.UPDATE);
        }
        return MetricConverter.convert2MetricResp(metricDO);
    }

    @Override
    public void batchUpdateStatus(MetaBatchReq metaBatchReq, User user) {
        if (CollectionUtils.isEmpty(metaBatchReq.getIds())) {
            return;
        }
        List<MetricDO> metricDOS = getMetrics(metaBatchReq.getIds());
        if (CollectionUtils.isEmpty(metricDOS)) {
            return;
        }
        metricDOS = metricDOS.stream().peek(metricDO -> {
            metricDO.setStatus(metaBatchReq.getStatus());
            metricDO.setUpdatedAt(new Date());
            metricDO.setUpdatedBy(user.getUserName());
        }).collect(Collectors.toList());
        metricRepository.batchUpdateStatus(metricDOS);
        if (StatusEnum.OFFLINE.getCode().equals(metaBatchReq.getStatus()) || StatusEnum.DELETED.getCode()
                                                                                               .equals(metaBatchReq.getStatus())) {
            sendEventBatch(metricDOS, EventType.DELETE);
        } else if (StatusEnum.ONLINE.getCode().equals(metaBatchReq.getStatus())) {
            sendEventBatch(metricDOS, EventType.ADD);
        }
    }

    @Transactional
    @Override
    public void batchPublish(List<Long> metricIds, User user) {
        List<MetricDO> metrics = getMetrics(metricIds);
        for (MetricDO metricDO : metrics) {
            metricDO.setUpdatedAt(new Date());
            metricDO.setUpdatedBy(user.getUserName());
            if (StatusEnum.ONLINE.getCode().equals(metricDO.getIsPublish())) {
                // 已经发布
                continue;
            }
            // 物化加速
            // createMetricView(metricDO);
            // 创建版本
            metricDO.setIsPublish(StatusEnum.ONLINE.getCode());
        }
        metricRepository.batchPublish(metrics);
    }

    @Override
    public void batchUnPublish(List<Long> metricIds, User user) {
        List<MetricDO> metrics = getMetrics(metricIds);
        for (MetricDO metricDO : metrics) {
            metricDO.setUpdatedAt(new Date());
            metricDO.setUpdatedBy(user.getUserName());
            metricDO.setIsPublish(StatusEnum.INITIALIZED.getCode());
            // 取消加速
            // deleteMetricView(metricDO);
            // 删除已发布的版本号
            metricVersionService.restorePreviousVersion(metricDO.getId());
        }
        metricRepository.batchUnPublish(metrics);
    }

    /**
     * 指标物化
     */
    @Async
    public boolean createMetricView(MetricDO metricDO) {
        DatabaseResp databaseResp = modelService.getDatabaseByModelId(metricDO.getModelId());
        if (databaseResp == null) {
            return false;
        }
        ModelResp modelResp = modelService.getModel(metricDO.getModelId());
        ModelDetail modelDetail = modelResp.getModelDetail();
        // 模型有维度，则创建视图
        if (modelDetail != null && StringUtils.isNotEmpty(metricDO.getRelateDimensions())) {
            String viewName = DorisUtils.getIndexViewName(metricDO.getBizName());
            try {
                // 单表模型
                if (StringUtils.isNotEmpty(modelDetail.getTableQuery())) {
                    DorisUtils.createView(viewName, buildMetricSql(modelResp, metricDO), databaseResp);
                }
                // 多表关联模型
                if (StringUtils.isNotEmpty(modelDetail.getSqlQuery())) {
                    // 如果第一个主表有分区字段时，按此字段分区 // todo: 获取主表的分区字段
                    String partitionKey = "";
                    DorisUtils.accelerateModelOrDataset(viewName, partitionKey, modelDetail.getSqlQuery(), databaseResp);
                }
                return true;
            } catch (Exception e) {
                // 模型不支持加速
                log.error("指标加速失败", e.getMessage());
            }
        }
        return false;
    }

    /**
     * 删除指标物化
     *
     * @param metricDO
     * @return
     */
    @Async
    public boolean deleteMetricView(MetricDO metricDO) {
        DatabaseResp databaseResp = modelService.getDatabaseByModelId(metricDO.getModelId());
        if (databaseResp == null) {
            return false;
        }
        ModelResp modelResp = modelService.getModel(metricDO.getModelId());
        ModelDetail modelDetail = modelResp.getModelDetail();
        if (modelDetail != null) {
            String tableName = modelDetail.getTableQuery();
            if (StringUtils.isNotEmpty(tableName) && tableName.startsWith("select")) {
                // 如果是SQL语句
                tableName = SqlSelectHelper.getTableName(modelDetail.getTableQuery());
            }
            String viewName = DorisUtils.getIndexViewName(metricDO.getBizName());
            DorisUtils.deleteView(viewName, tableName, databaseResp);
            return true;
        }
        return false;
    }

    private String buildMetricSql(ModelResp modelResp, MetricDO metricDO) {
        return buildMetricSql(modelResp, Lists.newArrayList(MetricConverter.convert2MetricResp(metricDO)), true, null, false);
    }

    @Override
    public String buildMetricSql(ModelResp modelResp, List<MetricResp> metricDOs, boolean simplify, PersistenceConf conf,
                                 boolean isForQuery) {
        List<Long> metricIds = metricDOs.stream().map(metricDO -> metricDO.getId()).collect(Collectors.toList());
        List<Long> dimensionIds = Lists.newArrayList();
        metricDOs.forEach(metricDO -> {
            // 增加关联维度
            if (metricDO.getRelateDimension() != null) {
                dimensionIds.addAll(metricDO.getRelateDimension().getDrillDownDimensions().stream()
                                            .map(drillDownDimension -> drillDownDimension.getDimensionId()).collect(Collectors.toList()));
            }
        });
        return buildMetricSql(modelResp.getDomainId(), modelResp.getModelDetail().getTableQuery(), dimensionIds, metricIds, simplify,
            conf, isForQuery);
    }

    private String buildMetricSql(Long domainId, String tableName, List<Long> dimensionIds, List<Long> metricIds, boolean simplify,
                                  PersistenceConf conf, boolean isForQuery) {
        QueryMetricReq metricReq = new QueryMetricReq();
        metricReq.setMetricIds(metricIds);
        metricReq.setDomainId(domainId);
        // 物化视图&全量，不需要时间条件
        metricReq.getDateInfo().setDateMode(DateConf.DateMode.IGNORE);
        if (!simplify && conf != null && "inc".equalsIgnoreCase(conf.getWay())) {
            // 是否增量更新，增量更新的条件（天，小时）sql中的where条件
            metricReq.getDateInfo().setDateMode(DateConf.DateMode.RECENT);
            metricReq.getDateInfo().setUnit(conf.getUnit());
            metricReq.getDateInfo().setStartDate(conf.getStartDate());
            metricReq.getDateInfo().setEndDate(conf.getEndDate());
            metricReq.getDateInfo().setPeriod(conf.getPeriod());
        }
        MetaFilter metaFilter = new MetaFilter();
        metaFilter.setIds(dimensionIds);
        metaFilter.setDomainId(domainId);
        List<DimensionResp> dims = dimensionService.getDimensions(metaFilter);
        // 返回原始数据(不转换百分比)
        metricReq.setOriginalData(true);
        QueryStructReq structReq = this.convert(metricReq);

        List<String> groups = dims.stream().filter(d -> !d.isTimeDimension()).map(dimension -> dimension.getBizName())
                                  .collect(Collectors.toList());
        // 日期维度
        DateModeUtils dateModeUtils = ContextUtils.getBean(DateModeUtils.class);
        groups.add(dateModeUtils.getSysDateCol(metricReq.getDateInfo()));

        structReq.setGroups(groups);
        structReq.setDataSetName(tableName);
        structReq.setLimit(-1L);
        String sql = null;
        try {
            structReq.setAutoAddTimestamp(isForQuery);
            if (!isForQuery) {
                // 仅生成外层SQL
                QuerySqlReq querySqlReq = structReq.convert(true);
                sql = querySqlReq.getSql();
                String tmpTableName = SqlSelectHelper.getTableName(sql);
                sql = sql.replace(tmpTableName, getMetricTableName(metricIds.get(0)));
                // 转为中文
                List<MetricDO> metricResps = getMetrics(metricIds);
                sql = convertToAlias(metricResps, dims, sql);
                return sql;
            }
            S2SemanticLayerService semanticLayerService = ContextUtils.getBean(S2SemanticLayerService.class);
            SemanticQueryReq queryReq = semanticLayerService.buildQueryReq(structReq);
            SemanticTranslator translator = ContextUtils.getBean(SemanticTranslator.class);
            QueryStatement queryStatement = semanticLayerService.buildQueryStatement(queryReq, User.getFakeUser());
            // 清空外层聚合（指标内部已聚合）
            queryStatement.setLimit(-1);
            if (queryStatement.getQueryParam() != null && queryStatement.getQueryParam().getAggregators() != null) {
                queryStatement.getQueryParam().getAggregators().clear();
            }
            translator.translate(queryStatement);
            sql = queryStatement.getSql();
            if (simplify && StringUtils.isNotEmpty(sql)) {
                // 此时的格式为select .. limit..
                PlainSelect select = SqlSelectHelper.getSelect(sql).getPlainSelect();
                // 5层select嵌套(字段转biz-聚合-转指标-关联-过滤
                if (select != null && select.getFromItem() != null) {
                    FromItem fromItem = select.getFromItem();
                    select = ((ParenthesedSelect) fromItem).getSelect().getPlainSelect();
                    if (select != null && select.getFromItem() != null) {
                        select = ((ParenthesedSelect) select.getFromItem()).getSelect().getPlainSelect();
                    }
                }
                sql = ((ParenthesedSelect) select.getFromItem()).getSelect().getPlainSelect().toString();
                sql = simplifySqlForMatialview(sql);
            }
        } catch (Exception e) {
            log.error("指标物化sql生成失败", e);
        }
        return sql;
    }

    private String convertToAlias(List<MetricDO> metricResps, List<DimensionResp> dims, String sql) {
        Map<String, String> filedNameToValueMap = new HashMap<>();
        for (MetricDO metricResp : metricResps) {
            filedNameToValueMap.put(metricResp.getBizName(), metricResp.getName());
        }
        dims.stream().forEach(dim -> {
            filedNameToValueMap.put(dim.getBizName(), dim.getDescription());
        });

        DateModeUtils dateModeUtils = ContextUtils.getBean(DateModeUtils.class);
        filedNameToValueMap.put(dateModeUtils.getSysDateCol(), "日期");
        filedNameToValueMap.put(dateModeUtils.getSysDateMonthCol(), "月份");
        filedNameToValueMap.put(dateModeUtils.getSysDateYearCol(), "年份");

        sql = SqlReplaceHelper.replaceAliasInSelect(sql, filedNameToValueMap);
        return sql;
    }

    // 物化视图简化(将嵌套select简化)
    private String simplifySqlForMatialview(String originalSql) {
        // Parse the original SQL
        PlainSelect plainSelect = SqlSelectHelper.getSelect(originalSql).getPlainSelect();

        // Get the inner query
        PlainSelect innerPlainSelect = ((ParenthesedSelect) plainSelect.getFromItem()).getPlainSelect();

        // Get aggrators from outer query
        List<SelectItem<?>> outerSelectItems = plainSelect.getSelectItems();
        Map<String, Expression> outerExpressionItems = outerSelectItems.stream()
                                                                       .filter(selectItem -> selectItem.getAlias() != null)
                                                                       .collect(
                                                                           Collectors.toMap(selectItem -> selectItem.getAlias().getName(),
                                                                               selectItem -> selectItem.getExpression()));

        for (SelectItem innerSelectItem : innerPlainSelect.getSelectItems()) {
            String fieldName = innerSelectItem.getAlias().getName();
            if (outerExpressionItems.containsKey(fieldName)) {
                Expression outerExpression = outerExpressionItems.get(fieldName);
                if (outerExpression instanceof Function) {
                    Function castFunction = new Function();
                    castFunction.setName(((Function) outerExpression).getName());
                    ExpressionList<Expression> expressionList = new ExpressionList<>();
                    expressionList.add(innerSelectItem.getExpression());
                    castFunction.setParameters(expressionList);
                    innerSelectItem.setExpression(castFunction);
                }
            }
        }
        if (plainSelect.getGroupBy() != null) {
            innerPlainSelect.setGroupByElement(plainSelect.getGroupBy());
        }

        return innerPlainSelect.toString();
    }

    @Override
    public void batchUpdateClassifications(MetaBatchReq metaBatchReq, User user) {
        List<MetricDO> metrics = getMetrics(metaBatchReq.getIds());
        for (MetricDO metricDO : metrics) {
            metricDO.setUpdatedAt(new Date());
            metricDO.setUpdatedBy(user.getUserName());
            fillClassifications(metaBatchReq, metricDO);
        }
        metricRepository.updateClassificationsBatch(metrics);
    }

    @Override
    public TaskLog queryTaskLog(Long metricId, int instanceId) {
        MetricPersistenceConfigService metricPersistenceConfigService = ContextUtils.getBean(MetricPersistenceConfigService.class);
        MetricPersistenceConfigResp configResp = metricPersistenceConfigService.getByMetricId(metricId);
        if (configResp == null) {
            return null;
        }
        SchedulerService schedulerService = ContextUtils.getBean(SchedulerService.class);
        TaskLog log = schedulerService.queryTaskLog(instanceId);
        return log;
    }

    @Override
    public List<TaskInstance> queryTaskInstances(Long metricId, String startDate, String endDate) {
        if (startDate == null) {
            startDate = "";
        }
        if (endDate == null) {
            endDate = "";
        }
        MetricPersistenceConfigService metricPersistence = ContextUtils.getBean(MetricPersistenceConfigService.class);
        MetricPersistenceConfigResp configResp = metricPersistence.getByMetricId(metricId);
        if (configResp != null) {
            if (StringUtils.isEmpty(configResp.getRunCycleCode())) {
                return null;
            }
            String taskName = getMetricTableName(metricId);
            SchedulerService schedulerService = ContextUtils.getBean(SchedulerService.class);
            List<TaskInstance> taskInstances = schedulerService.getTaskInstances(taskName, startDate, endDate);
            return taskInstances;
        }
        return null;
    }

    @Override
    public String getMetricTableName(Long metricId) {
        MetricResp metricResp = getMetric(metricId);
        return String.format("index_metrics_%s", metricResp.getBizName().toLowerCase(Locale.ROOT));
    }

    @Override
    public boolean runTask(Long metricId) {
        MetricResp metricResp = getMetric(metricId);
        if (metricResp == null) {
            return false;
        }
        if (metricResp.getIsPublish() == null || metricResp.getIsPublish() == 0) {
            throw new RuntimeException("指标未发布，无法执行任务");
        }

        SchedulerService schedulerService = ContextUtils.getBean(SchedulerService.class);
        return schedulerService.runTask(metricResp.getId());
    }

    @Override
    public void buildDependencyDAG(List<MetricResp> metricRespList) {
        List<MetricResp> dependencyList = new ArrayList<>();
        for (MetricResp metricResp : metricRespList) {
            fetchMetricLineage(metricResp, dependencyList);
        }
        // 加入所有的依赖指标
        if (metricRespList.size() > 0) {
            metricRespList.addAll(dependencyList);
        }
    }

    /**
     * 递归获取依赖指标
     *
     * @param metricResp
     * @param dependencyList
     */
    private void fetchMetricLineage(MetricResp metricResp, List<MetricResp> dependencyList) {
        MetricLineageService metricLineageService = ContextUtils.getBean(MetricLineageService.class);
        MetricLineage metricLineage = metricLineageService.getLineages(metricResp.getId());
        if (metricLineage == null) {
            return;
        }
        if (metricLineage.getChildren() != null && metricLineage.getChildren().size() > 0) {
            List<MetricLineage> children = metricLineage.getChildren();
            children.forEach(child -> {
                MetricResp childMetric = getMetric(child.getMetricId());
                if (childMetric != null) {
                    dependencyList.add(childMetric);
                }
                if (child.getChildren() != null && child.getChildren().size() > 0) {
                    fetchMetricLineage(childMetric, dependencyList);
                }
            });
        }
    }

    private void fillClassifications(MetaBatchReq metaBatchReq, MetricDO metricDO) {
        String classificationStr = metricDO.getClassifications();
        Set<String> classificationsList;
        if (StringUtils.isBlank(classificationStr)) {
            classificationsList = new HashSet<>();
        } else {
            classificationsList = new HashSet<>(Arrays.asList(classificationStr.split(",")));
        }

        if (EventType.ADD.equals(metaBatchReq.getType())) {
            classificationsList.addAll(metaBatchReq.getClassifications());
        }
        if (EventType.DELETE.equals(metaBatchReq.getType())) {
            classificationsList.removeAll(metaBatchReq.getClassifications());
        }
        String classifications = "";
        if (!CollectionUtils.isEmpty(classificationsList)) {
            classifications = StringUtils.join(classificationsList, ",");
        }
        metricDO.setClassifications(classifications);
    }

    @Override
    public void batchUpdateSensitiveLevel(MetaBatchReq metaBatchReq, User user) {
        List<MetricDO> metrics = getMetrics(metaBatchReq.getIds());
        for (MetricDO metricDO : metrics) {
            metricDO.setUpdatedAt(new Date());
            metricDO.setUpdatedBy(user.getUserName());
            metricDO.setSensitiveLevel(metaBatchReq.getSensitiveLevel());
        }
        updateBatchById(metrics);
    }

    @Override
    public void deleteMetric(Long id, User user) {
        MetricDO metricDO = metricRepository.getMetricById(id);
        if (metricDO == null) {
            throw new RuntimeException(String.format("the metric %s not exist", id));
        }
        metricDO.setStatus(StatusEnum.DELETED.getCode());
        metricDO.setUpdatedAt(new Date());
        metricDO.setUpdatedBy(user.getUserName());
        metricRepository.updateMetric(metricDO);
        sendEventBatch(Lists.newArrayList(metricDO), EventType.DELETE);

        // 删除持久化
        MetricPersistenceConfigService metricPersistenceConfigService = ContextUtils.getBean(MetricPersistenceConfigService.class);
        MetricPersistenceConfigResp configResp = metricPersistenceConfigService.getByMetricId(id);
        if (configResp != null) {
            metricPersistenceConfigService.delete(configResp.getId());
        }
    }

    @Override
    public PageInfo<MetricResp> queryMetricMarket(PageMetricReq pageMetricReq, User user) {
        //search by whole text
        PageInfo<MetricResp> metricRespPageInfo = queryMetric(pageMetricReq, user);
        if (metricRespPageInfo.hasContent() || StringUtils.isBlank(pageMetricReq.getKey())) {
            return metricRespPageInfo;
        }
        //search by text split
        QueryMapReq queryMapReq = new QueryMapReq();
        queryMapReq.setQueryText(pageMetricReq.getKey());
        queryMapReq.setUser(user);
        queryMapReq.setMapModeEnum(MapModeEnum.LOOSE);
        MapInfoResp mapMeta = chatQueryService.map(queryMapReq);
        Map<String, DataSetMapInfo> dataSetMapInfoMap = mapMeta.getDataSetMapInfo();
        if (CollectionUtils.isEmpty(dataSetMapInfoMap)) {
            return metricRespPageInfo;
        }
        Map<Long, Double> result =
            dataSetMapInfoMap.values().stream().map(DataSetMapInfo::getMapFields).filter(Objects::nonNull).flatMap(Collection::stream)
                             .filter(schemaElementMatch -> SchemaElementType.METRIC.equals(schemaElementMatch.getElement().getType()))
                             .collect(Collectors.toMap(schemaElementMatch -> schemaElementMatch.getElement().getId(),
                                 SchemaElementMatch::getSimilarity, (existingValue, newValue) -> existingValue));
        List<Long> metricIds = new ArrayList<>(result.keySet());
        if (CollectionUtils.isEmpty(result.keySet())) {
            return metricRespPageInfo;
        }
        pageMetricReq.setIds(metricIds);
        pageMetricReq.setKey("");
        PageInfo<MetricResp> metricPage = queryMetric(pageMetricReq, user);
        for (MetricResp metricResp : metricPage.getList()) {
            metricResp.setSimilarity(result.get(metricResp.getId()));
        }
        metricPage.getList().sort(Comparator.comparingDouble(MetricResp::getSimilarity).reversed());
        return metricPage;
    }

    @Override
    public PageInfo<MetricResp> queryMetric(PageMetricReq pageMetricReq, User user) {
        MetricFilter metricFilter = new MetricFilter();
        metricFilter.setUserName(user.getUserName());
        BeanUtils.copyProperties(pageMetricReq, metricFilter);
        if (!CollectionUtils.isEmpty(pageMetricReq.getDomainIds())) {
            List<ModelResp> modelResps = modelService.getAllModelByDomainIds(pageMetricReq.getDomainIds());
            List<Long> modelIds = modelResps.stream().map(ModelResp::getId).collect(Collectors.toList());
            pageMetricReq.getModelIds().addAll(modelIds);
        }
        PageInfo<MetricResp> pageInfo = new PageInfo<>();
        if (!CollectionUtils.isEmpty(pageMetricReq.getModelIds())) {
            metricFilter.setModelIds(pageMetricReq.getModelIds());
            List<Long> collectIds = getCollectIds(pageMetricReq, user);
            List<Long> idsToFilter = getIdsToFilter(pageMetricReq, collectIds);
            metricFilter.setIds(idsToFilter);
            PageInfo<MetricDO> metricDOPageInfo =
                PageHelper.startPage(pageMetricReq.getCurrent(), pageMetricReq.getPageSize())
                          .doSelectPageInfo(() -> queryMetric(metricFilter));
            BeanUtils.copyProperties(metricDOPageInfo, pageInfo);
            List<MetricResp> metricResps = convertList(metricDOPageInfo.getList(), collectIds);
            filterAdminRes(metricResps, user);
            pageInfo.setList(metricResps);
        }
        return pageInfo;
    }

    @Override
    public List<MetricDO> queryMetric(MetricFilter metricFilter) {
        return metricRepository.getMetric(metricFilter);
    }

    private List<MetricDO> getMetrics(List<Long> ids) {
        if (CollectionUtils.isEmpty(ids)) {
            return Lists.newArrayList();
        }
        MetricFilter metricFilter = new MetricFilter();
        metricFilter.setIds(ids);
        return metricRepository.getMetric(metricFilter);
    }

    @Override
    public List<MetricResp> getMetrics(MetaFilter metaFilter) {
        MetricFilter metricFilter = new MetricFilter();
        BeanUtils.copyProperties(metaFilter, metricFilter);
        List<MetricResp> metricResps = convertList(queryMetric(metricFilter));
        if (!CollectionUtils.isEmpty(metaFilter.getFieldsDepend())) {
            return filterByField(metricResps, metaFilter.getFieldsDepend());
        }
        if (metaFilter.getDataSetId() != null) {
            DataSetResp dataSetResp = dataSetService.getDataSet(metaFilter.getDataSetId());
            return MetricConverter.filterByDataSet(metricResps, dataSetResp);
        }
        return metricResps;
    }

    private List<Long> getCollectIds(PageMetricReq pageMetricReq, User user) {
        List<Collect> collectList = collectService.getCollectList(user.getUserName(), TypeEnums.METRIC);
        List<Long> collectIds = collectList.stream().map(Collect::getCollectId).collect(Collectors.toList());
        return collectIds;
    }

    private List<Long> getIdsToFilter(PageMetricReq pageMetricReq, List<Long> collectIds) {
        // 仅显示已收藏的指标
        if (pageMetricReq.isHasCollect()) {
            return collectIds;
        }
        return pageMetricReq.getIds();
    }

    private List<MetricResp> filterByField(List<MetricResp> metricResps, List<String> fields) {
        Set<MetricResp> metricRespFiltered = Sets.newHashSet();
        for (MetricResp metricResp : metricResps) {
            filterByField(metricResps, metricResp, fields, metricRespFiltered);
        }
        return new ArrayList<>(metricRespFiltered);
    }

    private boolean filterByField(List<MetricResp> metricResps, MetricResp metricResp, List<String> fields,
                                  Set<MetricResp> metricRespFiltered) {
        if (MetricDefineType.METRIC.equals(metricResp.getMetricDefineType())) {
            List<Long> ids =
                metricResp.getMetricDefineByMetricParams().getMetrics().stream().map(MetricParam::getId).collect(Collectors.toList());
            List<MetricResp> metricById = metricResps.stream().filter(metric -> ids.contains(metric.getId())).collect(Collectors.toList());
            for (MetricResp metric : metricById) {
                if (filterByField(metricResps, metric, fields, metricRespFiltered)) {
                    metricRespFiltered.add(metricResp);
                    return true;
                }
            }
        } else if (MetricDefineType.CUSTOM.equals(metricResp.getMetricDefineType())) {
            if (fields.stream().anyMatch(field -> metricResp.getExpr().contains(field))) {
                metricRespFiltered.add(metricResp);
                return true;
            }
        } else if (MetricDefineType.MEASURE.equals(metricResp.getMetricDefineType())) {
            List<MeasureParam> measures = metricResp.getMetricDefineByMeasureParams().getMeasures();
            List<String> fieldNameDepended = measures.stream().map(MeasureParam::getBizName)
                                                     //measure bizName = model bizName_fieldName
                                                     .map(name -> name.replaceFirst(metricResp.getModelBizName() + "_", ""))
                                                     .collect(Collectors.toList());
            if (fields.stream().anyMatch(fieldNameDepended::contains)) {
                metricRespFiltered.add(metricResp);
                return true;
            }
        }
        return false;
    }

    // 根据modelId获取所有的原子指标（度量和字段）
    @Override
    public List<MetricResp> getMetricsToCreateNewMetric(Long modelId) {
        MetricFilter metricFilter = new MetricFilter();
        metricFilter.setModelIds(Lists.newArrayList(modelId));
        List<MetricResp> metricResps = getMetrics(metricFilter);
        return metricResps.stream().filter(
            metricResp -> MetricDefineType.CUSTOM.equals(metricResp.getMetricDefineType()) || MetricDefineType.MEASURE.equals(
                metricResp.getMetricDefineType())).collect(Collectors.toList());
    }

    private void filterAdminRes(List<MetricResp> metricResps, User user) {
        filterUserRes(metricResps, user, AuthType.ADMIN);
    }

    private void filterUserRes(List<MetricResp> metricResps, User user, AuthType authType) {
        RoleAuthService roleAuthService = ContextUtils.getBean(RoleAuthService.class);
        List<RoleAuthResp> authList = roleAuthService.getMetricRoleAuthList(user.getRoleId());
        if (CollectionUtils.isEmpty(authList)) {
            return;
        }
        // 去掉未授权的指标
        metricResps.removeIf(metricResp -> !authList.stream().anyMatch(
            auth -> authType.name().equals(auth.getAuthType()) && auth.getDataId().equals(metricResp.getId())));

    }

    @Deprecated
    @Override
    public MetricResp getMetric(Long modelId, String bizName) {
        MetricFilter metricFilter = new MetricFilter();
        metricFilter.setBizName(bizName);
        metricFilter.setModelIds(Lists.newArrayList(modelId));
        List<MetricResp> metricResps = getMetrics(metricFilter);
        if (CollectionUtils.isEmpty(metricResps)) {
            return null;
        }
        return metricResps.get(0);
    }

    @Override
    public MetricResp getMetric(Long id, User user) {
        MetricDO metricDO = metricRepository.getMetricById(id);
        if (metricDO == null) {
            return null;
        }
        ModelFilter modelFilter = new ModelFilter(false, Lists.newArrayList(metricDO.getModelId()));
        Map<Long, ModelResp> modelMap = modelService.getModelMap(modelFilter);
        List<Collect> collectList = collectService.getCollectList(user.getUserName(), TypeEnums.METRIC);
        List<Long> collect = collectList.stream().map(Collect::getCollectId).collect(Collectors.toList());
        MetricResp metricResp = MetricConverter.convert2MetricResp(metricDO, modelMap, collect);

        // 修正指标名称 by charlie
        if (metricResp.getMetricDefineByMetricParams() != null) {
            metricResp.getMetricDefineByMetricParams().getMetrics().stream().forEach(metricParam -> {
                MetricDO metric = metricRepository.getMetricById(metricParam.getId());
                if (metric != null) {
                    metricParam.setName(metric.getName());
                }
            });
        }

        filterAdminRes(Lists.newArrayList(metricResp), user);
        return metricResp;
    }

    @Override
    public MetricResp getMetric(Long id) {
        MetricDO metricDO = metricRepository.getMetricById(id);
        if (metricDO == null) {
            return null;
        }
        return MetricConverter.convert2MetricResp(metricDO, new HashMap<>(), Lists.newArrayList());
    }

    /**
     * 获取最新且已发布的版本信息
     *
     * @param id
     * @return
     */
    @Override
    public MetricResp getLastVersionMetric(Long id) {
        MetricVersionService metricVersionService = ContextUtils.getBean(MetricVersionService.class);
        MetricResp metricResp = getMetric(id);
        List<MetricVersion> metricVersions = metricVersionService.queryMetricVersion(id);
        if (!CollectionUtils.isEmpty(metricVersions)) {
            MetricVersion metricVersion = metricVersions.stream().filter(v -> v.isStatus()).findFirst().orElse(null);
            if (metricVersion != null) {
                metricResp = JSON.parseObject(metricVersion.getDefaultConfig(), MetricResp.class);
                metricResp.setLastVersionNo(metricVersion.getVersionNo());
            }
        }
        return metricResp;
    }

    @Override
    public List<DrillDownDimension> getDrillDownDimension(Long metricId) {
        List<DrillDownDimension> drillDownDimensions = Lists.newArrayList();
        MetricResp metricResp = getMetric(metricId);
        if (metricResp == null) {
            return drillDownDimensions;
        }
        if (metricResp.getRelateDimension() != null && !CollectionUtils.isEmpty(metricResp.getRelateDimension().getDrillDownDimensions())) {
            for (DrillDownDimension drillDownDimension : metricResp.getRelateDimension().getDrillDownDimensions()) {
                if (drillDownDimension.isInheritedFromModel() && !drillDownDimension.isNecessary()) {
                    continue;
                }
                drillDownDimensions.add(drillDownDimension);
            }
        }
        ModelResp modelResp = modelService.getModel(metricResp.getModelId());
        if (modelResp.getDrillDownDimensions() == null) {
            return drillDownDimensions;
        }
        for (DrillDownDimension drillDownDimension : modelResp.getDrillDownDimensions()) {
            if (!drillDownDimensions.stream().map(DrillDownDimension::getDimensionId).collect(Collectors.toList())
                                    .contains(drillDownDimension.getDimensionId())) {
                drillDownDimension.setInheritedFromModel(true);
                drillDownDimensions.add(drillDownDimension);
            }
        }
        return drillDownDimensions;
    }

    @Override
    public void saveMetricQueryDefaultConfig(MetricQueryDefaultConfig defaultConfig, User user) {
        MetricQueryDefaultConfigDO defaultConfigDO =
            metricRepository.getDefaultQueryConfig(defaultConfig.getMetricId(), user.getUserName());
        if (defaultConfigDO == null) {
            defaultConfigDO = new MetricQueryDefaultConfigDO();
            defaultConfig.createdBy(user.getUserName());
            BeanMapper.mapper(defaultConfig, defaultConfigDO);
            metricRepository.saveDefaultQueryConfig(defaultConfigDO);
        } else {
            defaultConfig.setId(defaultConfigDO.getId());
            defaultConfig.updatedBy(user.getUserName());
            BeanMapper.mapper(defaultConfig, defaultConfigDO);
            metricRepository.updateDefaultQueryConfig(defaultConfigDO);
        }
    }

    @Override
    public MetricQueryDefaultConfig getMetricQueryDefaultConfig(Long metricId, User user) {
        MetricQueryDefaultConfigDO metricQueryDefaultConfigDO = metricRepository.getDefaultQueryConfig(metricId, user.getUserName());
        MetricQueryDefaultConfig metricQueryDefaultConfig = new MetricQueryDefaultConfig();
        BeanMapper.mapper(metricQueryDefaultConfigDO, metricQueryDefaultConfig);
        return metricQueryDefaultConfig;
    }

    // 查询List对象，返回符合条件的map
    private static <K, V> Map<K, V> collectToMap(List<V> list, java.util.function.Function<V, K> keyMapper) {
        return list.stream().collect(Collectors.toMap(keyMapper, java.util.function.Function.identity(), (k1, k2) -> k1));
    }

    /**
     * 检查指标是否重复
     */
    private void checkExist(List<MetricBaseReq> metricReqs) {
        Long modelId = metricReqs.get(0).getModelId();
        MetaFilter metaFilter = new MetaFilter();
        metaFilter.setModelIds(Lists.newArrayList(modelId));
        List<MetricResp> metricResps = getMetrics(metaFilter);
        Map<String, MetricResp> bizNameMap = collectToMap(metricResps, MetricResp::getBizName);
        Map<String, MetricResp> nameMap = collectToMap(metricResps, MetricResp::getName);
        for (MetricBaseReq metricReq : metricReqs) {
            if (bizNameMap.containsKey(metricReq.getBizName())) {
                MetricResp metricResp = bizNameMap.get(metricReq.getBizName());
                if (!metricResp.getId().equals(metricReq.getId())) {
                    throw new RuntimeException(
                        String.format("该模型下存在相同的指标字段名:%s 创建人:%s", metricReq.getBizName(), metricResp.getCreatedBy()));
                }
            }
            if (nameMap.containsKey(metricReq.getName())) {
                MetricResp metricResp = nameMap.get(metricReq.getName());
                if (!metricResp.getId().equals(metricReq.getId())) {
                    throw new RuntimeException(String.format("该模型下存在相同的指标名:%s 创建人:%s", metricReq.getName(), metricResp.getCreatedBy()));
                }
            }
        }
    }

    private List<MetricResp> convertList(List<MetricDO> metricDOS) {
        return convertList(metricDOS, Lists.newArrayList());
    }

    private List<MetricResp> convertList(List<MetricDO> metricDOS, List<Long> collect) {
        List<MetricResp> metricResps = Lists.newArrayList();
        List<Long> modelIds = metricDOS.stream().map(MetricDO::getModelId).collect(Collectors.toList());
        ModelFilter modelFilter = new ModelFilter(false, modelIds);
        Map<Long, ModelResp> modelMap = modelService.getModelMap(modelFilter);
        if (!CollectionUtils.isEmpty(metricDOS)) {
            metricResps = metricDOS.stream().map(metricDO -> MetricConverter.convert2MetricResp(metricDO, modelMap, collect))
                                   .collect(Collectors.toList());
        }

        return metricResps;
    }

    @Override
    public void sendMetricEventBatch(List<Long> modelIds, EventType eventType) {
        MetricFilter metricFilter = new MetricFilter();
        metricFilter.setModelIds(modelIds);
        List<MetricDO> metricDOS = queryMetric(metricFilter);
        sendEventBatch(metricDOS, eventType);
    }

    @Override
    public List<MetricResp> queryMetrics(MetricsFilter metricsFilter) {
        List<MetricDO> metricDOS = metricRepository.getMetrics(metricsFilter);
        return convertList(metricDOS, new ArrayList<>());
    }

    @Override
    public DataEvent getDataEvent() {
        MetricsFilter metricsFilter = new MetricsFilter();
        List<MetricDO> metricDOS = metricRepository.getMetrics(metricsFilter);
        return getDataEvent(metricDOS, EventType.ADD);
    }

    private DataEvent getDataEvent(List<MetricDO> metricDOS, EventType eventType) {
        List<DataItem> dataItems = metricDOS.stream().map(this::getDataItem).collect(Collectors.toList());
        return new DataEvent(this, dataItems, eventType);
    }

    private void sendEventBatch(List<MetricDO> metricDOS, EventType eventType) {
        DataEvent dataEvent = getDataEvent(metricDOS, eventType);
        eventPublisher.publishEvent(dataEvent);
    }

    private void sendEvent(DataItem dataItem, EventType eventType) {
        eventPublisher.publishEvent(new DataEvent(this, Lists.newArrayList(dataItem), eventType));
    }

    private DataItem getDataItem(MetricDO metricDO) {
        MetricResp metricResp = MetricConverter.convert2MetricResp(metricDO, new HashMap<>(), Lists.newArrayList());
        fillDefaultAgg(metricResp);
        return DataItem.builder().id(metricDO.getId() + Constants.UNDERLINE).name(metricDO.getName()).bizName(metricDO.getBizName())
                       .modelId(metricDO.getModelId() + Constants.UNDERLINE).type(TypeEnums.METRIC).defaultAgg(metricResp.getDefaultAgg())
                       .build();
    }

    @Override
    public void batchFillMetricDefaultAgg(List<MetricResp> metricResps, List<ModelResp> modelResps) {
        Map<Long, ModelResp> modelRespMap = modelResps.stream().collect(Collectors.toMap(ModelResp::getId, m -> m));
        for (MetricResp metricResp : metricResps) {
            fillDefaultAgg(metricResp, modelRespMap.get(metricResp.getModelId()));
        }
    }

    private void fillDefaultAgg(MetricResp metricResp) {
        if (MetricDefineType.MEASURE.equals(metricResp.getMetricDefineType())) {
            Long modelId = metricResp.getModelId();
            ModelResp modelResp = modelService.getModel(modelId);
            fillDefaultAgg(metricResp, modelResp);
        }
    }

    private void fillDefaultAgg(MetricResp metricResp, ModelResp modelResp) {
        metricResp.setDefaultAgg(getDefaultAgg(metricResp, modelResp));
    }

    private String getDefaultAgg(MetricResp metricResp, ModelResp modelResp) {
        return metricResp.getOperations();
    }

    @Override
    public QueryMetricReq newQueryMetricReq(Long id) {
        MetricResp metric = getMetric(id, User.getFakeUser());
        if (metric == null) {
            throw new RuntimeException("指标不存在");
        }
        List<Long> dimIds =
            metric.getRelateDimension().getDrillDownDimensions().stream().map(dd -> dd.getDimensionId()).collect(Collectors.toList());
        QueryMetricReq queryMetricReq = new QueryMetricReq();
        queryMetricReq.setMetricIds(Arrays.asList(id));
        queryMetricReq.setDomainId(metric.getDomainId());
        queryMetricReq.setDimensionIds(dimIds);
        queryMetricReq.setLimit(100L);
        DateConf dateConf = new DateConf();
        dateConf.setUnit(7);
        dateConf.setPeriod(Constants.DAY);
        dateConf.setDateMode(DateConf.DateMode.BETWEEN);
        dateConf.setStartDate(DateUtils.getBeforeDate(7));
        dateConf.setEndDate(DateUtils.getBeforeDate(1));
        dateConf.setGroupByDate(true);
        queryMetricReq.setDateInfo(dateConf);
        return queryMetricReq;
    }

    @Override
    public QueryStructReq convert(QueryMetricReq queryMetricReq) {
        //1. 主题域中的所有模型
        Set<Long> modelIdsByDomainId = getModelIdsByDomainId(queryMetricReq);

        //2. 指标和维度
        List<MetricResp> metricResps = getMetricResps(queryMetricReq, modelIdsByDomainId);

        List<DimensionResp> dimensionResps = getDimensionResps(modelIdsByDomainId);
        Map<Long, DimensionResp> dimensionRespMap = dimensionResps.stream().collect(Collectors.toMap(DimensionResp::getId, d -> d));

        //3. 数据集模型
        Set<Long> modelIds = getModelIds(modelIdsByDomainId, metricResps, dimensionResps);
        ModelCluster modelCluster = getModelCluster(metricResps, modelIds);
        if (modelCluster == null) {
            throw new IllegalArgumentException("输入参数无效，请检查");
        }
        //4. 聚合
        List<String> dimensionBizNames = dimensionResps.stream()
                                                       .filter(entry -> !entry.isTimeDimension())
                                                       .filter(entry -> modelCluster.getModelIds().contains(entry.getModelId()))
                                                       .filter(entry -> queryMetricReq.getDimensionNames().contains(entry.getName())
                                                                        || queryMetricReq.getDimensionNames().contains(entry.getBizName())
                                                                        || queryMetricReq.getDimensionIds().contains(entry.getId()))
                                                       .map(SchemaItem::getBizName).collect(Collectors.toList());

        QueryStructReq queryStructReq = new QueryStructReq();
        // 区分统计周期和查询维度 2024-07-02
        // 多个指标同时查询时，统计周期应保持一致
        // 根据指标定义加载默认的统计周期 @2024-07-08
        String statisticsPeriod = queryMetricReq.getDateInfo().getPeriod();
        if (metricResps.size() > 0) {
            queryMetricReq.getDateInfo().setPeriod(metricResps.get(0).getStatisticsPeriod());
        }
        // 默认按日期分组
        if (CollectionUtils.isEmpty(queryMetricReq.getDimensionIds()) || (StringUtils.isNotEmpty(statisticsPeriod)
                                                                          && queryMetricReq.getDateInfo().isGroupByDate())) {
            queryStructReq.getGroups().add(queryMetricReq.getDateInfo().getGroupByTimeDimension());
        }

        if (!CollectionUtils.isEmpty(dimensionBizNames)) {
            queryStructReq.getGroups().addAll(dimensionBizNames);
        }

        List<String> metricBizNames = metricResps.stream()
                                                 .filter(m -> modelCluster.getModelIds().contains(m.getModelId()))
                                                 .map(MetricResp::getBizName).collect(Collectors.toList());

        if (CollectionUtils.isEmpty(metricBizNames)) {
            throw new IllegalArgumentException("指标数据异常，请检查bizName");
        }
        // 百分百是否转化
        if (queryMetricReq.isOriginalData()) {
            queryStructReq.setConvertToPercent(false);
        }

        // 特定指标
        List<MetricResp> queryMetrics = metricResps.stream().filter(m -> queryMetricReq.getMetricIds().contains(m.getId())).collect(
            Collectors.toList());
        // 聚合算子(指标定义中的聚合算子)
        this.buildAggregator(queryMetrics, queryStructReq);

        // 返回格式
        this.buildCastor(queryStructReq, queryMetrics);


        queryStructReq.setLimit(queryMetricReq.getLimit());
        //6. 模型
        for (Long modelId : modelCluster.getModelIds()) {
            queryStructReq.addModelId(modelId);
        }
        // 过滤条件
        List<Filter> filters = queryMetricReq.getFilters();
        for (Filter filter : filters) {
            if (StringUtils.isBlank(filter.getBizName())) {
                DimensionResp dimensionResp = dimensionRespMap.get(filter.getId());
                if (dimensionResp != null) {
                    filter.setBizName(dimensionResp.getBizName());
                }
            }
        }
        queryStructReq.setDimensionFilters(filters);
        //7. 日期
        queryStructReq.setDateInfo(queryMetricReq.getDateInfo());
        //8. 指标
        queryStructReq.setMetricIds(queryMetrics.stream().map(MetricResp::getId).collect(Collectors.toList()));
        return queryStructReq;
    }

    @Override
    public void buildCastor(QueryStructReq queryStructReq, List<MetricResp> metricResps) {
        // 处理返回的数据格式
        queryStructReq.getAggregators().forEach(agg -> {
            MetricResp metricResp = metricResps.stream().filter(m -> agg.getColumn().equals(m.getBizName())).findFirst().orElse(null);
            if (metricResp == null) {
                return;
            }
            List<Castor> castors = new ArrayList<>();
            if (DataFormatTypeEnum.DECIMAL.equals(metricResp.getDataFormatType())) {
                Castor castor = new Castor();
                castor.setName(metricResp.getBizName());
                castor.setFunction(Constants.ROUND);
                castor.setArgs(Arrays.asList(new LongValue(metricResp.getDataFormat().getDecimalPlaces())));
                castors.add(castor);
            }
            if (DataFormatTypeEnum.PERCENT.equals(metricResp.getDataFormatType())) {
                Castor castor = new Castor();

                if (queryStructReq.isConvertToPercent()) {
                    castor.setName(metricResp.getBizName());
                    castor.setFunction(Constants.MULTIPLY);
                    castor.setArgs(Arrays.asList(new LongValue(100)));
                    castors.add(castor);

                    castor = new Castor();
                    castor.setName(metricResp.getBizName());
                    castor.setFunction(Constants.ROUND);
                    castor.setArgs(Arrays.asList(new LongValue(metricResp.getDataFormat().getDecimalPlaces())));
                    castors.add(castor);

                    castor = new Castor();
                    castor.setName(metricResp.getBizName());
                    castor.setFunction(Constants.CONCAT);
                    castor.setArgs(Arrays.asList(new StringValue(Constants.PERCENT_SIGN)));
                    castors.add(castor);
                } else {
                    // 不转换，保留小数点后n+2位
                    castor.setName(metricResp.getBizName());
                    castor.setFunction(Constants.ROUND);
                    int n = metricResp.getDataFormat().getDecimalPlaces() + 2;
                    castor.setArgs(Arrays.asList(new LongValue(n)));
                    castors.add(castor);
                }
            }
            agg.setCastors(castors);
        });
    }

    @Override
    public void buildAggregator(List<MetricResp> metricResps, QueryStructReq queryStructReq) {
        List<Aggregator> aggregators = new ArrayList<>();
        // 处理聚合算子 增加同比和环比
        List<MetricYamlTpl> metricYamlTpls = MetricYamlManager.convert2YamlObj(metricResps);
        metricYamlTpls.stream().forEach(m -> {
            Aggregator aggregator = new Aggregator();
            aggregator.setFunc(AggOperatorEnum.UNKNOWN);
            aggregator.setColumn(m.getName());
            if (AggOperatorEnum.RATIO_ROLL.name().equalsIgnoreCase(m.getTypeParams().getOperations())) {
                queryStructReq.getRollOverMap().put(m.getName(), String.format("%s%s", m.getName(), Constants.ROLL_SUFFIX));
                aggregator.setFunc(AggOperatorEnum.RATIO_ROLL);
            } else if (AggOperatorEnum.RATIO_OVER.name().equalsIgnoreCase(m.getTypeParams().getOperations())) {
                queryStructReq.getRollOverMap().put(m.getName(), String.format("%s%s", m.getName(), Constants.OVER_SUFFIX));
                aggregator.setFunc(AggOperatorEnum.RATIO_OVER);
            } else if (StringUtils.isNotEmpty(m.getTypeParams().getOperations())) {
                aggregator.setFunc(AggOperatorEnum.of(m.getTypeParams().getOperations()));
            }
            aggregators.add(aggregator);
        });
        queryStructReq.setAggregators(aggregators);
    }

    private ModelCluster getModelCluster(List<MetricResp> metricResps, Set<Long> modelIds) {
        Map<String, ModelCluster> modelClusterMap = ModelClusterBuilder.buildModelClusters(new ArrayList<>(modelIds));

        Map<String, List<SchemaItem>> modelClusterToMatchCount = new HashMap<>();
        for (ModelCluster modelCluster : modelClusterMap.values()) {
            for (MetricResp metricResp : metricResps) {
                if (modelCluster.getModelIds().contains(metricResp.getModelId())) {
                    modelClusterToMatchCount.computeIfAbsent(modelCluster.getKey(), k -> new ArrayList<>()).add(metricResp);
                }
            }
        }
        String keyWithMaxSize = modelClusterToMatchCount.entrySet().stream().max(Comparator.comparingInt(entry -> entry.getValue().size()))
                                                        .map(Map.Entry::getKey).orElse(null);

        return modelClusterMap.get(keyWithMaxSize);
    }

    private Set<Long> getModelIds(Set<Long> modelIdsByDomainId, List<MetricResp> metricResps, List<DimensionResp> dimensionResps) {
        Set<Long> result = new HashSet<>();
        if (org.apache.commons.collections.CollectionUtils.isNotEmpty(modelIdsByDomainId)) {
            result.addAll(modelIdsByDomainId);
            return result;
        }
        Set<Long> metricModelIds = metricResps.stream().map(entry -> entry.getModelId()).collect(Collectors.toSet());
        result.addAll(metricModelIds);

        Set<Long> dimensionModelIds = dimensionResps.stream().map(entry -> entry.getModelId()).collect(Collectors.toSet());
        result.addAll(dimensionModelIds);
        return result;
    }

    private List<DimensionResp> getDimensionResps(Set<Long> modelIds) {
        DimensionsFilter dimensionsFilter = new DimensionsFilter();
        dimensionsFilter.setModelIds(new ArrayList<>(modelIds));
        return dimensionService.queryDimensions(dimensionsFilter);
    }

    private List<MetricResp> getMetricResps(QueryMetricReq queryMetricReq, Set<Long> modelIds) {
        MetricsFilter metricsFilter = new MetricsFilter();
        BeanUtils.copyProperties(queryMetricReq, metricsFilter);
        metricsFilter.setModelIds(new ArrayList<>(modelIds));
        return queryMetrics(metricsFilter);
    }

    private Set<Long> getModelIdsByDomainId(QueryMetricReq queryMetricReq) {
        List<ModelResp> modelResps = modelService.getAllModelByDomainIds(Collections.singletonList(queryMetricReq.getDomainId()));
        return modelResps.stream().map(ModelResp::getId).collect(Collectors.toSet());
    }

    @Override
    public String buildAsSelectSql(Long metricId) {
        MetricDO metricDO = metricRepository.getMetricById(metricId);
        if (metricDO == null) {
            return null;
        }
        ModelResp modelResp = modelService.getModel(metricDO.getModelId());
        if (modelResp == null) {
            return null;
        }
        return buildMetricSql(modelResp, metricDO);
    }
}
