/*
 * 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.datastore.page.encoding;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

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.compression.Compressor;
import org.apache.carbondata.core.datastore.compression.CompressorFactory;
import org.apache.carbondata.core.datastore.page.ColumnPage;
import org.apache.carbondata.core.datastore.page.ComplexColumnPage;
import org.apache.carbondata.core.datastore.page.encoding.compress.DirectCompressCodec;
import org.apache.carbondata.core.metadata.datatype.DataType;
import org.apache.carbondata.core.metadata.datatype.DataTypes;
import org.apache.carbondata.core.util.CarbonMetadataUtil;
import org.apache.carbondata.core.util.CarbonUtil;
import org.apache.carbondata.core.util.DataTypeUtil;
import org.apache.carbondata.format.BlockletMinMaxIndex;
import org.apache.carbondata.format.DataChunk2;
import org.apache.carbondata.format.Encoding;
import org.apache.carbondata.format.LocalDictionaryChunk;
import org.apache.carbondata.format.LocalDictionaryChunkMeta;
import org.apache.carbondata.format.PresenceMeta;

import static org.apache.carbondata.core.datastore.page.encoding.DefaultEncodingFactory.selectCodecByAlgorithmForFloating;
import static org.apache.carbondata.core.datastore.page.encoding.DefaultEncodingFactory.selectCodecByAlgorithmForIntegral;

import org.apache.log4j.Logger;

public abstract class ColumnPageEncoder {

  /**
   * logger
   */
  private static final Logger LOGGER =
      LogServiceFactory.getLogService(ColumnPageEncoder.class.getName());

  protected abstract ByteBuffer encodeData(ColumnPage input) throws IOException;

  protected abstract List<Encoding> getEncodingList();

  protected abstract ColumnPageEncoderMeta getEncoderMeta(ColumnPage inputPage);

  /**
   * Get the target data type of the page if encoded
   *
   * @param inputPage
   * @return
   */
  public DataType getTargetDataType(ColumnPage inputPage) {
    ColumnPageEncoderMeta encoderMeta = getEncoderMeta(inputPage);
    if (null != encoderMeta) {
      return encoderMeta.getStoreDataType();
    } else {
      return null;
    }
  }

  public Encoding getEncodingType() {
    List<Encoding> currEncodingList = getEncodingList();
    if (CarbonUtil.isEncodedWithMeta(currEncodingList)) {
      return currEncodingList.get(0);
    }
    return null;
  }

  /**
   * Return a encoded column page by encoding the input page
   * The encoded binary data and metadata are wrapped in encoding column page
   */
  public EncodedColumnPage encode(ColumnPage inputPage) throws IOException {
    ByteBuffer encodedBytes = encodeData(inputPage);
    DataChunk2 pageMetadata = buildPageMetadata(inputPage, encodedBytes);
    return new EncodedColumnPage(pageMetadata, encodedBytes, inputPage);
  }

  private DataChunk2 buildPageMetadata(ColumnPage inputPage, ByteBuffer encodedBytes)
      throws IOException {
    DataChunk2 dataChunk = new DataChunk2();
    dataChunk.setData_page_length(encodedBytes.limit() - encodedBytes.position());
    fillBasicFields(inputPage, dataChunk);
    fillNullBitSet(inputPage, dataChunk);
    fillEncoding(inputPage, dataChunk);
    fillMinMaxIndex(inputPage, dataChunk);
    fillLegacyFields(dataChunk);
    return dataChunk;
  }

  private void fillBasicFields(ColumnPage inputPage, DataChunk2 dataChunk)
      throws IOException {
    dataChunk.setChunk_meta(CarbonMetadataUtil.getChunkCompressorMeta(inputPage,
        dataChunk.getData_page_length()));
    dataChunk.setNumberOfRowsInpage(inputPage.getPageSize());
    dataChunk.setRowMajor(false);
  }

  private void fillNullBitSet(ColumnPage inputPage, DataChunk2 dataChunk) {
    PresenceMeta presenceMeta = new PresenceMeta();
    presenceMeta.setPresent_bit_streamIsSet(true);
    Compressor compressor = CompressorFactory.getInstance().getCompressor(
        inputPage.getColumnCompressorName());
    presenceMeta.setPresent_bit_stream(
        compressor.compressByte(inputPage.getNullBits().toByteArray()));
    dataChunk.setPresence(presenceMeta);
  }

  private void fillEncoding(ColumnPage inputPage, DataChunk2 dataChunk) throws IOException {
    dataChunk.setEncoders(getEncodingList());
    dataChunk.setEncoder_meta(buildEncoderMeta(inputPage));
  }

  private List<ByteBuffer> buildEncoderMeta(ColumnPage inputPage) throws IOException {
    ColumnPageEncoderMeta meta = getEncoderMeta(inputPage);
    List<ByteBuffer> metaData = new ArrayList<>();
    if (meta != null) {
      ByteArrayOutputStream stream = new ByteArrayOutputStream();
      DataOutputStream out = new DataOutputStream(stream);
      meta.write(out);
      metaData.add(ByteBuffer.wrap(stream.toByteArray()));
    }
    return metaData;
  }

  private void fillMinMaxIndex(ColumnPage inputPage, DataChunk2 dataChunk) {
    dataChunk.setMin_max(buildMinMaxIndex(inputPage, dataChunk.encoders));
  }

  private BlockletMinMaxIndex buildMinMaxIndex(ColumnPage inputPage, List<Encoding> encoders) {
    BlockletMinMaxIndex index = new BlockletMinMaxIndex();
    ByteBuffer max;
    ByteBuffer min;
    if (CarbonUtil.isEncodedWithMeta(encoders)
        && inputPage.getColumnSpec().getColumnType() == ColumnType.PLAIN_VALUE) {
      max = ByteBuffer.wrap(DataTypeUtil
          .getMinMaxBytesBasedOnDataTypeForNoDictionaryColumn(inputPage.getStatistics().getMax(),
              inputPage.getDataType()));
      min = ByteBuffer.wrap(DataTypeUtil
          .getMinMaxBytesBasedOnDataTypeForNoDictionaryColumn(inputPage.getStatistics().getMin(),
              inputPage.getDataType()));
    } else {
      byte[] bytes =
          CarbonUtil.getValueAsBytes(inputPage.getDataType(), inputPage.getStatistics().getMax());
      max = ByteBuffer.wrap(bytes);
      min = ByteBuffer.wrap(
          CarbonUtil.getValueAsBytes(inputPage.getDataType(), inputPage.getStatistics().getMin()));
    }
    index.addToMax_values(max);
    index.addToMin_values(min);
    index.addToMin_max_presence(inputPage.getStatistics().writeMinMax());
    return index;
  }

  /**
   * `buildPageMetadata` will call this for backward compatibility
   */
  protected void fillLegacyFields(DataChunk2 dataChunk) {
    // Subclass should override this to update DataChunk2 if any backward compatibility if required,
    // For example, when using IndexStorageCodec, rle_page_length and rowId_page_length need to be
    // updated
  }

  /**
   * Apply encoding algorithm for complex column page and return the coded data
   * TODO: remove this interface after complex column page is unified with column page
   */
  public static EncodedColumnPage[] encodeComplexColumn(ComplexColumnPage input)
      throws IOException {
    EncodedColumnPage[] encodedPages = new EncodedColumnPage[input.getComplexColumnIndex()];
    int index = 0;
    while (index < input.getComplexColumnIndex()) {
      ColumnPage subColumnPage = input.getColumnPage(index);
      encodedPages[index] = encodedColumn(subColumnPage);
      // by default add this encoding,
      // it is used for checking length of
      // complex child byte array columns (short and int)
      encodedPages[index].getPageMetadata().getEncoders()
          .add(Encoding.INT_LENGTH_COMPLEX_CHILD_BYTE_ARRAY);
      index++;
    }
    return encodedPages;
  }

  public static EncodedColumnPage encodedColumn(ColumnPage page) throws IOException {
    ColumnPageEncoder pageEncoder = createCodecForDimension(page);
    if (pageEncoder == null) {
      ColumnPageEncoder encoder = new DirectCompressCodec(DataTypes.BYTE_ARRAY).createEncoder(null);
      return encoder.encode(page);
    } else {
      LOGGER.debug("Encoder result ---> Source data type: " + pageEncoder.getEncoderMeta(page)
          .getColumnSpec().getSchemaDataType() + " Destination data type: " + pageEncoder
          .getEncoderMeta(page).getStoreDataType() + " for the column: " + pageEncoder
          .getEncoderMeta(page).getColumnSpec().getFieldName());

      return pageEncoder.encode(page);
    }
  }

  private static ColumnPageEncoder createCodecForDimension(ColumnPage inputPage) {
    TableSpec.ColumnSpec columnSpec = inputPage.getColumnSpec();
    if (columnSpec.getColumnType() == ColumnType.COMPLEX_PRIMITIVE) {
      if (inputPage.getDataType() == DataTypes.BYTE_ARRAY
          || inputPage.getDataType() == DataTypes.STRING) {
        // use legacy encoder
        return null;
      } else if ((inputPage.getDataType() == DataTypes.BYTE) || (inputPage.getDataType()
          == DataTypes.SHORT) || (inputPage.getDataType() == DataTypes.INT) || (
          inputPage.getDataType() == DataTypes.LONG)) {
        return selectCodecByAlgorithmForIntegral(inputPage.getStatistics(), true, columnSpec)
            .createEncoder(null);
      } else if ((inputPage.getDataType() == DataTypes.FLOAT) || (inputPage.getDataType()
          == DataTypes.DOUBLE)) {
        return selectCodecByAlgorithmForFloating(inputPage.getStatistics(), true, columnSpec)
            .createEncoder(null);
      }
    }
    // use legacy encoder
    return null;
  }

  /**
   * Below method to encode the dictionary page
   * @param dictionaryPage
   * dictionary column page
   * @return local dictionary chunk
   * @throws IOException
   * Problem in encoding
   */
  public LocalDictionaryChunk encodeDictionary(ColumnPage dictionaryPage)
      throws IOException {
    LocalDictionaryChunk localDictionaryChunk = new LocalDictionaryChunk();
    localDictionaryChunk.setDictionary_data(encodeData(dictionaryPage));
    LocalDictionaryChunkMeta localDictionaryChunkMeta = new LocalDictionaryChunkMeta();
    localDictionaryChunkMeta.setEncoders(getEncodingList());
    localDictionaryChunkMeta.setEncoder_meta(buildEncoderMeta(dictionaryPage));
    localDictionaryChunk.setDictionary_meta(localDictionaryChunkMeta);
    return localDictionaryChunk;
  }

}