/*
 * 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.hbase.util;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.kylin.common.KylinConfig;
import org.apache.kylin.common.persistence.JsonSerializer;
import org.apache.kylin.common.persistence.ResourceStore;
import org.apache.kylin.common.persistence.Serializer;
import org.apache.kylin.common.util.Bytes;
import org.apache.kylin.common.util.DateFormat;
import org.apache.kylin.cube.CubeDescManager;
import org.apache.kylin.cube.CubeInstance;
import org.apache.kylin.cube.CubeManager;
import org.apache.kylin.cube.CubeSegment;
import org.apache.kylin.cube.model.CubeDesc;
import org.apache.kylin.metadata.MetadataManager;
import org.apache.kylin.metadata.model.DataModelDesc;
import org.apache.kylin.metadata.model.IEngineAware;
import org.apache.kylin.metadata.model.IStorageAware;
import org.apache.kylin.metadata.project.ProjectInstance;
import org.apache.kylin.metadata.project.ProjectManager;
import org.apache.kylin.metadata.project.RealizationEntry;
import org.apache.kylin.metadata.realization.RealizationStatusEnum;
import org.apache.kylin.metadata.realization.RealizationType;
import org.apache.kylin.storage.hbase.HBaseConnection;
import org.apache.kylin.storage.hybrid.HybridInstance;
import org.apache.kylin.storage.hybrid.HybridManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.Lists;

/**
 * Created by dongli on 12/29/15.
 */
public class ExtendCubeToHybridCLI {
    public static final String ACL_INFO_FAMILY = "i";
    private static final String CUBE_POSTFIX = "_old";
    private static final String HYBRID_POSTFIX = "_hybrid";
    private static final Logger logger = LoggerFactory.getLogger(ExtendCubeToHybridCLI.class);
    private static final String ACL_INFO_FAMILY_PARENT_COLUMN = "p";

    private KylinConfig kylinConfig;
    private CubeManager cubeManager;
    private CubeDescManager cubeDescManager;
    private MetadataManager metadataManager;
    private ResourceStore store;

    public ExtendCubeToHybridCLI() {
        this.kylinConfig = KylinConfig.getInstanceFromEnv();
        this.store = ResourceStore.getStore(kylinConfig);
        this.cubeManager = CubeManager.getInstance(kylinConfig);
        this.cubeDescManager = CubeDescManager.getInstance(kylinConfig);
        this.metadataManager = MetadataManager.getInstance(kylinConfig);
    }

    public static void main(String[] args) throws Exception {
        if (args.length != 2 && args.length != 3) {
            System.out.println("Usage: ExtendCubeToHybridCLI project cube [partition_date]");
            return;
        }

        ExtendCubeToHybridCLI tool = new ExtendCubeToHybridCLI();

        String projectName = args[0];
        String cubeName = args[1];
        String partitionDate = args.length == 3 ? args[2] : null;

        try {
            tool.createFromCube(projectName, cubeName, partitionDate);
            tool.verify();
            logger.info("Job Finished.");
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("Job Aborted.", e.getMessage());
        }
    }

    private boolean validateCubeInstance(CubeInstance cubeInstance) {
        if (cubeInstance == null) {
            logger.error("This cube does not exist.");
            return false;
        }
        if (cubeInstance.getSegments().isEmpty()) {
            logger.error("No segments in this cube, no need to extend.");
            return false;
        }
        return true;
    }

    public void createFromCube(String projectName, String cubeName, String partitionDateStr) throws Exception {
        logger.info("Create hybrid for cube[" + cubeName + "], project[" + projectName + "], partition_date[" + partitionDateStr + "].");

        CubeInstance cubeInstance = cubeManager.getCube(cubeName);
        if (!validateCubeInstance(cubeInstance)) {
            return;
        }

        CubeDesc cubeDesc = cubeDescManager.getCubeDesc(cubeInstance.getDescName());
        DataModelDesc dataModelDesc = metadataManager.getDataModelDesc(cubeDesc.getModelName());
        if (StringUtils.isEmpty(dataModelDesc.getPartitionDesc().getPartitionDateColumn())) {
            logger.error("No incremental cube, no need to extend.");
            return;
        }

        String owner = cubeInstance.getOwner();
        String dateFormat = dataModelDesc.getPartitionDesc().getPartitionDateFormat();
        long partitionDate = partitionDateStr != null ? DateFormat.stringToMillis(partitionDateStr, dateFormat) : 0;

        // get new name for old cube and cube_desc
        String newCubeDescName = renameCube(cubeDesc.getName());
        String newCubeInstanceName = renameCube(cubeInstance.getName());
        while (cubeDescManager.getCubeDesc(newCubeDescName) != null)
            newCubeDescName = renameCube(newCubeDescName);
        while (cubeManager.getCube(newCubeInstanceName) != null)
            newCubeInstanceName = renameCube(newCubeInstanceName);

        // create new cube_instance for old segments
        CubeInstance newCubeInstance = CubeInstance.getCopyOf(cubeInstance);
        newCubeInstance.setName(newCubeInstanceName);
        newCubeInstance.setDescName(newCubeDescName);
        newCubeInstance.updateRandomUuid();
        Iterator<CubeSegment> segmentIterator = newCubeInstance.getSegments().iterator();
        CubeSegment currentSeg = null;
        while (segmentIterator.hasNext()) {
            currentSeg = segmentIterator.next();
            if (partitionDateStr != null && (currentSeg.getDateRangeStart() >= partitionDate || currentSeg.getDateRangeEnd() > partitionDate)) {
                segmentIterator.remove();
                logger.info("CubeSegment[" + currentSeg + "] was removed.");
            }
        }
        if (partitionDateStr != null && partitionDate != currentSeg.getDateRangeEnd()) {
            logger.error("PartitionDate must be end date of one segment.");
            return;
        }
        if (currentSeg != null && partitionDateStr == null)
            partitionDate = currentSeg.getDateRangeEnd();

        cubeManager.createCube(newCubeInstance, projectName, owner);
        logger.info("CubeInstance was saved at: " + newCubeInstance.getResourcePath());

        // create new cube for old segments
        CubeDesc newCubeDesc = CubeDesc.getCopyOf(cubeDesc);
        newCubeDesc.setName(newCubeDescName);
        newCubeDesc.updateRandomUuid();
        newCubeDesc.init(kylinConfig, metadataManager.getAllTablesMap());
        newCubeDesc.setPartitionDateEnd(partitionDate);
        newCubeDesc.calculateSignature();
        cubeDescManager.createCubeDesc(newCubeDesc);
        logger.info("CubeDesc was saved at: " + newCubeDesc.getResourcePath());

        // update old cube_desc to new-version metadata
        cubeDesc.setPartitionDateStart(partitionDate);
        cubeDesc.setEngineType(IEngineAware.ID_MR_V2);
        cubeDesc.setStorageType(IStorageAware.ID_SHARDED_HBASE);
        cubeDesc.calculateSignature();
        cubeDescManager.updateCubeDesc(cubeDesc);
        logger.info("CubeDesc was saved at: " + cubeDesc.getResourcePath());

        // clear segments for old cube
        cubeInstance.setSegments(new ArrayList<CubeSegment>());
        cubeInstance.setStatus(RealizationStatusEnum.DISABLED);
        store.putResource(cubeInstance.getResourcePath(), cubeInstance, CubeManager.CUBE_SERIALIZER);
        logger.info("CubeInstance was saved at: " + cubeInstance.getResourcePath());

        // create hybrid model for these two cubes
        List<RealizationEntry> realizationEntries = Lists.newArrayListWithCapacity(2);
        realizationEntries.add(RealizationEntry.create(RealizationType.CUBE, cubeInstance.getName()));
        realizationEntries.add(RealizationEntry.create(RealizationType.CUBE, newCubeInstance.getName()));
        HybridInstance hybridInstance = HybridInstance.create(kylinConfig, renameHybrid(cubeInstance.getName()), realizationEntries);
        store.putResource(hybridInstance.getResourcePath(), hybridInstance, HybridManager.HYBRID_SERIALIZER);
        ProjectManager.getInstance(kylinConfig).moveRealizationToProject(RealizationType.HYBRID, hybridInstance.getName(), projectName, owner);
        logger.info("HybridInstance was saved at: " + hybridInstance.getResourcePath());

        // copy Acl from old cube to new cube
        copyAcl(cubeInstance.getId(), newCubeInstance.getId(), projectName);
        logger.info("Acl copied from [" + cubeName + "] to [" + newCubeInstanceName + "].");
    }

    private void verify() {
        CubeDescManager.clearCache();
        CubeDescManager.getInstance(kylinConfig);

        CubeManager.clearCache();
        CubeManager.getInstance(kylinConfig);

        ProjectManager.clearCache();
        ProjectManager.getInstance(kylinConfig);

        HybridManager.clearCache();
        HybridManager.getInstance(kylinConfig);
    }

    private String renameCube(String origName) {
        return origName + CUBE_POSTFIX;
    }

    private String renameHybrid(String origName) {
        return origName + HYBRID_POSTFIX;
    }

    private void copyAcl(String origCubeId, String newCubeId, String projectName) throws Exception {
        String projectResPath = ProjectInstance.concatResourcePath(projectName);
        Serializer<ProjectInstance> projectSerializer = new JsonSerializer<ProjectInstance>(ProjectInstance.class);
        ProjectInstance project = store.getResource(projectResPath, ProjectInstance.class, projectSerializer);
        String projUUID = project.getUuid();
        Table aclHtable = null;
        try {
            aclHtable = HBaseConnection.get(kylinConfig.getStorageUrl()).getTable(TableName.valueOf(kylinConfig.getMetadataUrlPrefix() + "_acl"));

            // cube acl
            Result result = aclHtable.get(new Get(Bytes.toBytes(origCubeId)));
            if (result.listCells() != null) {
                for (Cell cell : result.listCells()) {
                    byte[] family = CellUtil.cloneFamily(cell);
                    byte[] column = CellUtil.cloneQualifier(cell);
                    byte[] value = CellUtil.cloneValue(cell);

                    // use the target project uuid as the parent
                    if (Bytes.toString(family).equals(ACL_INFO_FAMILY) && Bytes.toString(column).equals(ACL_INFO_FAMILY_PARENT_COLUMN)) {
                        String valueString = "{\"id\":\"" + projUUID + "\",\"type\":\"org.apache.kylin.metadata.project.ProjectInstance\"}";
                        value = Bytes.toBytes(valueString);
                    }
                    Put put = new Put(Bytes.toBytes(newCubeId));
                    put.add(family, column, value);
                    aclHtable.put(put);
                }
            }
        } finally {
            IOUtils.closeQuietly(aclHtable);
        }
    }
}