/*
 * 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.processing.store;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.datastore.ColumnType;
import org.apache.carbondata.core.datastore.TableSpec;
import org.apache.carbondata.core.datastore.page.ColumnPage;
import org.apache.carbondata.core.datastore.page.ComplexColumnPage;
import org.apache.carbondata.core.datastore.page.EncodedTablePage;
import org.apache.carbondata.core.datastore.page.encoding.ColumnPageEncoder;
import org.apache.carbondata.core.datastore.page.encoding.ColumnPageEncoderMeta;
import org.apache.carbondata.core.datastore.page.encoding.DefaultEncodingFactory;
import org.apache.carbondata.core.datastore.page.encoding.EncodedColumnPage;
import org.apache.carbondata.core.datastore.page.encoding.EncodingFactory;
import org.apache.carbondata.core.datastore.page.key.TablePageKey;
import org.apache.carbondata.core.datastore.page.statistics.KeyPageStatsCollector;
import org.apache.carbondata.core.datastore.page.statistics.LVLongStringStatsCollector;
import org.apache.carbondata.core.datastore.page.statistics.LVShortStringStatsCollector;
import org.apache.carbondata.core.datastore.page.statistics.PrimitivePageStatsCollector;
import org.apache.carbondata.core.datastore.row.CarbonRow;
import org.apache.carbondata.core.datastore.row.ComplexColumnInfo;
import org.apache.carbondata.core.datastore.row.WriteStepRowUtil;
import org.apache.carbondata.core.keygenerator.KeyGenException;
import org.apache.carbondata.core.localdictionary.generator.LocalDictionaryGenerator;
import org.apache.carbondata.core.memory.MemoryException;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.util.DataTypeUtil;
import org.apache.carbondata.processing.datatypes.GenericDataType;

import org.apache.log4j.Logger;

/**
 * Represent a page data for all columns, we store its data in columnar layout, so that
 * all processing apply to TablePage can be done in vectorized fashion.
 */
public class TablePage {

  private static final Logger LOGGER =
      LogServiceFactory.getLogService(TablePage.class.getName());

  // For all dimension and measure columns, we store the column data directly in the page,
  // the length of the page is the number of rows.

  // TODO: we should have separate class for key columns so that keys are stored together in
  // one vector to make it efficient for sorting
  private ColumnPage[] dictDimensionPages;
  private ColumnPage[] noDictDimensionPages;
  private ColumnPage[] measurePages;
  private ComplexColumnPage[] complexDimensionPages;

  // the num of rows in this page, it must be less than short value (65536)
  private int pageSize;

  private CarbonFactDataHandlerModel model;

  private TablePageKey key;

  private EncodedTablePage encodedTablePage;

  private EncodingFactory encodingFactory = DefaultEncodingFactory.getInstance();

  // true if it is last page of all input rows
  private boolean isLastPage;

  // used for complex column to deserilize the byte array in input CarbonRow
  private Map<Integer, GenericDataType> complexIndexMap = null;
  // name of compressor that used to compress column data,
  // currently all the columns share the same compressor.
  private String columnCompressor;

  TablePage(CarbonFactDataHandlerModel model, int pageSize) throws MemoryException {
    this.model = model;
    this.pageSize = pageSize;
    int numDictDimension = model.getMDKeyGenerator().getDimCount();
    TableSpec tableSpec = model.getTableSpec();
    this.columnCompressor = model.getColumnCompressor();

    dictDimensionPages = new ColumnPage[numDictDimension];
    noDictDimensionPages = new ColumnPage[model.getNoDictionaryCount()];
    int tmpNumDictDimIdx = 0;
    int tmpNumNoDictDimIdx = 0;
    for (int i = 0; i < dictDimensionPages.length + noDictDimensionPages.length; i++) {
      TableSpec.DimensionSpec spec = tableSpec.getDimensionSpec(i);
      ColumnType columnType = tableSpec.getDimensionSpec(i).getColumnType();
      ColumnPage page;
      if (ColumnType.DIRECT_DICTIONARY == columnType) {
        page = ColumnPage.newPage(
            new ColumnPageEncoderMeta(spec, DataTypes.BYTE_ARRAY, columnCompressor), pageSize);
        page.setStatsCollector(KeyPageStatsCollector.newInstance(DataTypes.BYTE_ARRAY));
        dictDimensionPages[tmpNumDictDimIdx++] = page;
      } else {
        // will be encoded using string page
        LocalDictionaryGenerator localDictionaryGenerator =
            model.getColumnLocalDictGenMap().get(spec.getFieldName());
        DataType dataType = DataTypes.STRING;
        if (DataTypes.VARCHAR == spec.getSchemaDataType()) {
          dataType = DataTypes.VARCHAR;
        } else if (DataTypes.BINARY == spec.getSchemaDataType()) {
          dataType = DataTypes.BINARY;
        }
        ColumnPageEncoderMeta columnPageEncoderMeta =
            new ColumnPageEncoderMeta(spec, dataType, columnCompressor);
        if (null != localDictionaryGenerator) {
          page = ColumnPage.newLocalDictPage(
              columnPageEncoderMeta, pageSize, localDictionaryGenerator, false);
        } else {
          if (DataTypeUtil.isPrimitiveColumn(spec.getSchemaDataType())) {
            if (spec.getSchemaDataType() == DataTypes.TIMESTAMP) {
              columnPageEncoderMeta =
                  new ColumnPageEncoderMeta(spec, DataTypes.LONG, columnCompressor);
            } else {
              columnPageEncoderMeta =
                  new ColumnPageEncoderMeta(spec, spec.getSchemaDataType(), columnCompressor);
            }
            // create the column page according to the data type for no dictionary numeric columns
            if (DataTypes.isDecimal(spec.getSchemaDataType())) {
              page = ColumnPage.newDecimalPage(columnPageEncoderMeta, pageSize);
            } else {
              page = ColumnPage.newPage(columnPageEncoderMeta, pageSize);
            }
          } else {
            page = ColumnPage.newPage(columnPageEncoderMeta, pageSize);
          }
        }
        // set the stats collector according to the data type of the columns
        if (DataTypes.VARCHAR == dataType || DataTypes.BINARY == dataType) {
          page.setStatsCollector(LVLongStringStatsCollector.newInstance());
        } else if (DataTypeUtil.isPrimitiveColumn(spec.getSchemaDataType())) {
          if (spec.getSchemaDataType() == DataTypes.TIMESTAMP) {
            page.setStatsCollector(PrimitivePageStatsCollector.newInstance(DataTypes.LONG));
          } else {
            page.setStatsCollector(
                PrimitivePageStatsCollector.newInstance(spec.getSchemaDataType()));
          }
        } else {
          page.setStatsCollector(LVShortStringStatsCollector.newInstance());
        }
        noDictDimensionPages[tmpNumNoDictDimIdx++] = page;
      }
    }
    complexDimensionPages = new ComplexColumnPage[model.getComplexColumnCount()];
    for (int i = 0; i < complexDimensionPages.length; i++) {
      // here we still do not the depth of the complex column, it will be initialized when
      // we get the first row.
      complexDimensionPages[i] = null;
    }
    measurePages = new ColumnPage[model.getMeasureCount()];
    DataType[] dataTypes = model.getMeasureDataType();
    for (int i = 0; i < measurePages.length; i++) {
      ColumnPageEncoderMeta columnPageEncoderMeta = new ColumnPageEncoderMeta(
          model.getTableSpec().getMeasureSpec(i), dataTypes[i], columnCompressor);
      ColumnPage page;
      if (DataTypes.isDecimal(columnPageEncoderMeta.getSchemaDataType())) {
        page = ColumnPage.newDecimalPage(columnPageEncoderMeta, pageSize);
      } else {
        page = ColumnPage.newPage(columnPageEncoderMeta, pageSize);
      }
      page.setStatsCollector(PrimitivePageStatsCollector.newInstance(dataTypes[i]));
      measurePages[i] = page;
    }

    boolean hasNoDictionary = noDictDimensionPages.length > 0;
    this.key = new TablePageKey(pageSize, model.getSegmentProperties(), hasNoDictionary);

    // for complex type, `complexIndexMap` is used in multithread (in multiple Producer),
    // we need to clone the index map to make it thread safe
    this.complexIndexMap = new HashMap<>();
    for (Map.Entry<Integer, GenericDataType> entry: model.getComplexIndexMap().entrySet()) {
      this.complexIndexMap.put(entry.getKey(), entry.getValue().deepCopy());
    }
  }

  /**
   * Add one row to the internal store
   *
   * @param rowId Id of the input row
   * @param row   row object
   */
  public void addRow(int rowId, CarbonRow row) throws KeyGenException {
    // convert each column category, update key and stats
    byte[] mdk = WriteStepRowUtil.getMdk(row, model.getMDKeyGenerator());
    convertToColumnarAndAddToPages(rowId, row, mdk);
    key.update(rowId, row, mdk);
  }

  // convert the input row object to columnar data and add to column pages
  private void convertToColumnarAndAddToPages(int rowId, CarbonRow row, byte[] mdk)
      throws KeyGenException {
    // 1. convert dictionary columns
    byte[][] keys = model.getSegmentProperties().getFixedLengthKeySplitter().splitKey(mdk);
    for (int i = 0; i < dictDimensionPages.length; i++) {
      dictDimensionPages[i].putData(rowId, keys[i]);
    }

    // 2. convert noDictionary columns and complex columns and varchar, binary columns.
    int noDictionaryCount = noDictDimensionPages.length;
    int complexColumnCount = complexDimensionPages.length;
    if (noDictionaryCount > 0 || complexColumnCount > 0) {
      TableSpec tableSpec = model.getTableSpec();
      List<TableSpec.DimensionSpec> noDictionaryDimensionSpec =
          tableSpec.getNoDictionaryDimensionSpec();
      Object[] noDictAndComplex = WriteStepRowUtil.getNoDictAndComplexDimension(row);
      for (int i = 0; i < noDictAndComplex.length; i++) {
        if (noDictionaryDimensionSpec.get(i).getSchemaDataType() == DataTypes.VARCHAR
            || noDictionaryDimensionSpec.get(i).getSchemaDataType() == DataTypes.BINARY) {
          byte[] valueWithLength = addIntLengthToByteArray((byte[]) noDictAndComplex[i]);
          noDictDimensionPages[i].putData(rowId, valueWithLength);
        } else if (i < noDictionaryCount) {
          if (DataTypeUtil
              .isPrimitiveColumn(noDictDimensionPages[i].getColumnSpec().getSchemaDataType())) {
            // put the actual data to the row
            Object value = noDictAndComplex[i];
            // in compaction flow the measure with decimal type will come as Spark decimal.
            // need to convert it to byte array.
            if (DataTypes.isDecimal(noDictDimensionPages[i].getDataType()) && model
                .isCompactionFlow() && value != null) {
              value = DataTypeUtil.getDataTypeConverter().convertFromDecimalToBigDecimal(value);
            }
            noDictDimensionPages[i].putData(rowId, value);
          } else {
            // noDictionary columns, since it is variable length, we need to prepare each
            // element as LV result byte array (first two bytes are the length of the array)
            byte[] valueWithLength = addShortLengthToByteArray((byte[]) noDictAndComplex[i]);
            noDictDimensionPages[i].putData(rowId, valueWithLength);
          }
        } else {
          // complex columns
          addComplexColumn(i - noDictionaryCount, rowId,
              (List<ArrayList<byte[]>>) noDictAndComplex[i]);
        }
      }
    }
    // 3. convert measure columns
    Object[] measureColumns = WriteStepRowUtil.getMeasure(row);
    for (int i = 0; i < measurePages.length; i++) {
      Object value = measureColumns[i];

      // in compaction flow the measure with decimal type will come as Spark decimal.
      // need to convert it to byte array.
      if (DataTypes.isDecimal(measurePages[i].getDataType()) &&
          model.isCompactionFlow() &&
          value != null) {
        value = DataTypeUtil.getDataTypeConverter().convertFromDecimalToBigDecimal(value);
      }
      measurePages[i].putData(rowId, value);
    }
  }

  /**
   * add a complex column into internal member complexDimensionPage
   *
   * @param index          index of the complexDimensionPage
   * @param rowId          Id of the input row
   * @param encodedComplexColumnar flatten data of complex column
   */
  // TODO: this function should be refactoried, ColumnPage should support complex type encoding
  // directly instead of doing it here
  private void addComplexColumn(int index, int rowId,
      List<ArrayList<byte[]>> encodedComplexColumnar) {
    GenericDataType complexDataType = complexIndexMap.get(
        index + model.getPrimitiveDimLens().length);
    // initialize the page if first row
    if (rowId == 0) {
      List<ComplexColumnInfo> complexColumnInfoList = new ArrayList<>();
      complexDataType.getComplexColumnInfo(complexColumnInfoList);
      complexDimensionPages[index] = new ComplexColumnPage(complexColumnInfoList);
      try {
        complexDimensionPages[index].initialize(
            model.getColumnLocalDictGenMap(), pageSize, columnCompressor);
      } catch (MemoryException e) {
        throw new RuntimeException(e);
      }
    }
    int depthInComplexColumn = complexDimensionPages[index].getComplexColumnIndex();
    for (int depth = 0; depth < depthInComplexColumn; depth++) {
      complexDimensionPages[index].putComplexData(depth, encodedComplexColumnar.get(depth));
    }
  }

  void freeMemory() {
    for (ColumnPage page : dictDimensionPages) {
      page.freeMemory();
    }
    for (ColumnPage page : noDictDimensionPages) {
      page.freeMemory();
    }
    for (ColumnPage page : measurePages) {
      page.freeMemory();
    }
    for (ComplexColumnPage page : complexDimensionPages) {
      if (null != page) {
        page.freeMemory();
      }
    }
  }

  // Adds length as a short element (first 2 bytes) to the head of the input byte array
  private byte[] addShortLengthToByteArray(byte[] input) {
    if (input.length > Short.MAX_VALUE) {
      throw new RuntimeException("input data length " + input.length +
          " bytes too long, maximum length supported is " + Short.MAX_VALUE + " bytes");
    }
    byte[] output = new byte[input.length + 2];
    ByteBuffer buffer = ByteBuffer.wrap(output);
    buffer.putShort((short)input.length);
    buffer.put(input, 0, input.length);
    return output;
  }

  // Adds length as a integer element (first 4 bytes) to the head of the input byte array
  private byte[] addIntLengthToByteArray(byte[] input) {
    byte[] output = new byte[input.length + 4];
    ByteBuffer buffer = ByteBuffer.wrap(output);
    buffer.putInt(input.length);
    buffer.put(input, 0, input.length);
    return output;
  }

  void encode() throws KeyGenException, MemoryException, IOException {
    // encode dimensions and measure
    EncodedColumnPage[] dimensions = encodeAndCompressDimensions();
    EncodedColumnPage[] measures = encodeAndCompressMeasures();
    this.encodedTablePage = EncodedTablePage.newInstance(pageSize, dimensions, measures, key);
  }

  public EncodedTablePage getEncodedTablePage() {
    return encodedTablePage;
  }

  // apply measure and set encodedData in `encodedData`
  private EncodedColumnPage[] encodeAndCompressMeasures()
      throws MemoryException, IOException {
    EncodedColumnPage[] encodedMeasures = new EncodedColumnPage[measurePages.length];
    for (int i = 0; i < measurePages.length; i++) {
      ColumnPageEncoder encoder = encodingFactory.createEncoder(
          model.getTableSpec().getMeasureSpec(i), measurePages[i]);
      encodedMeasures[i] = encoder.encode(measurePages[i]);
    }
    return encodedMeasures;
  }

  // apply and compress each dimension, set encoded data in `encodedData`
  private EncodedColumnPage[] encodeAndCompressDimensions()
      throws KeyGenException, IOException, MemoryException {
    List<EncodedColumnPage> encodedDimensions = new ArrayList<>();
    List<EncodedColumnPage> encodedComplexDimensions = new ArrayList<>();
    TableSpec tableSpec = model.getTableSpec();
    int dictIndex = 0;
    int noDictIndex = 0;
    int complexDimIndex = 0;
    int numDimensions = tableSpec.getNumDimensions();
    for (int i = 0; i < numDimensions; i++) {
      ColumnPageEncoder columnPageEncoder;
      EncodedColumnPage encodedPage;
      TableSpec.DimensionSpec spec = tableSpec.getDimensionSpec(i);
      switch (spec.getColumnType()) {
        case DIRECT_DICTIONARY:
          columnPageEncoder = encodingFactory.createEncoder(
              spec,
              dictDimensionPages[dictIndex]);
          encodedPage = columnPageEncoder.encode(dictDimensionPages[dictIndex++]);
          encodedDimensions.add(encodedPage);
          break;
        case PLAIN_VALUE:
          columnPageEncoder = encodingFactory.createEncoder(
              spec,
              noDictDimensionPages[noDictIndex]);
          encodedPage = columnPageEncoder.encode(noDictDimensionPages[noDictIndex]);
          if (LOGGER.isDebugEnabled()) {
            DataType targetDataType =
                columnPageEncoder.getTargetDataType(noDictDimensionPages[noDictIndex]);
            if (null != targetDataType) {
              LOGGER.debug(
                  "Encoder result ---> Source data type: " + noDictDimensionPages[noDictIndex]
                      .getDataType().getName() + " Destination data type: " + targetDataType
                      .getName() + " for the column: " + noDictDimensionPages[noDictIndex]
                      .getColumnSpec().getFieldName() + " having encoding type: "
                      + columnPageEncoder.getEncodingType());
            }
          }
          noDictIndex++;
          encodedDimensions.add(encodedPage);
          break;
        case COMPLEX:
          EncodedColumnPage[] encodedPages = ColumnPageEncoder.encodeComplexColumn(
              complexDimensionPages[complexDimIndex++]);
          encodedComplexDimensions.addAll(Arrays.asList(encodedPages));
          break;
        default:
          throw new IllegalArgumentException("unsupported dimension type:" + spec
              .getColumnType());
      }
    }

    encodedDimensions.addAll(encodedComplexDimensions);
    return encodedDimensions.toArray(new EncodedColumnPage[encodedDimensions.size()]);
  }

  /**
   * return column page of specified column name
   */
  public ColumnPage getColumnPage(String columnName) {
    int dictDimensionIndex = -1;
    int noDictDimensionIndex = -1;
    ColumnPage page = null;
    TableSpec spec = model.getTableSpec();
    int numDimensions = spec.getNumDimensions();
    for (int i = 0; i < numDimensions; i++) {
      ColumnType type = spec.getDimensionSpec(i).getColumnType();
      if (type == ColumnType.DIRECT_DICTIONARY) {
        page = dictDimensionPages[++dictDimensionIndex];
      } else if (type == ColumnType.PLAIN_VALUE) {
        page = noDictDimensionPages[++noDictDimensionIndex];
      } else {
        // do not support datamap on complex column
        continue;
      }
      String fieldName = spec.getDimensionSpec(i).getFieldName();
      if (fieldName.equalsIgnoreCase(columnName)) {
        return page;
      }
    }
    int numMeasures = spec.getNumMeasures();
    for (int i = 0; i < numMeasures; i++) {
      String fieldName = spec.getMeasureSpec(i).getFieldName();
      if (fieldName.equalsIgnoreCase(columnName)) {
        return measurePages[i];
      }
    }
    throw new IllegalArgumentException("DataMap: must have '" + columnName + "' column in schema");
  }

  public boolean isLastPage() {
    return isLastPage;
  }

  public void setIsLastPage(boolean isWriteAll) {
    this.isLastPage = isWriteAll;
  }

  public int getPageSize() {
    return pageSize;
  }
}