/*
 * 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.result.iterator;

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

import org.apache.carbondata.common.CarbonIterator;
import org.apache.carbondata.common.logging.LogServiceFactory;
import org.apache.carbondata.core.datastore.block.SegmentProperties;
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.util.RestructureUtil;
import org.apache.carbondata.core.scan.result.RowBatch;
import org.apache.carbondata.core.scan.wrappers.ByteArrayWrapper;
import org.apache.carbondata.core.util.DataTypeUtil;

import org.apache.log4j.Logger;

/**
 * This is a wrapper iterator over the detail raw query iterator.
 * This iterator will handle the processing of the raw rows.
 * This will handle the batch results and will iterate on the batches and give single row.
 */
public class ColumnDriftRawResultIterator extends RawResultIterator {

  // column reorder for no-dictionary column
  private int noDictCount;
  private int[] noDictMap;
  // column drift
  private boolean[] isColumnDrift;
  private int measureCount;
  private DataType[] measureDataTypes;

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

  public ColumnDriftRawResultIterator(CarbonIterator<RowBatch> detailRawQueryResultIterator,
      SegmentProperties sourceSegProperties, SegmentProperties destinationSegProperties) {
    super(detailRawQueryResultIterator, sourceSegProperties, destinationSegProperties, false);
    initForColumnDrift();
    init();
  }

  private void initForColumnDrift() {
    List<CarbonDimension> noDictDims =
        new ArrayList<>(destinationSegProperties.getDimensions().size());
    for (CarbonDimension dimension : destinationSegProperties.getDimensions()) {
      if (dimension.getNumberOfChild() == 0) {
        if (dimension.getDataType() != DataTypes.DATE) {
          noDictDims.add(dimension);
        }
      }
    }
    measureCount = destinationSegProperties.getMeasures().size();
    noDictCount = noDictDims.size();
    isColumnDrift = new boolean[noDictCount];
    noDictMap = new int[noDictCount];
    measureDataTypes = new DataType[noDictCount];
    List<CarbonMeasure> sourceMeasures = sourceSegProperties.getMeasures();
    for (int i = 0; i < noDictCount; i++) {
      for (CarbonMeasure sourceMeasure : sourceMeasures) {
        if (RestructureUtil.isColumnMatches(true, noDictDims.get(i), sourceMeasure)) {
          isColumnDrift[i] = true;
          measureDataTypes[i] = sourceMeasure.getDataType();
          break;
        }
      }
    }
    int noDictIndex = 0;
    // the column drift are at the end of measures
    int measureIndex = measureCount + 1;
    for (int i = 0; i < noDictCount; i++) {
      if (isColumnDrift[i]) {
        noDictMap[i] = measureIndex++;
      } else {
        noDictMap[i] = noDictIndex++;
      }
    }
  }

  @Override
  protected Object[] convertRow(Object[] rawRow) {
    super.convertRow(rawRow);
    ByteArrayWrapper dimObject = (ByteArrayWrapper) rawRow[0];
    // need move measure to dimension and return new row by current schema
    byte[][] noDicts = dimObject.getNoDictionaryKeys();
    byte[][] newNoDicts = new byte[noDictCount][];
    for (int i = 0; i < noDictCount; i++) {
      if (isColumnDrift[i]) {
        newNoDicts[i] = DataTypeUtil
            .getBytesDataDataTypeForNoDictionaryColumn(rawRow[noDictMap[i]], measureDataTypes[i]);
      } else {
        newNoDicts[i] = noDicts[noDictMap[i]];
      }
    }
    ByteArrayWrapper newWrapper = new ByteArrayWrapper();
    newWrapper.setDictionaryKey(dimObject.getDictionaryKey());
    newWrapper.setNoDictionaryKeys(newNoDicts);
    newWrapper.setComplexTypesKeys(dimObject.getComplexTypesKeys());
    newWrapper.setImplicitColumnByteArray(dimObject.getImplicitColumnByteArray());
    Object[] finalRawRow = new Object[1 + measureCount];
    finalRawRow[0] = newWrapper;
    System.arraycopy(rawRow, 1, finalRawRow, 1, measureCount);
    return finalRawRow;
  }
}
