added datablock out of order ops

src/util/datablock/datablock.c

@@ -180,3 +180,28 @@ void DataBlock_Free(DataBlock *dataBlock) {
 	assert(pthread_mutex_destroy(&dataBlock->mutex) == 0);
 	rm_free(dataBlock);
 }
+
+/* --------- Out of Order interface implementation --------*/
+
+void *DataBlock_AllocateItemOutOfOrder(DataBlock *dataBlock, uint64_t idx) {
+	DataBlock_Accommodate(dataBlock, idx);
+	dataBlock->itemCount++;
+	Block *block = GET_ITEM_BLOCK(dataBlock, idx);
+	idx = ITEM_POSITION_WITHIN_BLOCK(idx);
+	DataBlockItemHeader *item_header = (DataBlockItemHeader *)block->data + (idx * block->itemSize);
+	MARK_HEADER_AS_NOT_DELETED(item_header);
+	return ITEM_DATA(item_header);
+}
+
+void DataBlock_DeleteItemOutOfOrder(DataBlock *dataBlock, uint64_t idx) {
+	DataBlock_Accommodate(dataBlock, idx);
+	// Get block.
+	uint blockIdx = ITEM_INDEX_TO_BLOCK_INDEX(idx);
+	Block *block = dataBlock->blocks[blockIdx];
+	uint blockPos = ITEM_POSITION_WITHIN_BLOCK(idx);
+	uint offset = blockPos * block->itemSize;
+	DataBlockItemHeader *item_header = (DataBlockItemHeader *)block->data + offset;
+	// Delete
+	MARK_HEADER_AS_DELETED(item_header);
+	dataBlock->deletedIdx = array_append(dataBlock->deletedIdx, idx);
+}

src/util/datablock/datablock.h

@@ -81,3 +81,20 @@ void DataBlock_DeleteItem(DataBlock *dataBlock, uint64_t idx);
 
 // Free block.
 void DataBlock_Free(DataBlock *block);
+
+/*----------- Data Block Out of Order Building -----------*/
+
+/*
+ * Note!!!
+ * This API is to be used only during out of order re-construction of a data block.
+ * DO NOT USE THE BELOW METHODS COMBINED WITH THE NORMAL DATA BLACK API.
+ */
+
+// Return a pointer to allocated item in a given index.
+// If needed, allocates new space.
+// The values in indices smaller then idx are undefined if not allocated before.
+void *DataBlock_AllocateItemOutOfOrder(DataBlock *dataBlock, uint64_t idx);
+
+// Marks an item within a given dataBlock, in a given index as deleted. Allocates additional space if need.
+// Item distructor is not called in this call.
+void DataBlock_DeleteItemOutOfOrder(DataBlock *dataBlock, uint64_t idx);

tests/unit/test_datablock.cpp

@@ -187,3 +187,49 @@ TEST_F(DataBlockTest, RemoveItem) {
 	DataBlock_Free(dataBlock);
 }
 
+TEST_F(DataBlockTest, OutOfOrderBuilding) {
+	// This test checks for a fragmented, data block out of order re-construction.
+	DataBlock *dataBlock = DataBlock_New(1, sizeof(int), NULL);
+	int insert_arr1[4] = {8, 2, 3, 6};
+	int delete_arr[2] = {4, 7};
+	int insert_arr2[4] = {9, 1, 5, 0};
+	int expected[8] = {0, 1, 2, 3, 5, 6, 8, 9};
+
+	// Insert the first array.
+	for(int i = 0; i < 4; i++) {
+		int *item = (int *)DataBlock_AllocateItemOutOfOrder(dataBlock, insert_arr1[i]);
+		*item = insert_arr1[i];
+	}
+	ASSERT_EQ(4, dataBlock->itemCount);
+	ASSERT_EQ(0, array_len(dataBlock->deletedIdx));
+
+	// Mark deleted values.
+	for(int i = 0; i < 2; i++)
+		DataBlock_DeleteItemOutOfOrder(dataBlock, delete_arr[i]);
+
+	ASSERT_EQ(4, dataBlock->itemCount);
+	ASSERT_EQ(2, array_len(dataBlock->deletedIdx));
+
+	// Add another array
+	for(int i = 0; i < 4; i++) {
+		int *item = (int *)DataBlock_AllocateItemOutOfOrder(dataBlock, insert_arr2[i]);
+		*item = insert_arr2[i];
+	}
+
+	ASSERT_EQ(8, dataBlock->itemCount);
+	ASSERT_EQ(2, array_len(dataBlock->deletedIdx));
+
+	// Validate
+	DataBlockIterator *it = DataBlock_Scan(dataBlock);
+	for(int i = 0; i < 8; i++) {
+		int *item = (int *)DataBlockIterator_Next(it);
+		ASSERT_EQ(*item, expected[i]);
+	}
+	ASSERT_FALSE(DataBlockIterator_Next(it));
+
+	ASSERT_TRUE(dataBlock->deletedIdx[0] == 4 || dataBlock->deletedIdx[0] == 7);
+	ASSERT_TRUE(dataBlock->deletedIdx[1] == 4 || dataBlock->deletedIdx[1] == 7);
+
+	DataBlock_Free(dataBlock);
+}
+