content: /*
   * 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.vector.impl;
  
  import java.math.BigDecimal;
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.BitSet;
  import java.util.List;
  
  import org.apache.carbondata.core.metadata.datatype.DataType;
  import org.apache.carbondata.core.metadata.datatype.DataTypes;
  import org.apache.carbondata.core.metadata.datatype.DecimalType;
  import org.apache.carbondata.core.scan.result.vector.CarbonColumnVector;
  import org.apache.carbondata.core.scan.result.vector.CarbonDictionary;
  import org.apache.carbondata.core.scan.scanner.LazyPageLoader;
  
  public class CarbonColumnVectorImpl implements CarbonColumnVector {
  
    private Object[] data;
  
    private int[] ints;
  
    private long[] longs;
  
    private BigDecimal[] decimals;
  
    private byte[] byteArr;
  
    private byte[][] bytes;
  
    private float[] floats;
  
    private double[] doubles;
  
    private short[] shorts;
  
    protected BitSet nullBytes;
  
    private DataType dataType;
  
    private DataType blockDataType;
  
    private int[] lengths;
  
    private int[] offsets;
  
    private int batchSize;
  
    /**
     * True if there is at least one NULL byte set. This is an optimization for the writer, to skip
     * having to clear NULL bits.
     */
    protected boolean anyNullsSet;
  
    private CarbonDictionary carbonDictionary;
  
    private CarbonColumnVector dictionaryVector;
  
    private List<CarbonColumnVector> childrenVector;
  
    private LazyPageLoader lazyPage;
  
    private boolean loaded;
  
    private List<Integer> numberOfChildElementsInEachRow;
  
    private CarbonDictionary localDictionary;
  
    public CarbonColumnVectorImpl(int batchSize, DataType dataType) {
      this.batchSize = batchSize;
      nullBytes = new BitSet(batchSize);
      this.dataType = dataType;
      if (dataType == DataTypes.BOOLEAN || dataType == DataTypes.BYTE) {
        byteArr = new byte[batchSize];
      } else if (dataType == DataTypes.SHORT) {
        shorts = new short[batchSize];
      } else if (dataType == DataTypes.INT || dataType == DataTypes.DATE) {
        ints = new int[batchSize];
      } else if (dataType == DataTypes.LONG || dataType == DataTypes.TIMESTAMP) {
        longs = new long[batchSize];
      } else if (dataType == DataTypes.FLOAT) {
        floats = new float[batchSize];
      } else if (dataType == DataTypes.DOUBLE) {
        doubles = new double[batchSize];
      } else if (dataType instanceof DecimalType) {
        decimals = new BigDecimal[batchSize];
      } else if (dataType == DataTypes.STRING || dataType == DataTypes.BYTE_ARRAY
          || dataType == DataTypes.VARCHAR || dataType == DataTypes.BINARY) {
        dictionaryVector = new CarbonColumnVectorImpl(batchSize, DataTypes.INT);
        bytes = new byte[batchSize][];
      } else {
        data = new Object[batchSize];
      }
  
    }
  
    @Override
    public List<CarbonColumnVector> getChildrenVector() {
      return childrenVector;
    }
  
    public void setChildrenVector(List<CarbonColumnVector> childrenVector) {
      this.childrenVector = childrenVector;
    }
  
    public List<Integer> getNumberOfChildrenElementsInEachRow() {
      return numberOfChildElementsInEachRow;
    }
  
    public void setNumberOfChildElementsInEachRow(List<Integer> numberOfChildElementsInEachRow) {
      this.numberOfChildElementsInEachRow = numberOfChildElementsInEachRow;
    }
  
    public void setNumberOfElementsInEachRowForArray(byte[] parentPageData, int pageSize) {
      // for complex array type, go through parent page to get the child information
      ByteBuffer childInfoBuffer = ByteBuffer.wrap(parentPageData);
      List<Integer> numberOfArrayElementsInEachRow = new ArrayList<>();
      // Parent page array data looks like
      // number of children in each row [4 byte], Offset [4 byte],
      // number of children in each row [4 byte], Offset [4 byte]...
      while (pageSize != numberOfArrayElementsInEachRow.size()) {
        // get the number of array elements in current row
        numberOfArrayElementsInEachRow.add(childInfoBuffer.getInt());
        // skip offset
        childInfoBuffer.getInt();
      }
      setNumberOfChildElementsInEachRow(numberOfArrayElementsInEachRow);
    }
  
    public void setNumberOfElementsInEachRowForStruct(byte[] parentPageData, int pageSize) {
      // for complex struct type, go through parent page to get the child information
      ByteBuffer childInfoBuffer = ByteBuffer.wrap(parentPageData);
      List<Integer> numberOfStructElementsInEachRow = new ArrayList<>();
      // Parent page struct data looks like
      // number of children in each row [2 byte], number of children in each row [2 byte],
      // number of children in each row [2 byte], number of children in each row [2 byte]...
      while (pageSize != numberOfStructElementsInEachRow.size()) {
        int elements = childInfoBuffer.getShort();
        numberOfStructElementsInEachRow.add(elements);
      }
      setNumberOfChildElementsInEachRow(numberOfStructElementsInEachRow);
    }
  
    public CarbonDictionary getLocalDictionary() {
      return localDictionary;
    }
  
    public void setLocalDictionary(CarbonDictionary localDictionary) {
      this.localDictionary = localDictionary;
    }
  
    @Override
    public void putBoolean(int rowId, boolean value) {
      byteArr[rowId] =  (byte)((value) ? 1 : 0);
    }
  
    @Override
    public void putFloat(int rowId, float value) {
      floats[rowId] = value;
    }
  
    @Override
    public void putShort(int rowId, short value) {
      shorts[rowId] = value;
    }
  
    @Override
    public void putShorts(int rowId, int count, short value) {
      for (int i = 0; i < count; ++i) {
        shorts[i + rowId] = value;
      }
    }
  
    @Override
    public void putInt(int rowId, int value) {
      ints[rowId] = value;
    }
  
    @Override
    public void putInts(int rowId, int count, int value) {
      for (int i = 0; i < count; ++i) {
        ints[i + rowId] = value;
      }
    }
  
    @Override
    public void putLong(int rowId, long value) {
      longs[rowId] = value;
    }
  
    @Override
    public void putLongs(int rowId, int count, long value) {
      for (int i = 0; i < count; ++i) {
        longs[i + rowId] = value;
      }
    }
  
    @Override
    public void putDecimal(int rowId, BigDecimal  value, int precision) {
      decimals[rowId] = value;
    }
  
    @Override
    public void putDecimals(int rowId, int count, BigDecimal value, int precision) {
      for (int i = 0; i < count; ++i) {
        decimals[i + rowId] = value;
      }
    }
  
    @Override
    public void putDouble(int rowId, double value) {
      doubles[rowId] = value;
    }
  
    @Override
    public void putDoubles(int rowId, int count, double value) {
      for (int i = 0; i < count; ++i) {
        doubles[i + rowId] = value;
      }
    }
  
    @Override
    public void putByteArray(int rowId, byte[] value) {
      bytes[rowId] = value;
    }
  
    @Override
    public void putByte(int rowId, byte value) {
      byteArr[rowId] = value;
    }
  
    @Override
    public void putByteArray(int rowId, int count, byte[] value) {
      for (int i = 0; i < count; ++i) {
        bytes[i + rowId] = value;
      }
    }
  
    @Override
    public void putByteArray(int rowId, int offset, int length, byte[] value) {
      bytes[rowId] = new byte[length];
      System.arraycopy(value, offset, bytes[rowId], 0, length);
    }
  
    @Override
    public void putNull(int rowId) {
      nullBytes.set(rowId);
      anyNullsSet = true;
    }
  
    @Override
    public void putNulls(int rowId, int count) {
      for (int i = 0; i < count; ++i) {
        nullBytes.set(rowId + i);
      }
      anyNullsSet = true;
    }
  
    @Override
    public void putNotNull(int rowId) {
  
    }
  
    @Override
    public void putNotNull(int rowId, int count) {
  
    }
  
    public boolean isNullAt(int rowId) {
      return nullBytes.get(rowId);
    }
  
    @Override
    public boolean isNull(int rowId) {
      return nullBytes.get(rowId);
    }
  
    @Override
    public void putObject(int rowId, Object obj) {
      data[rowId] = obj;
    }
  
    @Override
    public Object getData(int rowId) {
      if (!loaded) {
        loadPage();
      }
      if (nullBytes.get(rowId)) {
        return null;
      }
      if (dataType == DataTypes.BOOLEAN || dataType == DataTypes.BYTE) {
        return  byteArr[rowId];
      } else if (dataType == DataTypes.SHORT) {
        return shorts[rowId];
      } else if (dataType == DataTypes.INT || dataType == DataTypes.DATE) {
        return ints[rowId];
      } else if (dataType == DataTypes.LONG || dataType == DataTypes.TIMESTAMP) {
        return longs[rowId];
      } else if (dataType == DataTypes.FLOAT) {
        return floats[rowId];
      } else if (dataType == DataTypes.DOUBLE) {
        return doubles[rowId];
      } else if (dataType instanceof DecimalType) {
        return decimals[rowId];
      } else if (dataType == DataTypes.STRING || dataType == DataTypes.BYTE_ARRAY
          || dataType == DataTypes.VARCHAR || dataType == DataTypes.BINARY) {
        if (null != carbonDictionary) {
          int dictKey = (Integer) dictionaryVector.getData(rowId);
          return carbonDictionary.getDictionaryValue(dictKey);
        } else if (byteArr != null) {
          byte[] bytes = new byte[lengths[rowId]];
          System.arraycopy(byteArr, offsets[rowId], bytes, 0, bytes.length);
          return bytes;
        } else {
          return bytes[rowId];
        }
      } else {
        return data[rowId];
      }
    }
  
    public Object getDataArray() {
      if (!loaded) {
        loadPage();
      }
      if (dataType == DataTypes.BOOLEAN || dataType == DataTypes.BYTE) {
        return  byteArr;
      } else if (dataType == DataTypes.SHORT) {
        return shorts;
      } else if (dataType == DataTypes.INT) {
        return ints;
      } else if (dataType == DataTypes.LONG || dataType == DataTypes.TIMESTAMP) {
        return longs;
      } else if (dataType == DataTypes.FLOAT) {
        return floats;
      } else if (dataType == DataTypes.DOUBLE) {
        return doubles;
      } else if (dataType instanceof DecimalType) {
        return decimals;
      } else if (dataType == DataTypes.STRING || dataType == DataTypes.BYTE_ARRAY || dataType ==
          DataTypes.VARCHAR) {
        if (null != carbonDictionary) {
          return ints;
        }
        return bytes;
      } else {
        return data;
      }
    }
  
    @Override
    public void reset() {
      nullBytes.clear();
      if (dataType == DataTypes.BOOLEAN || dataType == DataTypes.BYTE) {
        Arrays.fill(byteArr, (byte) 0);
      } else if (dataType == DataTypes.SHORT) {
        Arrays.fill(shorts, (short) 0);
      } else if (dataType == DataTypes.INT || dataType == DataTypes.DATE) {
        Arrays.fill(ints, 0);
      } else if (dataType == DataTypes.LONG || dataType == DataTypes.TIMESTAMP) {
        Arrays.fill(longs, 0);
      } else if (dataType == DataTypes.FLOAT) {
        Arrays.fill(floats, 0);
      } else if (dataType == DataTypes.DOUBLE) {
        Arrays.fill(doubles, 0);
      } else if (dataType instanceof DecimalType) {
        Arrays.fill(decimals, null);
      } else if (dataType == DataTypes.STRING || dataType == DataTypes.BYTE_ARRAY
          || dataType == DataTypes.VARCHAR || dataType == DataTypes.BINARY) {
        Arrays.fill(bytes, null);
        this.dictionaryVector.reset();
      } else {
        Arrays.fill(data, null);
      }
      loaded = false;
  
    }
  
    @Override
    public DataType getType() {
      return dataType;
    }
  
    @Override
    public DataType getBlockDataType() {
      return blockDataType;
    }
  
    @Override
    public void setBlockDataType(DataType blockDataType) {
      this.blockDataType = blockDataType;
    }
  
    @Override
    public void setFilteredRowsExist(boolean filteredRowsExist) {
  
    }
  
    @Override
    public void setDictionary(CarbonDictionary dictionary) {
      this.carbonDictionary = dictionary;
    }
  
    @Override
    public boolean hasDictionary() {
      return null != this.carbonDictionary;
    }
  
    @Override
    public CarbonColumnVector getDictionaryVector() {
      return dictionaryVector;
    }
  
    /**
     * Returns true if any of the nulls indicator are set for this column. This can be used
     * as an optimization to prevent setting nulls.
     */
    public final boolean anyNullsSet() {
      return anyNullsSet;
    }
  
    @Override
    public void putFloats(int rowId, int count, float[] src, int srcIndex) {
      for (int i = srcIndex; i < count; i++) {
        floats[rowId++] = src[i];
      }
    }
  
    @Override
    public void putShorts(int rowId, int count, short[] src, int srcIndex) {
      for (int i = srcIndex; i < count; i++) {
        shorts[rowId++] = src[i];
      }
    }
  
    @Override
    public void putInts(int rowId, int count, int[] src, int srcIndex) {
      for (int i = srcIndex; i < count; i++) {
        ints[rowId++] = src[i];
      }
    }
  
    @Override
    public void putLongs(int rowId, int count, long[] src, int srcIndex) {
      for (int i = srcIndex; i < count; i++) {
        longs[rowId++] = src[i];
      }
    }
  
    @Override
    public void putDoubles(int rowId, int count, double[] src, int srcIndex) {
      for (int i = srcIndex; i < count; i++) {
        doubles[rowId++] = src[i];
      }
    }
  
    @Override
    public void putBytes(int rowId, int count, byte[] src, int srcIndex) {
      for (int i = srcIndex; i < count; i++) {
        byteArr[rowId++] = src[i];
      }
    }
  
    @Override
    public void setLazyPage(LazyPageLoader lazyPage) {
      this.lazyPage = lazyPage;
    }
  
    public void loadPage() {
      if (lazyPage != null) {
        lazyPage.loadPage();
      }
      loaded = true;
    }
  
    @Override
    public void putArray(int rowId, int offset, int length) {
      if (offsets == null) {
        offsets = new int[batchSize];
        lengths = new int[batchSize];
      }
      offsets[rowId] = offset;
      lengths[rowId] = length;
    }
  
    @Override
    public void setCarbonDataFileWrittenVersion(String carbonDataFileWrittenVersion) {
     // do nothing here
    }
  
    @Override
    public void putAllByteArray(byte[] data, int offset, int length) {
      byteArr = data;
    }
  
    public int[] getLengths() {
      return lengths;
    }
  
    public int getIntArraySize() {
      return ints.length;
    }
  
    public void increaseIntArraySize(int size) {
      this.ints = new int[size];
    }
  
    public int[] getOffsets() {
      return offsets;
    }
  }
  