/*
 * 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.carbondata.core.scan.collector.impl;

import java.math.BigDecimal;
import java.util.List;

import org.apache.carbondata.core.keygenerator.directdictionary.DirectDictionaryKeyGeneratorFactory;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonDimension;
import org.apache.carbondata.core.metadata.schema.table.column.CarbonMeasure;
import org.apache.carbondata.core.scan.executor.infos.BlockExecutionInfo;
import org.apache.carbondata.core.scan.executor.util.RestructureUtil;
import org.apache.carbondata.core.scan.result.BlockletScannedResult;
import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
import org.apache.carbondata.core.scan.result.vector.CarbonColumnarBatch;
import org.apache.carbondata.core.scan.result.vector.ColumnVectorInfo;

/**
 * It is not a collector it is just a scanned result holder.
 */
public class RestructureBasedVectorResultCollector extends DictionaryBasedVectorResultCollector {

  private Object[] measureDefaultValues = null;

  public RestructureBasedVectorResultCollector(BlockExecutionInfo blockExecutionInfos) {
    super(blockExecutionInfos);
    queryDimensions = executionInfo.getActualQueryDimensions();
    queryMeasures = executionInfo.getActualQueryMeasures();
    measureDefaultValues = new Object[queryMeasures.length];
    allColumnInfo = new ColumnVectorInfo[queryDimensions.length + queryMeasures.length];
    createVectorForNewlyAddedDimensions();
    createVectorForNewlyAddedMeasures();
    prepareDimensionAndMeasureColumnVectors();
  }

  /**
   * create column vector for newly added dimension columns
   */
  private void createVectorForNewlyAddedDimensions() {
    for (int i = 0; i < queryDimensions.length; i++) {
      if (!dimensionInfo.getDimensionExists()[i]) {
        // add a dummy column vector result collector object
        ColumnVectorInfo columnVectorInfo = new ColumnVectorInfo();
        columnVectorInfo.dimension = queryDimensions[i];
        if (queryDimensions[i].getDimension().getDataType().equals(DataTypes.TIMESTAMP)
            || queryDimensions[i].getDimension().getDataType().equals(DataTypes.DATE)) {
          columnVectorInfo.directDictionaryGenerator = DirectDictionaryKeyGeneratorFactory
              .getDirectDictionaryGenerator(queryDimensions[i].getDimension().getDataType());
        }
        allColumnInfo[queryDimensions[i].getOrdinal()] = columnVectorInfo;
      }
    }
  }

  /**
   * create column vector for newly added measure columns
   */
  private void createVectorForNewlyAddedMeasures() {
    for (int i = 0; i < queryMeasures.length; i++) {
      if (!measureInfo.getMeasureExists()[i]) {
        // add a dummy column vector result collector object
        ColumnVectorInfo columnVectorInfo = new ColumnVectorInfo();
        allColumnInfo[queryMeasures[i].getOrdinal()] = columnVectorInfo;
        columnVectorInfo.measure = queryMeasures[i];
        measureDefaultValues[i] = getMeasureDefaultValue(queryMeasures[i].getMeasure());
      }
    }
  }

  /**
   * Gets the default value for each CarbonMeasure
   *
   * @param carbonMeasure
   * @return
   */
  private Object getMeasureDefaultValue(CarbonMeasure carbonMeasure) {
    return RestructureUtil.getMeasureDefaultValueByType(carbonMeasure.getColumnSchema(),
        carbonMeasure.getDefaultValue());
  }

  @Override
  public List<Object[]> collectResultInRow(BlockletScannedResult scannedResult, int batchSize) {
    throw new UnsupportedOperationException("collectResultInRow is not supported here");
  }

  @Override
  public void collectResultInColumnarBatch(BlockletScannedResult scannedResult,
      CarbonColumnarBatch columnarBatch) {
    int numberOfPages = scannedResult.numberOfPages();
    while (scannedResult.getCurrentPageCounter() < numberOfPages) {
      int currentPageRowCount = scannedResult.getCurrentPageRowCount();
      if (currentPageRowCount == 0) {
        scannedResult.incrementPageCounter();
        continue;
      }
      int rowCounter = scannedResult.getRowCounter();
      int availableRows = currentPageRowCount - rowCounter;
      int requiredRows = columnarBatch.getBatchSize() - columnarBatch.getRowCounter();
      requiredRows = Math.min(requiredRows, availableRows);
      if (requiredRows < 1) {
        return;
      }
      fillColumnVectorDetails(columnarBatch, rowCounter, requiredRows);
      int filteredRows = scannedResult
          .markFilteredRows(columnarBatch, rowCounter, requiredRows, columnarBatch.getRowCounter());
      // fill default values for non existing dimensions and measures
      fillDataForNonExistingDimensions();
      fillDataForNonExistingMeasures();
      // fill existing dimensions and measures data
      fillResultToColumnarBatch(
          scannedResult, columnarBatch, rowCounter, availableRows, requiredRows);
      columnarBatch.setActualSize(columnarBatch.getActualSize() + requiredRows - filteredRows);
    }
  }

  /**
   * This method will fill the default values of non existing dimensions in the current block
   */
  private void fillDataForNonExistingDimensions() {
    for (int i = 0; i < executionInfo.getActualQueryDimensions().length; i++) {
      if (!dimensionInfo.getDimensionExists()[i]) {
        int queryOrder = executionInfo.getActualQueryDimensions()[i].getOrdinal();
        CarbonDimension dimension =
            executionInfo.getActualQueryDimensions()[i].getDimension();
        if (dimension.getDataType() == DataTypes.DATE) {
          // fill direct dictionary column data
          fillDirectDictionaryData(allColumnInfo[queryOrder].vector, allColumnInfo[queryOrder],
              dimensionInfo.getDefaultValues()[i]);
        } else {
          // fill no dictionary data
          fillNoDictionaryData(allColumnInfo[queryOrder].vector, allColumnInfo[queryOrder],
              dimensionInfo.getDefaultValues()[i]);
        }
      }
    }
  }

  /**
   * This method will fill the direct dictionary column data
   *
   * @param vector
   * @param columnVectorInfo
   * @param defaultValue
   */
  private void fillDirectDictionaryData(CarbonColumnVector vector,
      ColumnVectorInfo columnVectorInfo, Object defaultValue) {
    if (null != defaultValue) {
      if (columnVectorInfo.directDictionaryGenerator.getReturnType().equals(DataTypes.INT)) {
        vector.putInts(columnVectorInfo.vectorOffset, columnVectorInfo.size, (int) defaultValue);
      } else {
        vector.putLongs(columnVectorInfo.vectorOffset, columnVectorInfo.size, (long) defaultValue);
      }
    } else {
      vector.putNulls(columnVectorInfo.vectorOffset, columnVectorInfo.size);
    }
  }

  /**
   * This method will fill the no dictionary column data
   *
   * @param vector
   * @param columnVectorInfo
   * @param defaultValue
   */
  private void fillNoDictionaryData(CarbonColumnVector vector, ColumnVectorInfo columnVectorInfo,
      Object defaultValue) {
    if (null != defaultValue) {
      DataType dataType = columnVectorInfo.dimension.getDimension().getDataType();
      if (dataType == DataTypes.INT) {
        vector.putInts(columnVectorInfo.vectorOffset, columnVectorInfo.size, (int) defaultValue);
      } else if (dataType == DataTypes.LONG || dataType == DataTypes.TIMESTAMP) {
        vector.putLongs(columnVectorInfo.vectorOffset, columnVectorInfo.size, (long) defaultValue);
      } else {
        vector.putByteArray(columnVectorInfo.vectorOffset, columnVectorInfo.size,
            (byte[]) defaultValue);
      }
    } else {
      vector.putNulls(columnVectorInfo.vectorOffset, columnVectorInfo.size);
    }
  }

  /**
   * This method will fill the default values of non existing measures in the current block
   */
  private void fillDataForNonExistingMeasures() {
    for (int i = 0; i < executionInfo.getActualQueryMeasures().length; i++) {
      if (!measureInfo.getMeasureExists()[i]) {
        int queryOrder = executionInfo.getActualQueryMeasures()[i].getOrdinal();
        CarbonMeasure measure = executionInfo.getActualQueryMeasures()[i].getMeasure();
        ColumnVectorInfo columnVectorInfo = allColumnInfo[queryOrder];
        CarbonColumnVector vector = columnVectorInfo.vector;
        Object defaultValue = measureDefaultValues[i];
        if (null == defaultValue) {
          vector.putNulls(columnVectorInfo.vectorOffset, columnVectorInfo.size);
        } else {
          DataType dataType = measureInfo.getMeasureDataTypes()[i];
          if (dataType == DataTypes.SHORT) {
            vector.putShorts(columnVectorInfo.vectorOffset, columnVectorInfo.size,
                (short) defaultValue);
          } else if (dataType == DataTypes.INT) {
            vector
                .putInts(columnVectorInfo.vectorOffset, columnVectorInfo.size, (int) defaultValue);
          } else if (dataType == DataTypes.LONG) {
            vector.putLongs(columnVectorInfo.vectorOffset, columnVectorInfo.size,
                (long) defaultValue);
          } else if (DataTypes.isDecimal(dataType)) {
            vector.putDecimals(columnVectorInfo.vectorOffset, columnVectorInfo.size,
                (BigDecimal) defaultValue, measure.getPrecision());
          } else if (dataType == DataTypes.BOOLEAN) {
            vector.putBoolean(columnVectorInfo.vectorOffset, (Boolean) defaultValue);
          } else {
            vector.putDoubles(columnVectorInfo.vectorOffset, columnVectorInfo.size,
                (double) defaultValue);
          }
        }
      }
    }
  }

}