ORC-1280:[C++] Implement block-based buffer(Part I)

c++/src/BlockBuffer.cc

@@ -0,0 +1,88 @@
+/**
+ * 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.
+ */
+
+#include "BlockBuffer.hh"
+
+#include <algorithm>
+
+namespace orc {
+
+ BlockBuffer::BlockBuffer(MemoryPool& pool, uint64_t _blockSize)
+ : memoryPool(pool),
+ currentSize(0),
+ currentCapacity(0),
+ blockSize(_blockSize) {
+ if (blockSize == 0) {
+ throw std::logic_error("Block size cannot be zero");
+ }
+ reserve(blockSize);
+ }
+
+ BlockBuffer::~BlockBuffer() {
+ for (size_t i = 0; i < blocks.size(); ++i) {
+ memoryPool.free(blocks[i]);
+ }
+ blocks.clear();
+ currentSize = currentCapacity = 0;
+ }
+
+ uint64_t BlockBuffer::getBlockNumber() const {
+ return (currentSize + blockSize - 1) / blockSize;
+ }
+
+ Block BlockBuffer::getBlock(uint64_t blockIndex) {
+ if (blockIndex >= getBlockNumber()) {
+ throw std::out_of_range("Block index out of range");
+ }
+ return Block(blocks[blockIndex],
+ std::min(currentSize - blockIndex * blockSize, blockSize));
+ }
+
+ Block BlockBuffer::getEmptyBlock() {
+ if (currentSize < currentCapacity) {
+ Block emptyBlock(blocks[currentSize / blockSize] + currentSize % blockSize,
+ blockSize - currentSize % blockSize);
+ currentSize = (currentSize / blockSize + 1) * blockSize;
+ return emptyBlock;
+ } else {
+ resize(currentSize + blockSize);
+ return Block(blocks.back(), blockSize);
+ }
+ }
+
+ void BlockBuffer::reserve(uint64_t capacity) {
+ while (currentCapacity < capacity) {
+ char* newBlockPtr = memoryPool.malloc(blockSize);
+ if (newBlockPtr != nullptr) {
+ blocks.push_back(newBlockPtr);
+ currentCapacity += blockSize;
+ } else {
+ break;
+ }
+ }
+ }
+
+ void BlockBuffer::resize(uint64_t size) {
+ reserve(size);
+ if (currentCapacity >= size) {
+ currentSize = size;
+ } else {
+ throw std::logic_error("Block buffer resize error");
+ }
+ }
+} // namespace orc

c++/src/BlockBuffer.hh

@@ -0,0 +1,90 @@
+
+/**
+ * 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.
+ */
+
+#ifndef ORC_MEMORYPOOL_IMPL_HH
+#define ORC_MEMORYPOOL_IMPL_HH
+
+#include "orc/MemoryPool.hh"
+
+#include <vector>
+
+namespace orc {
+
+ struct Block {
+ char* data;
+ uint64_t size;
+
+ Block() : data(nullptr), size(0) {}
+ Block(char* _data, uint64_t _size) : data(_data), size(_size) {}
+ Block(const Block& block) = default;
+ ~Block() {}
+ };
+
+ class BlockBuffer {
+ private:
+ MemoryPool& memoryPool;
+ // current buffer size
+ uint64_t currentSize;
+ // maximal capacity (actual allocated memory)
+ uint64_t currentCapacity;
+ // unit for buffer expansion
+ const uint64_t blockSize;
+ // pointers to the start of each block
+ std::vector<char*> blocks;
+
+ // non-copy-constructible
+ BlockBuffer(BlockBuffer& buffer) = delete;
+ BlockBuffer& operator=(BlockBuffer& buffer) = delete;
+ BlockBuffer(BlockBuffer&& buffer) = delete;
+ BlockBuffer& operator=(BlockBuffer&& buffer) = delete;
+
+ public:
+ BlockBuffer(MemoryPool& pool, uint64_t blockSize);
+
+ ~BlockBuffer();
+
+ /**
+ * Get the Block object
+ */
+ Block getBlock(uint64_t blockIndex);
+
+ /**
+ * Get a empty block or allocate a new block if the buffer is exhausted
+ */
+ Block getEmptyBlock();
+
+ /**
+ * Get the Block number
+ */
+ uint64_t getBlockNumber() const;
+
+ uint64_t size() const {
+ return currentSize;
+ }
+
+ uint64_t capacity() const {
+ return currentCapacity;
+ }
+
+ void resize(uint64_t size);
+ void reserve(uint64_t capacity);
+ };
+} // namespace
+
+#endif

c++/src/CMakeLists.txt

@@ -205,6 +205,7 @@ set(SOURCE_FILES
  sargs/TruthValue.cc
  wrap/orc-proto-wrapper.cc
  Adaptor.cc
+ BlockBuffer.cc
  BloomFilter.cc
  ByteRLE.cc
  ColumnPrinter.cc

c++/test/CMakeLists.txt

@@ -22,6 +22,7 @@ add_executable (orc-test
  MemoryInputStream.cc
  MemoryOutputStream.cc
  TestAttributes.cc
+ TestBlockBuffer.cc
  TestBufferedOutputStream.cc
  TestBloomFilter.cc
  TestByteRle.cc

c++/test/TestBlockBuffer.cc

@@ -0,0 +1,82 @@
+
+/**
+ * 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.
+ */
+
+#include "BlockBuffer.hh"
+#include "orc/OrcFile.hh"
+#include "wrap/gtest-wrapper.h"
+
+namespace orc {
+
+ TEST(TestBlockBuffer, size_and_capacity) {
+ MemoryPool* pool = getDefaultPool();
+ BlockBuffer buffer(*pool, 1024);
+
+ // block buffer will preallocate one block during initialization
+ EXPECT_EQ(buffer.getBlockNumber(), 0);
+ EXPECT_EQ(buffer.size(), 0);
+ EXPECT_EQ(buffer.capacity(), 1024);
+
+ buffer.reserve(128 * 1024);
+ EXPECT_EQ(buffer.getBlockNumber(), 0);
+ EXPECT_EQ(buffer.size(), 0);
+ EXPECT_EQ(buffer.capacity(), 128 * 1024);
+
+ // new size < old capacity
+ buffer.resize(64 * 1024);
+ EXPECT_EQ(buffer.getBlockNumber(), 64);
+ EXPECT_EQ(buffer.size(), 64 * 1024);
+ EXPECT_EQ(buffer.capacity(), 128 * 1024);
+
+ // new size > old capacity
+ buffer.resize(256 * 1024);
+ EXPECT_EQ(buffer.getBlockNumber(), 256);
+ EXPECT_EQ(buffer.size(), 256 * 1024);
+ EXPECT_EQ(buffer.capacity(), 256 * 1024);
+ }
+
+ TEST(TestBlockBuffer, get_block) {
+ MemoryPool* pool = getDefaultPool();
+ BlockBuffer buffer(*pool, 1024);
+
+ EXPECT_EQ(buffer.getBlockNumber(), 0);
+ for (uint64_t i = 0; i < 10; ++i) {
+ Block block = buffer.getEmptyBlock();
+ EXPECT_EQ(buffer.getBlockNumber(), i + 1);
+ for (uint64_t j = 0; j < block.size; ++j) {
+ if (i % 2 == 0) {
+ block.data[j] = static_cast<char>('A' + (i + j) % 26);
+ } else {
+ block.data[j] = static_cast<char>('a' + (i + j) % 26);
+ }
+ }
+ }
+
+ // verify the block data
+ for (uint64_t i = 0; i < buffer.getBlockNumber(); ++i) {
+ Block block = buffer.getBlock(i);
+ for (uint64_t j = 0; j < block.size; ++j) {
+ if (i % 2 == 0) {
+ EXPECT_EQ(block.data[j], 'A' + (i + j) % 26);
+ } else {
+ EXPECT_EQ(block.data[j], 'a' + (i + j) % 26);
+ }
+ }
+ }
+ }
+}