public int add(byte[] value) {
    int offsetBufferIndex = _numValues >>> OFFSET_BUFFER_SHIFT_OFFSET;
    int offsetIndex = _numValues & OFFSET_BUFFER_MASK;

    PinotDataBuffer offsetBuffer;
    // If this is the first entry in the offset buffer, allocate a new buffer and store the end offset of the previous
    // value
    if (offsetIndex == 0) {
      offsetBuffer = _memoryManager.allocate(OFFSET_BUFFER_SIZE, _allocationContext);
      offsetBuffer.putInt(0, _previousValueEndOffset);
      _offsetBuffers.set(offsetBufferIndex, offsetBuffer);
      _totalBufferSize += OFFSET_BUFFER_SIZE;
    } else {
      offsetBuffer = _offsetBuffers.get(offsetBufferIndex);
    }

    int valueLength = value.length;
    if (valueLength == 0) {
      offsetBuffer.putInt((offsetIndex + 1) << 2, _previousValueEndOffset);
      return _numValues++;
    }

    int valueBufferIndex = (_previousValueEndOffset + valueLength - 1) >>> VALUE_BUFFER_SHIFT_OFFSET;
    PinotDataBuffer valueBuffer;
    int valueStartOffset;
    // If the current value buffer does not have enough space, allocate a new buffer to store the value
    if ((_previousValueEndOffset - 1) >>> VALUE_BUFFER_SHIFT_OFFSET != valueBufferIndex) {
      valueBuffer = _memoryManager.allocate(VALUE_BUFFER_SIZE, _allocationContext);
      _valueBuffers.set(valueBufferIndex, valueBuffer);
      _totalBufferSize += VALUE_BUFFER_SIZE;
      valueStartOffset = valueBufferIndex << VALUE_BUFFER_SHIFT_OFFSET;
    } else {
      valueBuffer = _valueBuffers.get(valueBufferIndex);
      valueStartOffset = _previousValueEndOffset;
    }

    int valueEndOffset = valueStartOffset + valueLength;
    offsetBuffer.putInt((offsetIndex + 1) << 2, valueEndOffset);
    valueBuffer.readFrom(valueStartOffset & VALUE_BUFFER_MASK, value);
    _previousValueEndOffset = valueEndOffset;

    return _numValues++;
  }
@Test
  public void testAdd()
      throws Exception {
    try (OffHeapMutableBytesStore offHeapMutableBytesStore = new OffHeapMutableBytesStore(_memoryManager, null)) {
      for (int i = 0; i < NUM_VALUES; i++) {
        Assert.assertEquals(offHeapMutableBytesStore.add(_values[i]), i);
      }
    }
  }