/*
 * 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.ArrayList;
import java.util.BitSet;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;

import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.debug.BackdoorToggles;
import org.apache.kylin.common.util.ByteArray;
import org.apache.kylin.common.util.ImmutableBitSet;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.common.FuzzyValueCombination;
import org.apache.kylin.cube.cuboid.Cuboid;
import org.apache.kylin.cube.gridtable.CubeGridTable;
import org.apache.kylin.cube.gridtable.CuboidToGridTableMapping;
import org.apache.kylin.cube.gridtable.RecordComparators;
import org.apache.kylin.cube.gridtable.ScanRangePlannerBase;
import org.apache.kylin.cube.kv.CubeDimEncMap;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.gridtable.GTInfo;
import org.apache.kylin.gridtable.GTRecord;
import org.apache.kylin.gridtable.GTScanRange;
import org.apache.kylin.gridtable.GTScanRequest;
import org.apache.kylin.gridtable.GTScanRequestBuilder;
import org.apache.kylin.gridtable.GTTwoLayerAggregateParam;
import org.apache.kylin.gridtable.GTUtil;
import org.apache.kylin.gridtable.IGTComparator;
import org.apache.kylin.metadata.expression.TupleExpression;
import org.apache.kylin.metadata.filter.TupleFilter;
import org.apache.kylin.metadata.model.DynamicFunctionDesc;
import org.apache.kylin.metadata.model.ExpressionDynamicFunctionDesc;
import org.apache.kylin.metadata.model.FunctionDesc;
import org.apache.kylin.metadata.model.TblColRef;
import org.apache.kylin.metadata.model.TwoLayerFunctionDesc;
import org.apache.kylin.storage.StorageContext;
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 class CubeScanRangePlanner extends ScanRangePlannerBase {

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

    protected int maxScanRanges;
    protected int maxFuzzyKeysPerSplit;
    protected int maxFuzzyKeys;

    //non-GT
    protected CubeSegment cubeSegment;
    protected CubeDesc cubeDesc;
    protected Cuboid cuboid;

    public CubeScanRangePlanner(CubeSegment cubeSegment, Cuboid cuboid, TupleFilter filter, Set<TblColRef> dimensions, //
            Set<TblColRef> groupByDims, List<TblColRef> dynGroupsDims, List<TupleExpression> dynGroupExprs, //
            Collection<FunctionDesc> metrics, List<DynamicFunctionDesc> dynFuncs, //
            TupleFilter havingFilter, StorageContext context) {
        this.context = context;

        this.maxScanRanges = cubeSegment.getConfig().getQueryStorageVisitScanRangeMax();
        this.maxFuzzyKeysPerSplit = cubeSegment.getConfig().getQueryScanFuzzyKeyMax();
        this.maxFuzzyKeys = maxFuzzyKeysPerSplit * cubeSegment.getConfig().getQueryScanFuzzyKeySplitMax();

        this.cubeSegment = cubeSegment;
        this.cubeDesc = cubeSegment.getCubeDesc();
        this.cuboid = cuboid;

        final CuboidToGridTableMapping mapping = context.getMapping();

        this.gtInfo = CubeGridTable.newGTInfo(cuboid, new CubeDimEncMap(cubeSegment), mapping);

        IGTComparator comp = gtInfo.getCodeSystem().getComparator();
        //start key GTRecord compare to start key GTRecord
        this.rangeStartComparator = RecordComparators.getRangeStartComparator(comp);
        //stop key GTRecord compare to stop key GTRecord
        this.rangeEndComparator = RecordComparators.getRangeEndComparator(comp);
        //start key GTRecord compare to stop key GTRecord
        this.rangeStartEndComparator = RecordComparators.getRangeStartEndComparator(comp);

        //replace the constant values in filter to dictionary codes
        Set<TblColRef> groupByPushDown = Sets.newHashSet(groupByDims);
        groupByPushDown.addAll(dynGroupsDims);
        this.gtFilter = GTUtil.convertFilterColumnsAndConstants(filter, gtInfo, mapping.getDim2gt(), groupByPushDown);
        this.havingFilter = havingFilter;

        this.gtDimensions = mapping.makeGridTableColumns(dimensions);
        this.gtAggrGroups = mapping
                .makeGridTableColumns(replaceDerivedColumns(groupByPushDown, cubeSegment.getCubeDesc()));
        this.gtAggrMetrics = mapping.makeGridTableColumns(metrics);
        this.gtAggrFuncs = mapping.makeAggrFuncs(metrics);

        // for dynamic cols, which are as appended columns to GTInfo
        BitSet tmpGtDynCols = new BitSet();

        this.tupleExpressionMap = Maps.newHashMap();
        // for dynamic dimensions
        for (int i = 0; i < dynGroupsDims.size(); i++) {
            int c = mapping.getIndexOf(dynGroupsDims.get(i));
            tmpGtDynCols.set(c);
            TupleExpression tupleExpr = GTUtil.convertFilterColumnsAndConstants(dynGroupExprs.get(i), gtInfo, mapping,
                    groupByPushDown);
            this.tupleExpressionMap.put(c, tupleExpr);
        }

        // for dynamic measures
        Set<FunctionDesc> tmpRtAggrMetrics = Sets.newHashSet();
        Set<TwoLayerFunctionDesc> twoLayerFuncs = Sets.newHashSet();
        Set<TblColRef> vanishDims = Sets.newHashSet();
        for (DynamicFunctionDesc dynFunc : dynFuncs) {
            int c = mapping.getIndexOf(dynFunc);
            tmpGtDynCols.set(c);
            tmpRtAggrMetrics.addAll(dynFunc.getRuntimeFuncMap().values());
            if (dynFunc instanceof ExpressionDynamicFunctionDesc) {
                TupleExpression tupleExpr = ((ExpressionDynamicFunctionDesc) dynFunc).getTupleExpression();
                this.tupleExpressionMap.put(c, GTUtil.convertFilterColumnsAndConstants(tupleExpr, gtInfo, mapping,
                        dynFunc.getRuntimeFuncMap(), groupByPushDown));
                tupleExpr = GTUtil.convertFilterColumnsAndConstants(tupleExpr, gtInfo, mapping,
                        dynFunc.getRuntimeFuncMap(), groupByPushDown);
                this.tupleExpressionMap.put(c, tupleExpr);
            } else if (dynFunc instanceof TwoLayerFunctionDesc) {
                TwoLayerFunctionDesc twoLayerFunc = (TwoLayerFunctionDesc) dynFunc;
                twoLayerFuncs.add(twoLayerFunc);
                vanishDims.addAll(twoLayerFunc.getRuntimeDimensions());
            }
        }
        this.gtDynColumns = new ImmutableBitSet(tmpGtDynCols);
        this.gtRtAggrMetrics = mapping.makeGridTableColumns(tmpRtAggrMetrics);

        //// for two layer measures
        ImmutableBitSet vDimMask = mapping.makeGridTableColumns(vanishDims);
        ImmutableBitSet outLayerMetrics = mapping.makeGridTableColumns(twoLayerFuncs);
        String[] outLayerMetricsFuncs = mapping.makeAggrFuncs(twoLayerFuncs);

        List<TwoLayerFunctionDesc> outLFuncList = Lists.newArrayList(twoLayerFuncs);
        int[] outLIndexes = mapping.getMetricsIndexes(outLFuncList);
        Map<Integer, TwoLayerFunctionDesc> outLFuncMap = Maps.newHashMap();
        for (int i = 0; i < outLIndexes.length; i++) {
            outLFuncMap.put(outLIndexes[i], outLFuncList.get(i));
        }

        int[] inLayerMetrics = new int[outLayerMetrics.trueBitCount()];
        String[] inLayerMetricsFuncs = new String[outLayerMetrics.trueBitCount()];
        for (int i = 0; i < outLayerMetrics.trueBitCount(); i++) {
            int c = outLayerMetrics.trueBitAt(i);
            FunctionDesc func = outLFuncMap.get(c).getRuntimeFuncMap().values().iterator().next();
            inLayerMetrics[i] = mapping.getIndexOf(func);
            inLayerMetricsFuncs[i] = func.getExpression();
        }

        this.twoLayerAggParam = new GTTwoLayerAggregateParam(vDimMask, outLayerMetrics, outLayerMetricsFuncs,
                inLayerMetrics, inLayerMetricsFuncs);

        this.gtAggrGroups = mapping
                .makeGridTableColumns(replaceDerivedColumns(groupByPushDown, cubeSegment.getCubeDesc()));
        this.gtAggrMetrics = mapping.makeGridTableColumns(metrics);
        this.gtAggrFuncs = mapping.makeAggrFuncs(metrics);
    }

    protected StorageContext context;

    /**
     * Construct  GTScanRangePlanner with incomplete information. For UT only.
     */
    public CubeScanRangePlanner(GTInfo info, TblColRef gtPartitionCol, TupleFilter gtFilter) {

        this.maxScanRanges = KylinConfig.getInstanceFromEnv().getQueryStorageVisitScanRangeMax();
        this.maxFuzzyKeysPerSplit = KylinConfig.getInstanceFromEnv().getQueryScanFuzzyKeyMax();
        this.maxFuzzyKeys = maxFuzzyKeysPerSplit * KylinConfig.getInstanceFromEnv().getQueryScanFuzzyKeySplitMax();

        this.gtInfo = info;

        IGTComparator comp = gtInfo.getCodeSystem().getComparator();
        //start key GTRecord compare to start key GTRecord
        this.rangeStartComparator = RecordComparators.getRangeStartComparator(comp);
        //stop key GTRecord compare to stop key GTRecord
        this.rangeEndComparator = RecordComparators.getRangeEndComparator(comp);
        //start key GTRecord compare to stop key GTRecord
        this.rangeStartEndComparator = RecordComparators.getRangeStartEndComparator(comp);

        this.gtFilter = gtFilter;
    }

    public GTScanRequest planScanRequest() {
        GTScanRequest scanRequest;
        List<GTScanRange> scanRanges = this.planScanRanges();
        if (scanRanges != null && !scanRanges.isEmpty()) {
            scanRequest = new GTScanRequestBuilder().setInfo(gtInfo).setRanges(scanRanges).setDimensions(gtDimensions)
                    .setAggrGroupBy(gtAggrGroups).setAggrMetrics(gtAggrMetrics).setAggrMetricsFuncs(gtAggrFuncs)
                    .setFilterPushDown(gtFilter)//
                    .setRtAggrMetrics(gtRtAggrMetrics).setDynamicColumns(gtDynColumns)
                    .setExprsPushDown(tupleExpressionMap)//
                    .setTwoLayerAggregateParam(twoLayerAggParam)//
                    .setAllowStorageAggregation(context.isNeedStorageAggregation())
                    .setAggCacheMemThreshold(cubeSegment.getConfig().getQueryCoprocessorMemGB())//
                    .setStoragePushDownLimit(context.getFinalPushDownLimit())
                    .setStorageLimitLevel(context.getStorageLimitLevel()).setHavingFilterPushDown(havingFilter)
                    .createGTScanRequest();
        } else {
            scanRequest = null;
        }
        return scanRequest;
    }

    /**
     * Overwrite this method to provide smarter storage visit plans
     * @return
     */
    public List<GTScanRange> planScanRanges() {
        TupleFilter flatFilter = flattenToOrAndFilter(gtFilter);

        List<Collection<ColumnRange>> orAndDimRanges = translateToOrAndDimRanges(flatFilter);

        List<GTScanRange> scanRanges = Lists.newArrayListWithCapacity(orAndDimRanges.size());
        for (Collection<ColumnRange> andDimRanges : orAndDimRanges) {
            GTScanRange scanRange = newScanRange(andDimRanges);
            if (scanRange != null)
                scanRanges.add(scanRange);
        }

        List<GTScanRange> mergedRanges = mergeOverlapRanges(scanRanges);
        mergedRanges = splitFuzzyKeys(mergedRanges);
        mergedRanges = mergeTooManyRanges(mergedRanges, maxScanRanges);

        return mergedRanges;
    }

    private Set<TblColRef> replaceDerivedColumns(Set<TblColRef> input, CubeDesc cubeDesc) {
        Set<TblColRef> ret = Sets.newHashSet();
        for (TblColRef col : input) {
            if (cubeDesc.hasHostColumn(col)) {
                for (TblColRef host : cubeDesc.getHostInfo(col).columns) {
                    ret.add(host);
                }
            } else {
                ret.add(col);
            }
        }
        return ret;
    }

    protected GTScanRange newScanRange(Collection<ColumnRange> andDimRanges) {
        GTRecord pkStart = new GTRecord(gtInfo);
        GTRecord pkEnd = new GTRecord(gtInfo);
        Map<Integer, Set<ByteArray>> fuzzyValues = Maps.newHashMap();

        for (ColumnRange range : andDimRanges) {

            int col = range.column.getColumnDesc().getZeroBasedIndex();
            if (!gtInfo.getPrimaryKey().get(col))
                continue;

            pkStart.set(col, range.begin);
            pkEnd.set(col, range.end);

            if (range.valueSet != null && !range.valueSet.isEmpty()) {
                fuzzyValues.put(col, range.valueSet);
            }
        }

        List<GTRecord> fuzzyKeys = buildFuzzyKeys(fuzzyValues);

        return new GTScanRange(pkStart, pkEnd, fuzzyKeys);
    }

    private List<GTRecord> buildFuzzyKeys(Map<Integer, Set<ByteArray>> fuzzyValueSet) {
        ArrayList<GTRecord> result = Lists.newArrayList();

        if (fuzzyValueSet.isEmpty())
            return result;

        // debug/profiling purpose
        if (BackdoorToggles.getDisableFuzzyKey()) {
            logger.info("The execution of this query will not use fuzzy key");
            return result;
        }

        List<Map<Integer, ByteArray>> fuzzyValueCombinations = FuzzyValueCombination.calculate(fuzzyValueSet,
                maxFuzzyKeys);
        for (Map<Integer, ByteArray> fuzzyValue : fuzzyValueCombinations) {

            GTRecord fuzzy = new GTRecord(gtInfo);
            for (Map.Entry<Integer, ByteArray> entry : fuzzyValue.entrySet()) {
                fuzzy.set(entry.getKey(), entry.getValue());
            }

            result.add(fuzzy);
        }
        return result;
    }

    protected List<GTScanRange> mergeOverlapRanges(List<GTScanRange> ranges) {
        if (ranges.size() <= 1) {
            return ranges;
        }

        // sort ranges by start key
        Collections.sort(ranges, new Comparator<GTScanRange>() {
            @Override
            public int compare(GTScanRange a, GTScanRange b) {
                return rangeStartComparator.compare(a.pkStart, b.pkStart);
            }
        });

        // merge the overlap range
        List<GTScanRange> mergedRanges = new ArrayList<GTScanRange>();
        int mergeBeginIndex = 0;
        GTRecord mergeEnd = ranges.get(0).pkEnd;
        for (int index = 1; index < ranges.size(); index++) {
            GTScanRange range = ranges.get(index);

            // if overlap, swallow it
            if (rangeStartEndComparator.compare(range.pkStart, mergeEnd) <= 0) {
                mergeEnd = rangeEndComparator.max(mergeEnd, range.pkEnd);
                continue;
            }

            // not overlap, split here
            GTScanRange mergedRange = mergeKeyRange(ranges.subList(mergeBeginIndex, index));
            mergedRanges.add(mergedRange);

            // start new split
            mergeBeginIndex = index;
            mergeEnd = range.pkEnd;
        }

        // don't miss the last range
        GTScanRange mergedRange = mergeKeyRange(ranges.subList(mergeBeginIndex, ranges.size()));
        mergedRanges.add(mergedRange);

        return mergedRanges;
    }

    private GTScanRange mergeKeyRange(List<GTScanRange> ranges) {
        GTScanRange first = ranges.get(0);
        if (ranges.size() == 1)
            return first;

        GTRecord start = first.pkStart;
        GTRecord end = first.pkEnd;
        Set<GTRecord> newFuzzyKeys = Sets.newLinkedHashSet();

        boolean hasNonFuzzyRange = false;
        for (GTScanRange range : ranges) {
            hasNonFuzzyRange = hasNonFuzzyRange || range.fuzzyKeys.isEmpty();
            newFuzzyKeys.addAll(range.fuzzyKeys);
            end = rangeEndComparator.max(end, range.pkEnd);
        }

        // if any range is non-fuzzy, then all fuzzy keys must be cleared
        // too many fuzzy keys will slow down HBase scan
        if (hasNonFuzzyRange || newFuzzyKeys.size() > maxFuzzyKeys) {
            if (newFuzzyKeys.size() > maxFuzzyKeys) {
                logger.debug("too many FuzzyKeys,  clean it!");
            }
            newFuzzyKeys.clear();
        }

        return new GTScanRange(start, end, Lists.newArrayList(newFuzzyKeys));
    }

    protected List<GTScanRange> mergeTooManyRanges(List<GTScanRange> ranges, int maxRanges) {
        if (ranges.size() <= maxRanges) {
            return ranges;
        }

        // TODO: check the distance between range and merge the large distance range
        List<GTScanRange> result = new ArrayList<GTScanRange>(1);
        GTScanRange mergedRange = mergeKeyRange(ranges);
        result.add(mergedRange);

        result = splitFuzzyKeys(result);
        return result;
    }

    private List<GTScanRange> splitFuzzyKeys(List<GTScanRange> mergedRanges) {
        List<GTScanRange> result = Lists.newArrayList();
        for (GTScanRange range : mergedRanges) {
            // if the fuzzy key is huge but still within in split range, then we split fuzzy keys to multiple ones.
            if (range.fuzzyKeys.size() > maxFuzzyKeysPerSplit && range.fuzzyKeys.size() <= maxFuzzyKeys) {
                List<GTRecord> fuzzyKeys = range.fuzzyKeys;
                Collections.sort(fuzzyKeys);
                int nSplit = (fuzzyKeys.size() - 1) / maxFuzzyKeysPerSplit + 1;
                int nFuzzyKeysPerSplit = fuzzyKeys.size() / nSplit;
                int startIndex = 0;
                for (int i = 1; i <= nSplit; i++) {
                    int endIndex = i == nSplit ? fuzzyKeys.size() : i * nFuzzyKeysPerSplit;
                    List<GTRecord> subFuzzyKeys = fuzzyKeys.subList(startIndex, endIndex);
                    result.add(new GTScanRange(range.pkStart, range.pkEnd, subFuzzyKeys));
                    startIndex = endIndex;
                }
                logger.debug(String.format(Locale.ROOT, "large FuzzyKeys split size : %d", result.size()));
            } else {
                result.add(range);
            }
        }
        return result;
    }

    public int getMaxScanRanges() {
        return maxScanRanges;
    }

    public void setMaxScanRanges(int maxScanRanges) {
        this.maxScanRanges = maxScanRanges;
    }

}
