/*
 * 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.storage.gtrecord;

import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.kylin.common.QueryContextFacade;
import org.apache.kylin.common.util.Pair;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.RawQueryLastHacker;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
import org.apache.kylin.cube.gridtable.CuboidToGridTableMappingExt;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.cube.model.CubeDesc.DeriveInfo;
import org.apache.kylin.cube.model.RowKeyColDesc;
import org.apache.kylin.gridtable.StorageLimitLevel;
import org.apache.kylin.measure.MeasureType;
import org.apache.kylin.measure.bitmap.BitmapMeasureType;
import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.filter.CaseTupleFilter;
import org.apache.kylin.metadata.filter.ColumnTupleFilter;
import org.apache.kylin.metadata.filter.CompareTupleFilter;
import org.apache.kylin.metadata.filter.LogicalTupleFilter;
import org.apache.kylin.metadata.filter.TupleFilter;
import org.apache.kylin.metadata.filter.TupleFilter.FilterOperatorEnum;
import org.apache.kylin.metadata.model.DynamicFunctionDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.MeasureDesc;
import org.apache.kylin.metadata.model.PartitionDesc;
import org.apache.kylin.metadata.model.SegmentStatusEnum;
import org.apache.kylin.metadata.model.Segments;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.realization.SQLDigest;
import org.apache.kylin.metadata.tuple.ITupleIterator;
import org.apache.kylin.metadata.tuple.TupleInfo;
import org.apache.kylin.storage.IStorageQuery;
import org.apache.kylin.storage.StorageContext;
import org.apache.kylin.storage.translate.DerivedFilterTranslator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.kylin.shaded.com.google.common.collect.Lists;
import org.apache.kylin.shaded.com.google.common.collect.Maps;
import org.apache.kylin.shaded.com.google.common.collect.Sets;

public abstract class GTCubeStorageQueryBase implements IStorageQuery {

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

    protected CubeInstance cubeInstance;
    protected CubeDesc cubeDesc;

    public GTCubeStorageQueryBase(CubeInstance cube) {
        this.cubeInstance = cube;
        this.cubeDesc = cube.getDescriptor();
    }

    @Override
    public ITupleIterator search(StorageContext context, SQLDigest sqlDigest, TupleInfo returnTupleInfo) {
        throw new UnsupportedOperationException("Removed in Kylin 4.0 .");
    }

    public GTCubeStorageQueryRequest getStorageQueryRequest(StorageContext context, SQLDigest sqlDigest,
            TupleInfo returnTupleInfo) {
        context.setStorageQuery(this);

        //cope with queries with no aggregations
        RawQueryLastHacker.hackNoAggregations(sqlDigest, cubeDesc, returnTupleInfo);

        // Customized measure taking effect: e.g. allow custom measures to help raw queries
        notifyBeforeStorageQuery(sqlDigest);

        Collection<TblColRef> groups = sqlDigest.groupbyColumns;
        TupleFilter filter = sqlDigest.filter;

        // build dimension & metrics
        Set<TblColRef> dimensions = new LinkedHashSet<>();
        Set<FunctionDesc> metrics = new LinkedHashSet<>();
        buildDimensionsAndMetrics(sqlDigest, dimensions, metrics);

        // all dimensions = groups + other(like filter) dimensions
        Set<TblColRef> otherDims = Sets.newHashSet(dimensions);
        otherDims.removeAll(groups);

        // expand derived (xxxD means contains host columns only, derived columns were translated)
        Set<TblColRef> derivedPostAggregation = Sets.newHashSet();
        Set<TblColRef> groupsD = expandDerived(groups, derivedPostAggregation);
        Set<TblColRef> otherDimsD = expandDerived(otherDims, derivedPostAggregation);
        otherDimsD.removeAll(groupsD);

        // identify cuboid
        Set<TblColRef> dimensionsD = new LinkedHashSet<>();
        dimensionsD.addAll(groupsD);
        dimensionsD.addAll(otherDimsD);
        Cuboid cuboid = findCuboid(cubeInstance, dimensionsD, metrics);
        context.setCuboid(cuboid);

        // set cuboid to GridTable mapping
        boolean noDynamicCols;
        // dynamic dimensions
        List<TblColRef> dynGroups = Lists.newArrayList(sqlDigest.dynGroupbyColumns.keySet());
        noDynamicCols = dynGroups.isEmpty();
        List<TupleExpression> dynGroupExprs = Lists.newArrayListWithExpectedSize(sqlDigest.dynGroupbyColumns.size());
        for (TblColRef dynGroupCol : dynGroups) {
            dynGroupExprs.add(sqlDigest.dynGroupbyColumns.get(dynGroupCol));
        }
        // dynamic measures
        List<DynamicFunctionDesc> dynFuncs = sqlDigest.dynAggregations;
        noDynamicCols = noDynamicCols && dynFuncs.isEmpty();

        CuboidToGridTableMapping mapping = noDynamicCols ? new CuboidToGridTableMapping(cuboid)
                : new CuboidToGridTableMappingExt(cuboid, dynGroups, dynFuncs);
        context.setMapping(mapping);

        // set whether to aggr at storage
        Set<TblColRef> singleValuesD = findSingleValueColumns(filter);
        context.setNeedStorageAggregation(isNeedStorageAggregation(cuboid, groupsD, singleValuesD));

        // exactAggregation mean: needn't aggregation at storage and query engine both.
        boolean exactAggregation = isExactAggregation(context, cuboid, groups, otherDimsD, singleValuesD,
                derivedPostAggregation, sqlDigest.aggregations, sqlDigest.aggrSqlCalls, sqlDigest.groupByExpression);
        context.setExactAggregation(exactAggregation);

        // replace derived columns in filter with host columns; columns on loosened condition must be added to group by
        Set<TblColRef> loosenedColumnD = Sets.newHashSet();
        Set<TblColRef> filterColumnD = Sets.newHashSet();
        TupleFilter filterD = translateDerived(filter, loosenedColumnD);
        groupsD.addAll(loosenedColumnD);
        TupleFilter.collectColumns(filterD, filterColumnD);
        context.setFilterMask(getQueryFilterMask(filterColumnD));

        // set limit push down
        enableStorageLimitIfPossible(cuboid, groups, dynGroups, derivedPostAggregation, groupsD, filterD,
                loosenedColumnD, sqlDigest, context);
        // set whether to aggregate results from multiple partitions
        enableStreamAggregateIfBeneficial(cuboid, groupsD, context);
        // check query deadline
        QueryContextFacade.current().checkMillisBeforeDeadline();

        // push down having clause filter if possible
        TupleFilter havingFilter = checkHavingCanPushDown(sqlDigest.havingFilter, groupsD, sqlDigest.aggregations,
                metrics);

        logger.info(
                "Cuboid identified: cube={}, cuboidId={}, groupsD={}, filterD={}, limitPushdown={}, limitLevel={}, storageAggr={}",
                cubeInstance.getName(), cuboid.getId(), groupsD, filterColumnD, context.getFinalPushDownLimit(),
                context.getStorageLimitLevel(), context.isNeedStorageAggregation());

        return new GTCubeStorageQueryRequest(cuboid, dimensionsD, groupsD, dynGroups, dynGroupExprs, filterColumnD,
                metrics, dynFuncs, filterD, havingFilter, context);
    }

    protected abstract String getGTStorage();

    protected Cuboid findCuboid(CubeInstance cubeInstance, Set<TblColRef> dimensionsD, Set<FunctionDesc> metrics) {
        return Cuboid.findCuboid(cubeInstance.getCuboidScheduler(), dimensionsD, metrics);
    }

    protected ITupleConverter newCubeTupleConverter(CubeSegment cubeSeg, Cuboid cuboid,
            Set<TblColRef> selectedDimensions, Set<FunctionDesc> selectedMetrics, int[] gtColIdx, TupleInfo tupleInfo) {
        return new CubeTupleConverter(cubeSeg, cuboid, selectedDimensions, selectedMetrics, gtColIdx, tupleInfo);
    }

    protected void buildDimensionsAndMetrics(SQLDigest sqlDigest, Collection<TblColRef> dimensions,
            Collection<FunctionDesc> metrics) {
        for (FunctionDesc func : sqlDigest.aggregations) {
            if (!func.isDimensionAsMetric() && !FunctionDesc.FUNC_GROUPING.equalsIgnoreCase(func.getExpression())) {
                // use the FunctionDesc from cube desc as much as possible, that has more info such as HLLC precision
                metrics.add(findAggrFuncFromCubeDesc(func));
            }
        }

        for (TblColRef column : sqlDigest.allColumns) {
            // skip measure columns
            if ((sqlDigest.metricColumns.contains(column) || sqlDigest.rtMetricColumns.contains(column))
                    && !(sqlDigest.groupbyColumns.contains(column) || sqlDigest.filterColumns.contains(column)
                            || sqlDigest.rtDimensionColumns.contains(column))) {
                continue;
            }

            dimensions.add(column);
        }
    }

    private FunctionDesc findAggrFuncFromCubeDesc(FunctionDesc aggrFunc) {
        for (MeasureDesc measure : cubeDesc.getMeasures()) {
            if (measure.getFunction().equals(aggrFunc))
                return measure.getFunction();
        }
        return aggrFunc;
    }

    protected Set<TblColRef> expandDerived(Collection<TblColRef> cols, Set<TblColRef> derivedPostAggregation) {
        Set<TblColRef> expanded = Sets.newHashSet();
        for (TblColRef col : cols) {
            if (cubeDesc.hasHostColumn(col)) {
                DeriveInfo hostInfo = cubeDesc.getHostInfo(col);
                for (TblColRef hostCol : hostInfo.columns) {
                    expanded.add(hostCol);
                    if (!hostInfo.isOneToOne)
                        derivedPostAggregation.add(hostCol);
                }
            } else {
                expanded.add(col);
            }
        }
        return expanded;
    }

    @SuppressWarnings("unchecked")
    protected Set<TblColRef> findSingleValueColumns(TupleFilter filter) {
        Set<CompareTupleFilter> compareTupleFilterSet = findSingleValuesCompFilters(filter);

        // expand derived
        Set<TblColRef> resultD = Sets.newHashSet();
        for (CompareTupleFilter compFilter : compareTupleFilterSet) {
            TblColRef tblColRef = compFilter.getColumn();
            if (cubeDesc.isExtendedColumn(tblColRef)) {
                throw new CubeDesc.CannotFilterExtendedColumnException(tblColRef);
            }
            if (cubeDesc.isDerived(compFilter.getColumn())) {
                DeriveInfo hostInfo = cubeDesc.getHostInfo(tblColRef);
                if (hostInfo.isOneToOne) {
                    resultD.addAll(Arrays.asList(hostInfo.columns));
                }
                //if not one2one, it will be pruned
            } else {
                resultD.add(compFilter.getColumn());
            }
        }
        return resultD;
    }

    // FIXME should go into nested AND expression
    protected Set<CompareTupleFilter> findSingleValuesCompFilters(TupleFilter filter) {
        Collection<? extends TupleFilter> toCheck;
        if (filter instanceof CompareTupleFilter) {
            toCheck = Collections.singleton(filter);
        } else if (filter instanceof LogicalTupleFilter && filter.getOperator() == FilterOperatorEnum.AND) {
            toCheck = filter.getChildren();
        } else {
            return Collections.emptySet();
        }

        Set<CompareTupleFilter> result = Sets.newHashSet();
        for (TupleFilter f : toCheck) {
            if (f instanceof CompareTupleFilter) {
                CompareTupleFilter compFilter = (CompareTupleFilter) f;
                // is COL=const ?
                if (compFilter.getOperator() == FilterOperatorEnum.EQ && compFilter.getValues().size() == 1
                        && compFilter.getColumn() != null) {
                    result.add(compFilter);
                }
            }
        }
        return result;
    }

    private long getQueryFilterMask(Set<TblColRef> filterColumnD) {
        long filterMask = 0;

        logger.info("Filter column set for query: {}", filterColumnD);
        if (filterColumnD.isEmpty() == false) {
            RowKeyColDesc[] allColumns = cubeDesc.getRowkey().getRowKeyColumns();
            for (int i = 0; i < allColumns.length; i++) {
                if (filterColumnD.contains(allColumns[i].getColRef())) {
                    filterMask |= 1L << allColumns[i].getBitIndex();
                }
            }
        }
        logger.info("Filter mask is: {}", filterMask);
        return filterMask;
    }

    public boolean isNeedStorageAggregation(Cuboid cuboid, Collection<TblColRef> groupD,
            Collection<TblColRef> singleValueD) {
        HashSet<TblColRef> temp = Sets.newHashSet();
        temp.addAll(groupD);
        temp.addAll(singleValueD);
        if (cuboid.getColumns().size() == temp.size()) {
            logger.debug("Does not need storage aggregation");
            return false;
        } else {
            logger.debug("Need storage aggregation");
            return true;
        }
    }

    @SuppressWarnings("unchecked")
    protected TupleFilter translateDerived(TupleFilter filter, Set<TblColRef> collector) {
        if (filter == null)
            return filter;

        if (filter instanceof CompareTupleFilter) {
            return translateDerivedInCompare((CompareTupleFilter) filter, collector);
        }

        List<TupleFilter> children = (List<TupleFilter>) filter.getChildren();
        List<TupleFilter> newChildren = Lists.newArrayListWithCapacity(children.size());
        boolean modified = false;
        for (TupleFilter child : children) {
            TupleFilter translated = translateDerived(child, collector);
            newChildren.add(translated);
            if (child != translated)
                modified = true;
        }
        if (modified) {
            filter = replaceChildren(filter, newChildren);
        }
        return filter;
    }

    private TupleFilter replaceChildren(TupleFilter filter, List<TupleFilter> newChildren) {
        if (filter instanceof LogicalTupleFilter) {
            LogicalTupleFilter r = new LogicalTupleFilter(filter.getOperator());
            r.addChildren(newChildren);
            return r;
        } else if (filter instanceof CaseTupleFilter) {
            CaseTupleFilter r = new CaseTupleFilter();
            r.addChildren(newChildren);
            return r;
        } else {
            throw new IllegalStateException("Cannot replaceChildren on " + filter);
        }
    }

    private TupleFilter translateDerivedInCompare(CompareTupleFilter compf, Set<TblColRef> collector) {
        if (compf.getColumn() == null)
            return compf;

        TblColRef derived = compf.getColumn();
        if (cubeDesc.isExtendedColumn(derived)) {
            throw new CubeDesc.CannotFilterExtendedColumnException(derived);
        }
        if (!cubeDesc.isDerived(derived))
            return compf;

        DeriveInfo hostInfo = cubeDesc.getHostInfo(derived);
//        ILookupTable lookup = cubeDesc.getHostInfo(derived).type == CubeDesc.DeriveType.PK_FK ? null
//                : getLookupStringTableForDerived(derived, hostInfo);
        Pair<TupleFilter, Boolean> translated = DerivedFilterTranslator.translate(hostInfo, compf);
//        try {
//            if (lookup != null) {
//                lookup.close();
//            }
//        } catch (IOException e) {
//            logger.error("error when close lookup table.", e);
//        }
        TupleFilter translatedFilter = translated.getFirst();
        boolean loosened = translated.getSecond();
        if (loosened) {
            collectColumnsRecursively(translatedFilter, collector);
        }
        return translatedFilter;
    }

//    @SuppressWarnings("unchecked")
//    protected ILookupTable getLookupStringTableForDerived(TblColRef derived, DeriveInfo hostInfo) {
//        CubeManager cubeMgr = CubeManager.getInstance(this.cubeInstance.getConfig());
//        CubeSegment seg = cubeInstance.getLatestReadySegment();
//        return cubeMgr.getLookupTable(seg, hostInfo.join);
//    }

    private void collectColumnsRecursively(TupleFilter filter, Set<TblColRef> collector) {
        if (filter == null)
            return;

        if (filter instanceof ColumnTupleFilter) {
            collector.add(((ColumnTupleFilter) filter).getColumn());
        }
        for (TupleFilter child : filter.getChildren()) {
            collectColumnsRecursively(child, collector);
        }
    }

    private void enableStorageLimitIfPossible(Cuboid cuboid, Collection<TblColRef> groups, List<TblColRef> dynGroups,
            Set<TblColRef> derivedPostAggregation, Collection<TblColRef> groupsD, TupleFilter filter,
            Set<TblColRef> loosenedColumnD, SQLDigest sqlDigest, StorageContext context) {
        Collection<FunctionDesc> functionDescs = sqlDigest.aggregations;

        StorageLimitLevel storageLimitLevel = StorageLimitLevel.LIMIT_ON_SCAN;

        //if groupsD is clustered at "head" of the rowkey, then limit push down is possible
        int size = groupsD.size();
        if (!groupsD.containsAll(cuboid.getColumns().subList(0, size))) {
            storageLimitLevel = StorageLimitLevel.LIMIT_ON_RETURN_SIZE;
            logger.debug(
                    "storageLimitLevel set to LIMIT_ON_RETURN_SIZE because groupD is not clustered at head, groupsD: {} with cuboid columns: {}",
                    groupsD, cuboid.getColumns());
        }

        if (!dynGroups.isEmpty()) {
            storageLimitLevel = StorageLimitLevel.NO_LIMIT;
            logger.debug("Storage limit push down is impossible because the query has dynamic groupby {}", dynGroups);
        }

        // derived aggregation is bad, unless expanded columns are already in group by
        if (!groups.containsAll(derivedPostAggregation)) {
            storageLimitLevel = StorageLimitLevel.NO_LIMIT;
            logger.debug("storageLimitLevel set to NO_LIMIT because derived column require post aggregation: {}",
                    derivedPostAggregation);
        }

        if (!TupleFilter.isEvaluableRecursively(filter)) {
            storageLimitLevel = StorageLimitLevel.NO_LIMIT;
            logger.debug("storageLimitLevel set to NO_LIMIT because the filter isn't evaluable");
        }

        if (!loosenedColumnD.isEmpty()) { // KYLIN-2173
            storageLimitLevel = StorageLimitLevel.NO_LIMIT;
            logger.debug("storageLimitLevel set to NO_LIMIT because filter is loosened: {}", loosenedColumnD);
        }

        if (context.hasSort()) {
            storageLimitLevel = StorageLimitLevel.NO_LIMIT;
            logger.debug("storageLimitLevel set to NO_LIMIT because the query has order by");
        }

        //if exists measures like max(cal_dt), then it's not a perfect cuboid match, cannot apply limit
        for (FunctionDesc functionDesc : functionDescs) {
            if (functionDesc.isDimensionAsMetric()) {
                storageLimitLevel = StorageLimitLevel.NO_LIMIT;
                logger.debug("storageLimitLevel set to NO_LIMIT because {} isDimensionAsMetric ", functionDesc);
            }
        }

        if (sqlDigest.groupByExpression) {
            storageLimitLevel = StorageLimitLevel.NO_LIMIT;
            logger.debug("storageLimitLevel set to NO_LIMIT because group by clause is an expression");
        }

        context.applyLimitPushDown(cubeInstance, storageLimitLevel);
    }

    private void enableStreamAggregateIfBeneficial(Cuboid cuboid, Set<TblColRef> groupsD, StorageContext context) {
        CubeDesc cubeDesc = cuboid.getCubeDesc();
        boolean enabled = cubeDesc.getConfig().isStreamAggregateEnabled();

        Set<TblColRef> shardByInGroups = Sets.newHashSet();
        for (TblColRef col : cubeDesc.getShardByColumns()) {
            if (groupsD.contains(col)) {
                shardByInGroups.add(col);
            }
        }
        if (!shardByInGroups.isEmpty()) {
            enabled = false;
            logger.debug("Aggregate partition results is not beneficial because shard by columns in groupD: {}",
                    shardByInGroups);
        }

        if (!context.isNeedStorageAggregation()) {
            enabled = false;
            logger.debug("Aggregate partition results is not beneficial because no storage aggregation");
        }

        if (enabled) {
            context.enableStreamAggregate();
        }
    }

    protected void notifyBeforeStorageQuery(SQLDigest sqlDigest) {
        Map<String, List<MeasureDesc>> map = Maps.newHashMap();
        for (MeasureDesc measure : cubeDesc.getMeasures()) {
            MeasureType<?> measureType = measure.getFunction().getMeasureType();

            String key = measureType.getClass().getCanonicalName();
            List<MeasureDesc> temp = null;
            if ((temp = map.get(key)) != null) {
                temp.add(measure);
            } else {
                map.put(key, Lists.<MeasureDesc> newArrayList(measure));
            }
        }

        for (List<MeasureDesc> sublist : map.values()) {
            sublist.get(0).getFunction().getMeasureType().adjustSqlDigest(sublist, sqlDigest);
        }
    }

    private TupleFilter checkHavingCanPushDown(TupleFilter havingFilter, Set<TblColRef> groupsD,
            List<FunctionDesc> aggregations, Set<FunctionDesc> metrics) {
        // must have only one segment
        Segments<CubeSegment> readySegs = cubeInstance.getSegments(SegmentStatusEnum.READY);
        if (readySegs.size() != 1) {
            logger.info("Can not push down having filter, must have only one segment");
            return null;
        }
        // sharded-by column must on group by
        CubeDesc desc = cubeInstance.getDescriptor();
        Set<TblColRef> shardBy = desc.getShardByColumns();
        if (groupsD == null || shardBy.isEmpty() || !groupsD.containsAll(shardBy))
            return null;

        // OK, push down
        logger.info("Push down having filter {}", havingFilter);

        // convert columns in the filter
        Set<TblColRef> aggrOutCols = new HashSet<>();
        TupleFilter.collectColumns(havingFilter, aggrOutCols);

        for (TblColRef aggrOutCol : aggrOutCols) {
            int aggrIdxOnSql = aggrOutCol.getColumnDesc().getZeroBasedIndex(); // aggr index marked in OLAPAggregateRel
            FunctionDesc aggrFunc = aggregations.get(aggrIdxOnSql);

            // calculate the index of this aggr among all the metrics that is sending to storage
            int aggrIdxAmongMetrics = 0;
            for (MeasureDesc m : cubeDesc.getMeasures()) {
                if (aggrFunc.equals(m.getFunction()))
                    break;
                if (metrics.contains(m.getFunction()))
                    aggrIdxAmongMetrics++;
            }
            aggrOutCol.getColumnDesc().setId("" + (aggrIdxAmongMetrics + 1));
        }
        return havingFilter;
    }

    private boolean isExactAggregation(StorageContext context, Cuboid cuboid, Collection<TblColRef> groups,
            Set<TblColRef> othersD, Set<TblColRef> singleValuesD, Set<TblColRef> derivedPostAggregation,
            Collection<FunctionDesc> functionDescs, List<SQLDigest.SQLCall> aggrSQLCalls, boolean groupByExpression) {
        if (context.isNeedStorageAggregation()) {
            logger.info("exactAggregation is false because need storage aggregation");
            return false;
        }

        if (cuboid.requirePostAggregation()) {
            logger.info("exactAggregation is false because cuboid {}=>{}", cuboid.getInputID(), cuboid.getId());
            return false;
        }

        // derived aggregation is bad, unless expanded columns are already in group by
        if (!groups.containsAll(derivedPostAggregation)) {
            logger.info("exactAggregation is false because derived column require post aggregation: {}",
                    derivedPostAggregation);
            return false;
        }

        // other columns (from filter) is bad, unless they are ensured to have single value
        if (!singleValuesD.containsAll(othersD)) {
            logger.info("exactAggregation is false because some column not on group by: {} (single value column: {})",
                    othersD, singleValuesD);
            return false;
        }

        //for DimensionAsMetric like max(cal_dt), the dimension column maybe not in real group by
        for (FunctionDesc functionDesc : functionDescs) {
            if (functionDesc.isDimensionAsMetric()) {
                logger.info("exactAggregation is false because has DimensionAsMetric");
                return false;
            }
        }
        for (SQLDigest.SQLCall aggrSQLCall : aggrSQLCalls) {
            if (aggrSQLCall.function.equals(BitmapMeasureType.FUNC_INTERSECT_COUNT_DISTINCT)
            || aggrSQLCall.function.equals(BitmapMeasureType.FUNC_INTERSECT_VALUE)) {
                logger.info("exactAggregation is false because has INTERSECT_COUNT OR INTERSECT_VALUE");
                return false;
            }
        }

        // for partitioned cube, the partition column must belong to group by or has single value
        PartitionDesc partDesc = cuboid.getCubeDesc().getModel().getPartitionDesc();
        if (partDesc.isPartitioned()) {
            TblColRef col = partDesc.getPartitionDateColumnRef();
            if (!groups.contains(col) && !singleValuesD.contains(col)) {
                logger.info("exactAggregation is false because cube is partitioned and {} is not on group by", col);
                return false;
            }
        }

        // for group by expression like: group by seller_id/100. seller_id_1(200) get 2, seller_id_2(201) also get 2, so can't aggregate exactly
        if (groupByExpression) {
            logger.info("exactAggregation is false because group by expression");
            return false;
        }

        logger.info("exactAggregation is true, cuboid id is {}", cuboid.getId());
        return true;
    }

}