@Test public void testAdd() throws Exception { try (OffHeapMutableBytesStore offHeapMutableBytesStore = new OffHeapMutableBytesStore(_memoryManager, null)) { for (int i = 0; i < NUM_VALUES; i++) { assertEquals(offHeapMutableBytesStore.add(_values[i]), i); } } }
public int add(byte[] value) { int offsetBufferIndex = _numValues >>> OFFSET_BUFFER_SHIFT_OFFSET; int offsetIndex = _numValues & OFFSET_BUFFER_MASK; PinotDataBuffer offsetBuffer; 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 ((_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++; }
OffHeapMutableBytesStore implements Closeable { public int add(byte[] value) { int offsetBufferIndex = _numValues >>> OFFSET_BUFFER_SHIFT_OFFSET; int offsetIndex = _numValues & OFFSET_BUFFER_MASK; PinotDataBuffer offsetBuffer; 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 ((_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++; } }
OffHeapMutableBytesStore implements Closeable { public int add(byte[] value) { int offsetBufferIndex = _numValues >>> OFFSET_BUFFER_SHIFT_OFFSET; int offsetIndex = _numValues & OFFSET_BUFFER_MASK; PinotDataBuffer offsetBuffer; 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 ((_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++; } OffHeapMutableBytesStore(PinotDataBufferMemoryManager memoryManager, String allocationContext); }
OffHeapMutableBytesStore implements Closeable { public int add(byte[] value) { int offsetBufferIndex = _numValues >>> OFFSET_BUFFER_SHIFT_OFFSET; int offsetIndex = _numValues & OFFSET_BUFFER_MASK; PinotDataBuffer offsetBuffer; 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 ((_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++; } OffHeapMutableBytesStore(PinotDataBufferMemoryManager memoryManager, String allocationContext); int add(byte[] value); @SuppressWarnings("Duplicates") byte[] get(int index); @SuppressWarnings("Duplicates") boolean equalsValueAt(int index, byte[] valueToCompare); int getNumValues(); int getTotalBufferSize(); @Override void close(); }
OffHeapMutableBytesStore implements Closeable { public int add(byte[] value) { int offsetBufferIndex = _numValues >>> OFFSET_BUFFER_SHIFT_OFFSET; int offsetIndex = _numValues & OFFSET_BUFFER_MASK; PinotDataBuffer offsetBuffer; 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 ((_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++; } OffHeapMutableBytesStore(PinotDataBufferMemoryManager memoryManager, String allocationContext); int add(byte[] value); @SuppressWarnings("Duplicates") byte[] get(int index); @SuppressWarnings("Duplicates") boolean equalsValueAt(int index, byte[] valueToCompare); int getNumValues(); int getTotalBufferSize(); @Override void close();  }