/*
 * 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.hadoop.hbase.regionserver;

import java.nio.ByteBuffer;
import java.util.Comparator;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.util.ByteBufferUtils;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ClassSize;
import org.apache.yetus.audience.InterfaceAudience;

/**
 * CellChunkMap is an array of serialized representations of Cell (pointing to Chunks with full Cell
 * data) and can be allocated both off-heap and on-heap.
 * <p>
 * CellChunkMap is a byte array (chunk) holding all that is needed to access a Cell, which is
 * actually saved on another deeper chunk. Per Cell we have a reference to this deeper byte array B
 * (chunk ID, integer), offset in bytes in B (integer), length in bytes in B (integer) and seqID of
 * the cell (long). In order to save reference to byte array we use the Chunk's ID given by
 * ChunkCreator.
 * <p>
 * The CellChunkMap memory layout on chunk A relevant to a deeper byte array B, holding the actual
 * cell data:
 *
 * <pre>
 *
 * < header > <---------------     first Cell     -----------------> <-- second Cell ...
 * --------------------------------------------------------------------------------------- ...
 *  integer  | integer      | integer      | integer     | long     |
 *  4 bytes  | 4 bytes      | 4 bytes      | 4 bytes     | 8 bytes  |
 *  ChunkID  | chunkID of   | offset in B  | length of   | sequence |          ...
 *  of this  | chunk B with | where Cell's | Cell's      | ID of    |
 *  chunk A  | Cell data    | data starts  | data in B   | the Cell |
 * --------------------------------------------------------------------------------------- ...
 * </pre>
 */
@InterfaceAudience.Private
public class CellChunkMap<T extends Cell> extends CellFlatMap<T> {

  private final Chunk[] chunks; // the array of chunks, on which the index is based

  // number of cell-representations in a chunk
  // depends on the size of the chunks (may be index chunks or regular data chunks)
  // each chunk starts with its own ID following the cells data
  private final int numOfCellRepsInChunk;

  /**
   * C-tor for creating CellChunkMap from existing Chunk array, which must be ordered (decreasingly
   * or increasingly according to parameter "descending")
   * @param comparator a tool for comparing cells
   * @param chunks     ordered array of index chunk with cell representations
   * @param min        the index of the first cell (usually 0)
   * @param max        number of Cells or the index of the cell after the maximal cell
   * @param descending the order of the given array
   */
  public CellChunkMap(Comparator<? super T> comparator, Chunk[] chunks, int min, int max,
    boolean descending) {
    super(comparator, min, max, descending);
    this.chunks = chunks;
    if (chunks != null && chunks.length != 0 && chunks[0] != null) {
      this.numOfCellRepsInChunk =
        (chunks[0].size - ChunkCreator.SIZEOF_CHUNK_HEADER) / ClassSize.CELL_CHUNK_MAP_ENTRY;
    } else { // In case the chunks array was not allocated
      this.numOfCellRepsInChunk = 0;
    }
  }

  /**
   * To be used by base (CellFlatMap) class only to create a sub-CellFlatMap Should be used only to
   * create only CellChunkMap from CellChunkMap
   */
  @Override
  protected CellFlatMap<T> createSubCellFlatMap(int min, int max, boolean descending) {
    return new CellChunkMap<>(this.comparator(), this.chunks, min, max, descending);
  }

  @Override
  protected T getCell(int i) {
    // get the index of the relevant chunk inside chunk array
    int chunkIndex = (i / numOfCellRepsInChunk);
    ByteBuffer block = chunks[chunkIndex].getData();// get the ByteBuffer of the relevant chunk
    int j = i - chunkIndex * numOfCellRepsInChunk; // get the index of the cell-representation

    // find inside the offset inside the chunk holding the index, skip bytes for chunk id
    int offsetInBytes = ChunkCreator.SIZEOF_CHUNK_HEADER + j * ClassSize.CELL_CHUNK_MAP_ENTRY;

    // find the chunk holding the data of the cell, the chunkID is stored first
    int chunkId = ByteBufferUtils.toInt(block, offsetInBytes);
    Chunk chunk = ChunkCreator.getInstance().getChunk(chunkId);
    if (chunk == null) {
      // this should not happen
      throw new IllegalArgumentException("In CellChunkMap, cell must be associated with chunk."
        + ". We were looking for a cell at index " + i);
    }

    // find the offset of the data of the cell, skip integer for chunkID, offset is stored second
    int offsetOfCell = ByteBufferUtils.toInt(block, offsetInBytes + Bytes.SIZEOF_INT);
    // find the length of the data of the cell, skip two integers for chunkID and offset,
    // length is stored third
    int lengthOfCell = ByteBufferUtils.toInt(block, offsetInBytes + 2 * Bytes.SIZEOF_INT);
    // find the seqID of the cell, skip three integers for chunkID, offset, and length
    // the seqID is plain written as part of the cell representation
    long cellSeqID = ByteBufferUtils.toLong(block, offsetInBytes + 3 * Bytes.SIZEOF_INT);

    ByteBuffer buf = chunk.getData(); // get the ByteBuffer where the cell data is stored
    if (buf == null) {
      // this should not happen
      throw new IllegalArgumentException(
        "In CellChunkMap, chunk must be associated with ByteBuffer." + " Chunk: " + chunk
          + " Chunk ID: " + chunk.getId() + ", is from pool: " + chunk.isFromPool()
          + ". We were looking for a cell at index " + i);
    }

    @SuppressWarnings("unchecked")
    T cell = (T) new ByteBufferChunkKeyValue(buf, offsetOfCell, lengthOfCell, cellSeqID);

    return cell;
  }
}