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.processing.loading.sort;
  
  import java.io.DataInputStream;
  import java.io.DataOutputStream;
  import java.io.IOException;
  import java.io.Serializable;
  import java.math.BigDecimal;
  import java.nio.ByteBuffer;
  import java.nio.charset.Charset;
  
  import org.apache.carbondata.core.constants.CarbonCommonConstants;
  import org.apache.carbondata.core.memory.CarbonUnsafe;
  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.CarbonUnsafeUtil;
  import org.apache.carbondata.core.util.DataTypeUtil;
  import org.apache.carbondata.core.util.NonDictionaryUtil;
  import org.apache.carbondata.core.util.ReUsableByteArrayDataOutputStream;
  import org.apache.carbondata.processing.loading.row.IntermediateSortTempRow;
  import org.apache.carbondata.processing.sort.SortTempRowUpdater;
  import org.apache.carbondata.processing.sort.sortdata.SortParameters;
  import org.apache.carbondata.processing.sort.sortdata.TableFieldStat;
  
  /**
   * This class is used to convert/write/read row in sort step in carbondata.
   * It consists the following function:
   * 1. convert raw row & intermediate sort temp row to 3-parted row
   * 2. read/write intermediate sort temp row to sort temp file & unsafe memory
   * 3. write raw row directly to sort temp file & unsafe memory as intermediate sort temp row
   */
  public class SortStepRowHandler implements Serializable {
    private static final long serialVersionUID = 1L;
    private int dictSortDimCnt = 0;
    private int dictNoSortDimCnt = 0;
    private int noDictSortDimCnt = 0;
    private int noDictNoSortDimCnt = 0;
    private int varcharDimCnt = 0;
    private int complexDimCnt = 0;
    private int measureCnt;
  
    // indices for dict & sort dimension columns
    private int[] dictSortDimIdx;
    // indices for dict & no-sort dimension columns
    private int[] dictNoSortDimIdx;
    // indices for no-dict & sort dimension columns
    private int[] noDictSortDimIdx;
    // indices for no-dict & no-sort dimension columns, excluding complex/varchar columns
    private int[] noDictNoSortDimIdx;
    private int[] varcharDimIdx;
    private int[] complexDimIdx;
    // indices for measure columns
    private int[] measureIdx;
  
    private DataType[] dataTypes;
  
    private DataType[] noDictSortDataTypes;
  
    private boolean[] noDictSortColMapping;
  
    private DataType[] noDictNoSortDataTypes;
  
    private boolean[] noDictNoSortColMapping;
  
    private SortTempRowUpdater sortTempRowUpdater;
  
    /**
     * constructor
     * @param tableFieldStat table field stat
     */
    public SortStepRowHandler(TableFieldStat tableFieldStat) {
      this.dictSortDimCnt = tableFieldStat.getDictSortDimCnt();
      this.dictNoSortDimCnt = tableFieldStat.getDictNoSortDimCnt();
      this.noDictSortDimCnt = tableFieldStat.getNoDictSortDimCnt();
      this.noDictNoSortDimCnt = tableFieldStat.getNoDictNoSortDimCnt();
      this.varcharDimCnt = tableFieldStat.getVarcharDimCnt();
      this.complexDimCnt = tableFieldStat.getComplexDimCnt();
      this.measureCnt = tableFieldStat.getMeasureCnt();
      this.dictSortDimIdx = tableFieldStat.getDictSortDimIdx();
      this.dictNoSortDimIdx = tableFieldStat.getDictNoSortDimIdx();
      this.noDictSortDimIdx = tableFieldStat.getNoDictSortDimIdx();
      this.noDictNoSortDimIdx = tableFieldStat.getNoDictNoSortDimIdx();
      this.varcharDimIdx = tableFieldStat.getVarcharDimIdx();
      this.complexDimIdx = tableFieldStat.getComplexDimIdx();
      this.measureIdx = tableFieldStat.getMeasureIdx();
      this.dataTypes = tableFieldStat.getMeasureDataType();
      this.noDictSortDataTypes = tableFieldStat.getNoDictSortDataType();
      noDictSortColMapping = new boolean[noDictSortDataTypes.length];
      for (int i = 0; i < noDictSortDataTypes.length; i++) {
        noDictSortColMapping[i] = DataTypeUtil.isPrimitiveColumn(noDictSortDataTypes[i]);
      }
      this.noDictNoSortDataTypes = tableFieldStat.getNoDictNoSortDataType();
      noDictNoSortColMapping = new boolean[noDictNoSortDataTypes.length];
      for (int i = 0; i < noDictNoSortDataTypes.length; i++) {
        noDictNoSortColMapping[i] = DataTypeUtil.isPrimitiveColumn(noDictNoSortDataTypes[i]);
      }
      this.sortTempRowUpdater = tableFieldStat.getSortTempRowUpdater();
    }
  
    /**
     * constructor
     * @param sortParameters sort parameters
     */
    public SortStepRowHandler(SortParameters sortParameters) {
      this(new TableFieldStat(sortParameters));
    }
  
    /**
     * Convert carbon row from raw format to 3-parted format.
     * This method is used in global-sort.
     *
     * @param row raw row whose length is the same as field number
     * @return 3-parted row whose length is 3. (1 for dict dims ,1 for non-dict and complex,
     * 1 for measures)
     */
    public Object[] convertRawRowTo3Parts(Object[] row) {
      Object[] holder = new Object[3];
      try {
        int[] dictDims = new int[this.dictSortDimCnt + this.dictNoSortDimCnt];
        Object[] nonDictArray = new Object[this.noDictSortDimCnt + this.noDictNoSortDimCnt
                                         + this.varcharDimCnt + this.complexDimCnt];
        Object[] measures = new Object[this.measureCnt];
  
        // convert dict & data
        int idxAcc = 0;
        for (int idx = 0; idx < this.dictSortDimCnt; idx++) {
          dictDims[idxAcc++] = (int) row[this.dictSortDimIdx[idx]];
        }
  
        // convert dict & no-sort
        for (int idx = 0; idx < this.dictNoSortDimCnt; idx++) {
          dictDims[idxAcc++] = (int) row[this.dictNoSortDimIdx[idx]];
        }
        // convert no-dict & sort
        idxAcc = 0;
        for (int idx = 0; idx < this.noDictSortDimCnt; idx++) {
          nonDictArray[idxAcc++] = row[this.noDictSortDimIdx[idx]];
        }
        // convert no-dict & no-sort
        for (int idx = 0; idx < this.noDictNoSortDimCnt; idx++) {
          nonDictArray[idxAcc++] = row[this.noDictNoSortDimIdx[idx]];
        }
        // convert varchar dims
        for (int idx = 0; idx < this.varcharDimCnt; idx++) {
          nonDictArray[idxAcc++] = row[this.varcharDimIdx[idx]];
        }
        // convert complex dims
        for (int idx = 0; idx < this.complexDimCnt; idx++) {
          nonDictArray[idxAcc++] = row[this.complexDimIdx[idx]];
        }
  
        // convert measure data
        for (int idx = 0; idx < this.measureCnt; idx++) {
          measures[idx] = row[this.measureIdx[idx]];
        }
        sortTempRowUpdater.updateOutputRow(holder, dictDims, nonDictArray, measures);
      } catch (Exception e) {
        throw new RuntimeException("Problem while converting row to 3 parts", e);
      }
      return holder;
    }
  
    /**
     * Convert intermediate sort temp row to 3-parted row.
     * This method is used in the final merge sort to feed rows to the next write step.
     *
     * @param sortTempRow intermediate sort temp row
     * @return 3-parted row
     */
    public Object[] convertIntermediateSortTempRowTo3Parted(IntermediateSortTempRow sortTempRow) {
      Object[] out = new Object[3];
      NonDictionaryUtil
          .prepareOutObj(out, sortTempRow.getDictSortDims(), sortTempRow.getNoDictSortDims(),
              sortTempRow.getMeasures());
      return out;
    }
  
    /**
     * Read intermediate sort temp row from InputStream.
     * This method is used during the intermediate merge sort phase to read row from sort temp file.
     *
     * @param inputStream input stream
     * @return a row that contains three parts
     * @throws IOException if error occrus while reading from stream
     */
    public IntermediateSortTempRow readWithoutNoSortFieldConvert(
        DataInputStream inputStream) throws IOException {
      int[] dictSortDims = new int[this.dictSortDimCnt];
      Object[] noDictSortDims = new Object[this.noDictSortDimCnt];
  
      // read dict & sort dim data
      for (int idx = 0; idx < this.dictSortDimCnt; idx++) {
        dictSortDims[idx] = inputStream.readInt();
      }
  
      // read no-dict & sort data
      for (int idx = 0; idx < this.noDictSortDimCnt; idx++) {
        // for no dict measure column get the original data
        noDictSortDims[idx] = getDataForNoDictSortColumn(inputStream, idx);
      }
  
      // read no-dict dims & measures
      int len = inputStream.readInt();
      byte[] noSortDimsAndMeasures = new byte[len];
      inputStream.readFully(noSortDimsAndMeasures);
      // keeping no sort fields and measure in pack byte array as it will not participate in sort
      return new IntermediateSortTempRow(dictSortDims, noDictSortDims, noSortDimsAndMeasures);
    }
  
    /**
     * Read intermediate sort temp row from InputStream.
     * This method is used during the final merge sort phase to read row from sort temp file and
     * merged sort temp file.
     *
     * @param inputStream input stream
     * @return a row that contains three parts
     * @throws IOException if error occrus while reading from stream
     */
    public IntermediateSortTempRow readWithNoSortFieldConvert(
        DataInputStream inputStream) throws IOException {
      int[] dictSortDims = new int[this.dictSortDimCnt + this.dictNoSortDimCnt];
      Object[] noDictSortDims =
          new Object[this.noDictSortDimCnt + this.noDictNoSortDimCnt + this.varcharDimCnt
              + this.complexDimCnt];
  
      // read dict & sort dim data
      for (int idx = 0; idx < this.dictSortDimCnt; idx++) {
        dictSortDims[idx] = inputStream.readInt();
      }
  
      // read no-dict & sort data
      for (int idx = 0; idx < this.noDictSortDimCnt; idx++) {
        // for no dict measure column get the original data
        noDictSortDims[idx] = getDataForNoDictSortColumn(inputStream, idx);
      }
  
      // read no-dict dims & measures
      int len = inputStream.readInt();
      byte[] noSortDimsAndMeasures = new byte[len];
      inputStream.readFully(noSortDimsAndMeasures);
      Object[] measure = new Object[this.measureCnt];
      // unpack the no sort fields and measure fields
      unpackNoSortFromBytes(noSortDimsAndMeasures, dictSortDims, noDictSortDims, measure);
      return new IntermediateSortTempRow(dictSortDims, noDictSortDims, measure);
    }
  
    /**
     * Return the data from the stream according to the column type
     *
     * @param inputStream
     * @param idx
     * @throws IOException
     */
    private Object getDataForNoDictSortColumn(DataInputStream inputStream, int idx)
        throws IOException {
      if (this.noDictSortColMapping[idx]) {
        return readDataFromStream(inputStream, idx);
      } else {
        short len = inputStream.readShort();
        byte[] bytes = new byte[len];
        inputStream.readFully(bytes);
        return bytes;
      }
    }
  
    /**
     * Read the data from the stream
     *
     * @param inputStream
     * @param idx
     * @return
     * @throws IOException
     */
    private Object readDataFromStream(DataInputStream inputStream, int idx) throws IOException {
      DataType dataType = noDictSortDataTypes[idx];
      Object data = null;
      if (!inputStream.readBoolean()) {
        return null;
      }
      if (dataType == DataTypes.BOOLEAN) {
        data = inputStream.readBoolean();
      } else if (dataType == DataTypes.BYTE) {
        data = inputStream.readByte();
      } else if (dataType == DataTypes.SHORT) {
        data = inputStream.readShort();
      } else if (dataType == DataTypes.INT) {
        data = inputStream.readInt();
      } else if (dataType == DataTypes.LONG || dataType == DataTypes.TIMESTAMP) {
        data = inputStream.readLong();
      } else if (dataType == DataTypes.DOUBLE) {
        data = inputStream.readDouble();
      } else if (dataType == DataTypes.FLOAT) {
        data = inputStream.readFloat();
      } else if (dataType == DataTypes.BYTE_ARRAY || DataTypes.isDecimal(dataType)) {
        byte[] bytes =
            inputStream.readUTF().getBytes(Charset.forName(CarbonCommonConstants.DEFAULT_CHARSET));
        data = bytes;
      }
      return data;
    }
  
    private void unpackNoSortFromBytes(byte[] noSortDimsAndMeasures, int[] dictDims,
        Object[] noDictDims, Object[] measures) {
      ByteBuffer rowBuffer = ByteBuffer.wrap(noSortDimsAndMeasures);
      // read dict_no_sort
      for (int i = dictSortDimCnt; i < dictDims.length; i++) {
        dictDims[i] = rowBuffer.getInt();
      }
  
      int noDictIndex = noDictSortDimCnt;
      // read no_dict_no_sort
      for (int i = 0; i < noDictNoSortDimCnt; i++) {
        // for no dict measure column get the original data
        if (this.noDictNoSortColMapping[i]) {
          noDictDims[noDictIndex++] = getDataFromRowBuffer(noDictNoSortDataTypes[i], rowBuffer);
        } else {
          short len = rowBuffer.getShort();
          byte[] bytes = new byte[len];
          rowBuffer.get(bytes);
          noDictDims[noDictIndex++] = bytes;
        }
      }
  
      // read varchar dims
      for (int i = 0; i < varcharDimCnt; i++) {
        int len = rowBuffer.getInt();
        byte[] bytes = new byte[len];
        rowBuffer.get(bytes);
        noDictDims[noDictIndex++] = bytes;
      }
  
      // read complex dims
      for (int i = 0; i < complexDimCnt; i++) {
        int len = rowBuffer.getInt();
        byte[] bytes = new byte[len];
        rowBuffer.get(bytes);
        noDictDims[noDictIndex++] = bytes;
      }
  
      // read measure
      int measureCnt = measures.length;
      Object tmpContent;
      for (short idx = 0; idx < measureCnt; idx++) {
        tmpContent = getDataFromRowBuffer(dataTypes[idx], rowBuffer);
        measures[idx] = tmpContent;
      }
    }
  
    /**
     * Retrieve/Get the data from the row buffer.
     *
     * @param tmpDataType
     * @param rowBuffer
     * @return
     */
    private Object getDataFromRowBuffer(DataType tmpDataType, ByteBuffer rowBuffer) {
      Object tmpContent;
      if ((byte) 0 == rowBuffer.get()) {
        return null;
      }
  
      if (DataTypes.BOOLEAN == tmpDataType) {
        if ((byte) 1 == rowBuffer.get()) {
          tmpContent = true;
        } else {
          tmpContent = false;
        }
      } else if (DataTypes.SHORT == tmpDataType) {
        tmpContent = rowBuffer.getShort();
      } else if (DataTypes.INT == tmpDataType) {
        tmpContent = rowBuffer.getInt();
      } else if (DataTypes.LONG == tmpDataType || DataTypes.TIMESTAMP == tmpDataType) {
        tmpContent = rowBuffer.getLong();
      } else if (DataTypes.DOUBLE == tmpDataType) {
        tmpContent = rowBuffer.getDouble();
      } else if (DataTypes.FLOAT == tmpDataType) {
        tmpContent = rowBuffer.getFloat();
      } else if (DataTypes.BYTE == tmpDataType) {
        tmpContent = rowBuffer.get();
      } else if (DataTypes.isDecimal(tmpDataType)) {
        short len = rowBuffer.getShort();
        byte[] decimalBytes = new byte[len];
        rowBuffer.get(decimalBytes);
        tmpContent = DataTypeUtil.byteToBigDecimal(decimalBytes);
      } else if (DataTypes.BINARY == tmpDataType) {
        int len = rowBuffer.getInt();
        byte[] bytes = new byte[len];
        rowBuffer.get(bytes);
        tmpContent = bytes;
      } else {
        throw new IllegalArgumentException("Unsupported data type: " + tmpDataType);
      }
      return tmpContent;
    }
  
    /**
     * Write intermediate sort temp row to OutputStream
     * This method is used during the merge sort phase to write row to sort temp file.
     *
     * @param sortTempRow intermediate sort temp row
     * @param outputStream output stream
     * @throws IOException if error occurs while writing to stream
     */
    public void writeIntermediateSortTempRowToOutputStream(IntermediateSortTempRow sortTempRow,
        DataOutputStream outputStream) throws IOException {
      // write dict & sort dim
      for (int idx = 0; idx < this.dictSortDimCnt; idx++) {
        outputStream.writeInt(sortTempRow.getDictSortDims()[idx]);
      }
  
      // write no-dict & sort dim
      for (int idx = 0; idx < this.noDictSortDimCnt; idx++) {
        if (this.noDictSortColMapping[idx]) {
          // write the original data to the stream
          writeDataToStream(sortTempRow.getNoDictSortDims()[idx], outputStream, idx);
        } else {
          byte[] bytes = (byte[]) sortTempRow.getNoDictSortDims()[idx];
          outputStream.writeShort(bytes.length);
          outputStream.write(bytes);
        }
      }
  
      // write packed no-sort dim & measure
      outputStream.writeInt(sortTempRow.getNoSortDimsAndMeasures().length);
      outputStream.write(sortTempRow.getNoSortDimsAndMeasures());
    }
  
    /**
     * Write raw row as an intermediate sort temp row to sort temp file.
     * This method is used in the beginning of the sort phase. Comparing with converting raw row to
     * intermediate sort temp row and then writing the converted one, Writing raw row directly will
     * save the intermediate trivial loss.
     * This method use an array backend buffer to save memory allocation. The buffer will be reused
     * for all rows (per thread).
     *
     * @param row raw row
     * @param outputStream output stream
     * @param reUsableByteArrayDataOutputStream DataOutputStream backend by ByteArrayOutputStream
     * @throws IOException if error occurs while writing to stream
     */
    public void writeRawRowAsIntermediateSortTempRowToOutputStream(Object[] row,
        DataOutputStream outputStream,
        ReUsableByteArrayDataOutputStream reUsableByteArrayDataOutputStream) throws IOException {
      // write dict & sort
      for (int idx = 0; idx < this.dictSortDimCnt; idx++) {
        outputStream.writeInt((int) row[this.dictSortDimIdx[idx]]);
      }
  
      // write no-dict & sort
      for (int idx = 0; idx < this.noDictSortDimCnt; idx++) {
        if (this.noDictSortColMapping[idx]) {
          // write the original data to the stream
          writeDataToStream(row[this.noDictSortDimIdx[idx]], outputStream, idx);
        } else {
          byte[] bytes = (byte[]) row[this.noDictSortDimIdx[idx]];
          outputStream.writeShort(bytes.length);
          outputStream.write(bytes);
        }
      }
  
      // pack no-sort
      reUsableByteArrayDataOutputStream.reset();
      packNoSortFieldsToBytes(row, reUsableByteArrayDataOutputStream);
      int packSize = reUsableByteArrayDataOutputStream.getSize();
  
      // write no-sort
      outputStream.writeInt(packSize);
      outputStream.write(reUsableByteArrayDataOutputStream.getByteArray(), 0, packSize);
    }
  
    /**
     * Write the data to stream
     *
     * @param data
     * @param outputStream
     * @param idx
     * @throws IOException
     */
    private void writeDataToStream(Object data, DataOutputStream outputStream, int idx)
        throws IOException {
      DataType dataType = noDictSortDataTypes[idx];
      if (null == data) {
        outputStream.writeBoolean(false);
      } else {
        outputStream.writeBoolean(true);
        if (dataType == DataTypes.BOOLEAN) {
          outputStream.writeBoolean((boolean) data);
        } else if (dataType == DataTypes.BYTE) {
          outputStream.writeByte((byte) data);
        } else if (dataType == DataTypes.SHORT) {
          outputStream.writeShort((short) data);
        } else if (dataType == DataTypes.INT) {
          outputStream.writeInt((int) data);
        } else if (dataType == DataTypes.LONG || dataType == DataTypes.TIMESTAMP) {
          outputStream.writeLong((long) data);
        } else if (dataType == DataTypes.DOUBLE) {
          outputStream.writeDouble((double) data);
        } else if (DataTypes.isDecimal(dataType)) {
          BigDecimal val = (BigDecimal) data;
          byte[] bigDecimalInBytes = DataTypeUtil.bigDecimalToByte(val);
          outputStream.writeShort(bigDecimalInBytes.length);
          outputStream.write(bigDecimalInBytes);
        } else if (dataType == DataTypes.FLOAT) {
          outputStream.writeFloat((float) data);
        } else if (dataType == DataTypes.BYTE_ARRAY) {
          outputStream.writeUTF(data.toString());
        }
      }
    }
  
    /**
     * Read intermediate sort temp row from unsafe memory.
     * This method is used during merge sort phase for off-heap sort.
     *
     * @param baseObject base object of memory block
     * @param address address of the row
     * @return intermediate sort temp row
     */
    public IntermediateSortTempRow readFromMemoryWithoutNoSortFieldConvert(Object baseObject,
        long address) {
      int size = 0;
  
      int[] dictSortDims = new int[this.dictSortDimCnt];
      byte[][] noDictSortDims = new byte[this.noDictSortDimCnt][];
  
      // read dict & sort dim
      for (int idx = 0; idx < dictSortDims.length; idx++) {
        dictSortDims[idx] = CarbonUnsafe.getUnsafe().getInt(baseObject, address + size);
        size += 4;
      }
  
      // read no-dict & sort dim
      for (int idx = 0; idx < noDictSortDims.length; idx++) {
        short length = CarbonUnsafe.getUnsafe().getShort(baseObject, address + size);
        size += 2;
        byte[] bytes = new byte[length];
        CarbonUnsafe.getUnsafe().copyMemory(baseObject, address + size,
            bytes, CarbonUnsafe.BYTE_ARRAY_OFFSET, length);
        size += length;
        noDictSortDims[idx] = bytes;
      }
  
      // read no-sort dims & measures
      int len = CarbonUnsafe.getUnsafe().getInt(baseObject, address + size);
      size += 4;
      byte[] noSortDimsAndMeasures = new byte[len];
      CarbonUnsafe.getUnsafe().copyMemory(baseObject, address + size,
          noSortDimsAndMeasures, CarbonUnsafe.BYTE_ARRAY_OFFSET, len);
  
      return new IntermediateSortTempRow(dictSortDims, noDictSortDims, noSortDimsAndMeasures);
    }
  
    /**
     * Read intermediate sort temp row from unsafe memory.
     * This method is used during merge sort phase for off-heap sort.
     *
     * @param baseObject base object of memory block
     * @param address address of the row
     * @return intermediate sort temp row
     */
    public IntermediateSortTempRow readRowFromMemoryWithNoSortFieldConvert(Object baseObject,
        long address) {
      int size = 0;
  
      int[] dictSortDims = new int[this.dictSortDimCnt + this.dictNoSortDimCnt];
      Object[] noDictSortDims =
          new Object[this.noDictSortDimCnt + this.noDictNoSortDimCnt + this.varcharDimCnt
              + this.complexDimCnt];
  
      // read dict & sort dim
      for (int idx = 0; idx < dictSortDimCnt; idx++) {
        dictSortDims[idx] = CarbonUnsafe.getUnsafe().getInt(baseObject, address + size);
        size += 4;
      }
  
      // read no-dict & sort dim
      for (int idx = 0; idx < this.noDictSortDimCnt; idx++) {
        short length = CarbonUnsafe.getUnsafe().getShort(baseObject, address + size);
        size += 2;
        if (this.noDictSortColMapping[idx]) {
          // get the original data from the unsafe memory
          if (0 == length) {
            // if the length is 0, the the data is null
            noDictSortDims[idx] = null;
          } else {
            Object data = CarbonUnsafeUtil
                .getDataFromUnsafe(noDictSortDataTypes[idx], baseObject, address, size, length);
            size += length;
            noDictSortDims[idx] = data;
          }
        } else {
          byte[] bytes = new byte[length];
          CarbonUnsafe.getUnsafe()
              .copyMemory(baseObject, address + size, bytes, CarbonUnsafe.BYTE_ARRAY_OFFSET, length);
          size += length;
          noDictSortDims[idx] = bytes;
        }
      }
  
      // read no-sort dims & measures
      int len = CarbonUnsafe.getUnsafe().getInt(baseObject, address + size);
      size += 4;
      byte[] noSortDimsAndMeasures = new byte[len];
      CarbonUnsafe.getUnsafe().copyMemory(baseObject, address + size,
          noSortDimsAndMeasures, CarbonUnsafe.BYTE_ARRAY_OFFSET, len);
      Object[] measures = new Object[measureCnt];
      unpackNoSortFromBytes(noSortDimsAndMeasures, dictSortDims, noDictSortDims, measures);
      return new IntermediateSortTempRow(dictSortDims, noDictSortDims, measures);
    }
  
    /**
     * Write intermediate sort temp row directly from unsafe memory to stream.
     * This method is used at the late beginning of the sort phase to write in-memory pages
     * to sort temp file. Comparing with reading intermediate sort temp row from memory and then
     * writing it, Writing directly from memory to stream will save the intermediate trivial loss.
     *
     * @param baseObject base object of the memory block
     * @param address base address of the row
     * @param outputStream output stream
     * @param unsafeTotalLength
     * @throws IOException if error occurs while writing to stream
     */
    public void writeIntermediateSortTempRowFromUnsafeMemoryToStream(Object baseObject, long address,
        DataOutputStream outputStream, long unsafeRemainingLength, long unsafeTotalLength)
        throws IOException, MemoryException {
      int size = 0;
  
      // dict & sort
      for (int idx = 0; idx < dictSortDimCnt; idx++) {
        outputStream.writeInt(CarbonUnsafe.getUnsafe().getInt(baseObject, address + size));
        size += 4;
      }
  
      // no-dict & sort
      for (int idx = 0; idx < noDictSortDimCnt; idx++) {
        short length = CarbonUnsafe.getUnsafe().getShort(baseObject, address + size);
        size += 2;
        if (this.noDictSortColMapping[idx]) {
          // get the original data from unsafe memory
          if (0 == length) {
            // if the length is 0, then the data is null
            writeDataToStream(null, outputStream, idx);
          } else {
            Object data = CarbonUnsafeUtil
                .getDataFromUnsafe(noDictSortDataTypes[idx], baseObject, address, size, length);
            size += length;
            writeDataToStream(data, outputStream, idx);
          }
        } else {
          validateUnsafeMemoryBlockSizeLimit(unsafeRemainingLength, length, unsafeTotalLength);
          byte[] bytes = new byte[length];
          CarbonUnsafe.getUnsafe()
              .copyMemory(baseObject, address + size, bytes, CarbonUnsafe.BYTE_ARRAY_OFFSET, length);
          size += length;
          outputStream.writeShort(length);
          outputStream.write(bytes);
        }
      }
  
      // packed no-sort & measure
      int len = CarbonUnsafe.getUnsafe().getInt(baseObject, address + size);
      size += 4;
      validateUnsafeMemoryBlockSizeLimit(unsafeRemainingLength, len, unsafeTotalLength);
      byte[] noSortDimsAndMeasures = new byte[len];
      CarbonUnsafe.getUnsafe().copyMemory(baseObject, address + size,
          noSortDimsAndMeasures, CarbonUnsafe.BYTE_ARRAY_OFFSET, len);
  
      outputStream.writeInt(len);
      outputStream.write(noSortDimsAndMeasures);
    }
  
    /**
     * Write raw row as an intermediate sort temp row to memory.
     * This method is used in the beginning of the off-heap sort phase. Comparing with converting
     * raw row to intermediate sort temp row and then writing the converted one,
     * Writing raw row directly will save the intermediate trivial loss.
     * This method use an array backend buffer to save memory allocation. The buffer will be reused
     * for all rows (per thread).
     *
     * @param row raw row
     * @param baseObject base object of the memory block
     * @param address base address for the row
     * @param unsafeTotalLength
     * @return number of bytes written to memory
     */
    public int writeRawRowAsIntermediateSortTempRowToUnsafeMemory(Object[] row, Object baseObject,
        long address, ReUsableByteArrayDataOutputStream reUsableByteArrayDataOutputStream,
        long unsafeRemainingLength, long unsafeTotalLength) throws MemoryException, IOException {
      int size = 0;
      // write dict & sort
      for (int idx = 0; idx < this.dictSortDimCnt; idx++) {
        validateUnsafeMemoryBlockSizeLimit(unsafeRemainingLength, 4, unsafeTotalLength);
        CarbonUnsafe.getUnsafe()
            .putInt(baseObject, address + size, (int) row[this.dictSortDimIdx[idx]]);
        size += 4;
      }
  
      // write no-dict & sort
      for (int idx = 0; idx < this.noDictSortDimCnt; idx++) {
        if (this.noDictSortColMapping[idx]) {
          Object data = row[this.noDictSortDimIdx[idx]];
          if (null == data) {
            // if the data is null, then write only the length as 0.
            CarbonUnsafe.getUnsafe().putShort(baseObject, address + size, (short) 0);
            size += 2;
          } else {
            int sizeInBytes = this.noDictSortDataTypes[idx].getSizeInBytes();
            if (this.noDictSortDataTypes[idx] == DataTypes.TIMESTAMP) {
              sizeInBytes = DataTypes.LONG.getSizeInBytes();
            }
            CarbonUnsafe.getUnsafe().putShort(baseObject, address + size, (short) sizeInBytes);
            size += 2;
            // put data to unsafe according to the data types
            CarbonUnsafeUtil
                .putDataToUnsafe(noDictSortDataTypes[idx], data, baseObject, address, size,
                    sizeInBytes);
            size += sizeInBytes;
          }
        } else {
          byte[] bytes = (byte[]) row[this.noDictSortDimIdx[idx]];
          validateUnsafeMemoryBlockSizeLimit(unsafeRemainingLength, 2 + bytes.length,
              unsafeTotalLength);
          CarbonUnsafe.getUnsafe().putShort(baseObject, address + size, (short) bytes.length);
          size += 2;
          CarbonUnsafe.getUnsafe()
              .copyMemory(bytes, CarbonUnsafe.BYTE_ARRAY_OFFSET, baseObject, address + size,
                  bytes.length);
          size += bytes.length;
        }
      }
  
      // convert pack no-sort
      reUsableByteArrayDataOutputStream.reset();
      packNoSortFieldsToBytes(row, reUsableByteArrayDataOutputStream);
      int packSize = reUsableByteArrayDataOutputStream.getSize();
  
      validateUnsafeMemoryBlockSizeLimit(unsafeRemainingLength, 4 + packSize, unsafeTotalLength);
      // write no-sort
      CarbonUnsafe.getUnsafe().putInt(baseObject, address + size, packSize);
      size += 4;
      CarbonUnsafe.getUnsafe().copyMemory(reUsableByteArrayDataOutputStream.getByteArray(),
          CarbonUnsafe.BYTE_ARRAY_OFFSET, baseObject, address + size, packSize);
      size += packSize;
      return size;
    }
  
    private void validateUnsafeMemoryBlockSizeLimit(long unsafeRemainingLength, int requestedSize,
        long unsafeTotalLength) throws MemoryException {
      if (unsafeTotalLength <= requestedSize) {
        throw new MemoryException(
            "not enough unsafe memory for sort: increase the 'offheap.sort.chunk.size.inmb' ");
      } else if (unsafeRemainingLength <= requestedSize) {
        throw new MemoryException("cannot handle this row. create new page");
      }
    }
  
    /**
     * Pack to no-sort fields to byte array
     *
     * @param row raw row
     * @param @param reUsableByteArrayDataOutputStream
     *        DataOutputStream backend by ByteArrayOutputStream
     */
    private void packNoSortFieldsToBytes(Object[] row,
        ReUsableByteArrayDataOutputStream reUsableByteArrayDataOutputStream) throws IOException {
      // convert dict & no-sort
      for (int idx = 0; idx < this.dictNoSortDimCnt; idx++) {
        reUsableByteArrayDataOutputStream.writeInt((int) row[this.dictNoSortDimIdx[idx]]);
      }
      // convert no-dict & no-sort
      for (int idx = 0; idx < this.noDictNoSortDimCnt; idx++) {
        if (this.noDictNoSortColMapping[idx]) {
          // put the original data to buffer
          putDataToRowBuffer(this.noDictNoSortDataTypes[idx], row[this.noDictNoSortDimIdx[idx]],
              reUsableByteArrayDataOutputStream);
        } else {
          byte[] bytes = (byte[]) row[this.noDictNoSortDimIdx[idx]];
          reUsableByteArrayDataOutputStream.writeShort((short) bytes.length);
          reUsableByteArrayDataOutputStream.write(bytes);
        }
      }
      // convert varchar dims
      for (int idx = 0; idx < this.varcharDimCnt; idx++) {
        byte[] bytes = (byte[]) row[this.varcharDimIdx[idx]];
        reUsableByteArrayDataOutputStream.writeInt(bytes.length);
        reUsableByteArrayDataOutputStream.write(bytes);
      }
      // convert complex dims
      for (int idx = 0; idx < this.complexDimCnt; idx++) {
        byte[] bytes = (byte[]) row[this.complexDimIdx[idx]];
        reUsableByteArrayDataOutputStream.writeInt(bytes.length);
        reUsableByteArrayDataOutputStream.write(bytes);
      }
  
      // convert measure
      for (int idx = 0; idx < this.measureCnt; idx++) {
        putDataToRowBuffer(this.dataTypes[idx], row[this.measureIdx[idx]],
            reUsableByteArrayDataOutputStream);
      }
    }
  
    /**
     * Put the data to the row buffer
     *  @param tmpDataType
     * @param tmpValue
     * @param reUsableByteArrayDataOutputStream
     */
    private void putDataToRowBuffer(DataType tmpDataType, Object tmpValue,
        ReUsableByteArrayDataOutputStream reUsableByteArrayDataOutputStream) throws IOException {
      if (null == tmpValue) {
        reUsableByteArrayDataOutputStream.write((byte) 0);
        return;
      }
      reUsableByteArrayDataOutputStream.write((byte) 1);
      if (DataTypes.BOOLEAN == tmpDataType) {
        if ((boolean) tmpValue) {
          reUsableByteArrayDataOutputStream.write((byte) 1);
        } else {
          reUsableByteArrayDataOutputStream.write((byte) 0);
        }
      } else if (DataTypes.SHORT == tmpDataType) {
        reUsableByteArrayDataOutputStream.writeShort((Short) tmpValue);
      } else if (DataTypes.INT == tmpDataType) {
        reUsableByteArrayDataOutputStream.writeInt((Integer) tmpValue);
      } else if (DataTypes.LONG == tmpDataType || DataTypes.TIMESTAMP == tmpDataType) {
        reUsableByteArrayDataOutputStream.writeLong((Long) tmpValue);
      } else if (DataTypes.DOUBLE == tmpDataType) {
        reUsableByteArrayDataOutputStream.writeDouble((Double) tmpValue);
      }  else if (DataTypes.FLOAT == tmpDataType) {
        reUsableByteArrayDataOutputStream.writeFloat((Float) tmpValue);
      } else if (DataTypes.BYTE == tmpDataType) {
        reUsableByteArrayDataOutputStream.write((byte) tmpValue);
      } else if (DataTypes.isDecimal(tmpDataType)) {
        byte[] decimalBytes = DataTypeUtil.bigDecimalToByte((BigDecimal) tmpValue);
        reUsableByteArrayDataOutputStream.writeShort((short) decimalBytes.length);
        reUsableByteArrayDataOutputStream.write(decimalBytes);
      } else if (DataTypes.BINARY == tmpDataType) {
        byte[] bytes = (byte[]) tmpValue;
        reUsableByteArrayDataOutputStream.writeInt(bytes.length);
        reUsableByteArrayDataOutputStream.write(bytes);
      } else {
        throw new IllegalArgumentException("Unsupported data type: " + tmpDataType);
      }
    }
  }
  