void
testReadPage ()
{
  BM_BufferPool *bm = MAKE_POOL();
  BM_PageHandle *h = MAKE_PAGE_HANDLE();
  testName = "Reading a page";

  CHECK(createPageFile("testbuffer.bin"));
  CHECK(initBufferPool(bm, "testbuffer.bin", 3, RS_FIFO, NULL));
  
  CHECK(pinPage(bm, h, 0));
  CHECK(pinPage(bm, h, 0));

  CHECK(markDirty(bm, h));

  CHECK(unpinPage(bm,h));
  CHECK(unpinPage(bm,h));

  CHECK(forcePage(bm, h));

  CHECK(shutdownBufferPool(bm));
  CHECK(destroyPageFile("testbuffer.bin"));

  free(bm);
  free(h);

  TEST_DONE();
}

recordTableInfo *initBufferManagerForRecDetails(char *name,
		struct BM_BufferPool *bManager, struct BM_PageHandle *pageHandle) {
	recordTableInfo *recordTable = NULL;
	initBufferPool(bManager, name, 3, RS_FIFO, NULL);
	if(pinPage(bManager, pageHandle, 0)==RC_OK)
	recordTable = TranslateStringRecordTabDetails(pageHandle->data);
	if (PAGE_SIZE > recordTable->sizeOfSchema)
		pinPage(bManager, pageHandle, 1);
	return recordTable;
}

// Simply marks tombstones, doesn't physically delete or allow for records to overwrite
RC deleteRecord(RM_TableData *rel, RID id) {
	RM_TableMgmtData *tableMgmtData = rel->mgmtData;
	RM_Page *page;

	RC pinPageResult;
	if (pinPageResult = pinPage(&tableMgmtData->bufferPool, &tableMgmtData->pageHandle, id.page) != RC_OK) {
		return pinPageResult;
	}

	page = (RM_Page*)tableMgmtData->pageHandle.data;
	char *slotOffset = ((char*)&page->data) + (id.slot*getRecordSize(rel->schema));


	// Update tombstone
	*(char*)slotOffset = -1;

	RC markDirtyResult;
	if (markDirtyResult = markDirty(&tableMgmtData->bufferPool, &tableMgmtData->pageHandle) != RC_OK) {
		return markDirtyResult;
	}

	RC unpinPageResult;
	if (unpinPageResult = unpinPage(&tableMgmtData->bufferPool, &tableMgmtData->pageHandle) != RC_OK) {
		return unpinPageResult;
	}

	tableMgmtData->numTuples--;

	return RC_OK;
}

RC updateRecord(RM_TableData *rel, Record *record) {
	RID *rid = &record->id;
	RM_TableMgmtData *tableMgmtData = rel->mgmtData;
	RM_Page *page;

	//  Check to see if feasible record
	if (rid->page == NO_PAGE || rid->slot == NO_SLOT) {
		return RC_RM_NO_RECORD;
	}

	RC pinPageResult;
	if (pinPageResult = pinPage(&tableMgmtData->bufferPool, &tableMgmtData->pageHandle, rid->page) != RC_OK) {
		return pinPageResult;
	}
	page = (RM_Page*)tableMgmtData->pageHandle.data;

	char *slotOffset = ((char*)&page->data) + (rid->slot * getRecordSize(rel->schema));


	int recordSize = getRecordSize(rel->schema);
	memcpy(slotOffset + 1, record->data, recordSize);

	RC markDirtyResult;
	if (markDirtyResult = markDirty(&tableMgmtData->bufferPool, &tableMgmtData->pageHandle) != RC_OK) {
		return markDirtyResult;
	}

	RC unpinPageResult;
	if (unpinPageResult = unpinPage(&tableMgmtData->bufferPool, &tableMgmtData->pageHandle) != RC_OK) {
		return unpinPageResult;
	}

	return RC_OK;
}

void addFreeSlot(RM_TableMgmtData *tableMgmtData, RM_Page *page, int pageNum) {
	// Am I the free slot?
	if (tableMgmtData->nextFreeSlot < 1) {
		page->next = page->prev = pageNum;
		tableMgmtData->nextFreeSlot = pageNum;
	}
	// Add the slot to the list
	else {
		BM_PageHandle pageHandle;
		RM_Page *firstPage;

		RC pinPageResult;
		if (pinPageResult = pinPage(&tableMgmtData->bufferPool, &pageHandle, tableMgmtData->nextFreeSlot) != RC_OK) {
			return;
		}
		firstPage = (RM_Page*)pageHandle.data;

		RC markDirtyResult;
		if (markDirtyResult = markDirty(&tableMgmtData->bufferPool, &pageHandle) != RC_OK) {
			return;
		}
		firstPage->prev = pageNum;

		RC unpinPageResult;
		if (unpinPageResult = unpinPage(&tableMgmtData->bufferPool, &pageHandle) != RC_OK) {
			return;
		}

		page->next = tableMgmtData->nextFreeSlot;
		tableMgmtData->nextFreeSlot = pageNum;
		page->prev = 0;
	}
}

RC setPageInfo(BTreeHandle *tree, int value, PageNumber pageNum, int pos) {
	BM_PageHandle *page = (BM_PageHandle *) malloc(sizeof(BM_PageHandle));
	RC rc;

	// pin page 0, page 0 is the information page of b+ tree
	rc = -99;
	rc = pinPage(BM, page, pageNum);

	if (rc != RC_OK) {
		return rc;
	}

	// set the first # of page value
	int *ip = (int *) page->data;
	ip[pos] = value;

	// unpin page 0
	rc = -99;
	rc = unpinPage(BM, page);

	if (rc != RC_OK) {
		return rc;
	}

	free(page);

	return RC_OK;
}

RC getPageInfo(BTreeHandle *tree, PageNumber pageNum, int pos, int *result) {
	BM_PageHandle *page = (BM_PageHandle *) malloc(sizeof(BM_PageHandle));
	RC rc;

	// pin page 0, page 0 is the information page of b+ tree
	rc = -99;
	rc = pinPage(BM, page, pageNum);

	if (rc != RC_OK) {
		return rc;
	}

	// get the first # of page 0, is # of nodes
	int *ip = (int *) page->data;
	*result = ip[pos];

	// unpin page 0
	rc = -99;
	rc = unpinPage(BM, page);

	if (rc != RC_OK) {
		return rc;
	}

	free(page);

	return RC_OK;
}

void 
checkDummyPages(BM_BufferPool *bm, int num)
{
  int i;
  BM_PageHandle *h = MAKE_PAGE_HANDLE();
  char *expected = malloc(sizeof(char) * 512);

  CHECK(initBufferPool(bm, "testbuffer.bin", 3, RS_FIFO, NULL));

  for (i = 0; i < num; i++)
    {
      CHECK(pinPage(bm, h, i));

      sprintf(expected, "%s-%i", "Page", h->pageNum);

      ASSERT_EQUALS_STRING(expected, h->data, "reading back dummy page content");
	 
      CHECK(unpinPage(bm,h));
    }

  CHECK(shutdownBufferPool(bm));

  free(expected);
  free(h);
}

void 
createDummyPages(BM_BufferPool *bm, int num)
{
  int i;
  BM_PageHandle *h = MAKE_PAGE_HANDLE();

  CHECK(initBufferPool(bm, "testbuffer.bin", 3, RS_FIFO, NULL));

  for (i = 0; i < num; i++)
    {
      CHECK(pinPage(bm, h, i));
	
	  sprintf(h->data, "%s-%i", "Page", h->pageNum);

	  CHECK(markDirty(bm, h));
	
	  CHECK(unpinPage(bm,h));

    }

  CHECK(shutdownBufferPool(bm));

  free(h);

}

/***************************************************************
 * Function Name: getNode 
 * 
 * Description:get node from tree
 *
 * Parameters:BTreeHandle *tree, int nodeNum, BT_Node **node
 *
 * Return: RC
 *
 * Author:Xincheng Yang
 *
 * History:
 *      Date            Name                        Content
 *  04/9/2016          Xincheng Yang             finish this function
***************************************************************/
RC getNode (BTreeHandle *tree, int nodeNum, BT_Node **node)
{
    BM_PageHandle *pg = (BM_PageHandle *)calloc(1, sizeof(BM_PageHandle));
    RC rv = RC_OK;
    bool isNull = false;
    int maxsize=2*(tree->n+2);
    
    if(*node == NULL){
        isNull = true;
        *node = (BT_Node *)calloc(1, sizeof(BT_Node));
    } 

    if((rv = pinPage(tree->bufferPool, pg, nodeNum)) == RC_OK){
        memcpy(&(*node)->isValid, pg->data, sizeof(int));
        if((*node)->isValid == 0){
            unpinPage(tree->bufferPool, pg);
            rv = RC_IM_NODE_NOT_EXIST;
        } else {
            memcpy(&(*node)->parent, pg->data + sizeof(int), sizeof(int));
            memcpy(&(*node)->current, pg->data + 2*sizeof(int) , sizeof(int));
            memcpy(&(*node)->size, pg->data + 3*sizeof(int), sizeof(int));
            memcpy(&(*node)->nodeType, pg->data + 4*sizeof(int), sizeof(int));
            if(isNull || (*node)->element == NULL){
                (*node)->element = (BT_Element *)malloc(sizeof(BT_Element) * maxsize);
            } 
            memcpy((*node)->element, pg->data + 5*sizeof(int), sizeof(BT_Element) * (*node)->size);
            rv = unpinPage(tree->bufferPool, pg);
        }
    }
    
    free(pg);
    return rv;
}

RC getRecord (RM_TableData *rel, RID id, Record *record){
    int offset=0;
    pinPage(bm, page, id.page);
    offset=numPageHeader+id.slot*slotSize+sizeof(RID);
    record->id=id;
    record->data=page->data+offset;
    unpinPage(bm, page);
    return RC_OK;
}

void removeFromMiddle(TableManager *tableManager, Pager *dp, int pageno){
        BM_PageHandle pageHandler2;
         BM_PageHandle pageHandler;
        Pager *pagerData;
        Pager *tmp_dp2;
        pinPage(&tableManager->pool, &pageHandler, (PageNumber)dp->prev);
        pinPage(&tableManager->pool, &pageHandler2, (PageNumber)dp->next);
        pagerData= (Pager*) pageHandler.data;
        tmp_dp2= (Pager*) pageHandler2.data;
        markDirty(&tableManager->pool, &pageHandler);
        markDirty(&tableManager->pool, &pageHandler2);
        pagerData->next= dp->next;
        tmp_dp2->prev= dp->prev;
        unpinPage(&tableManager->pool, &pageHandler);
        unpinPage(&tableManager->pool, &pageHandler2);
        dp->next=dp->prev= 0;

}

RC updateRecord (RM_TableData *rel, Record *record){
    int offset=0;
    pinPage(bm, page, record->id.page);
    offset=numPageHeader+record->id.slot*slotSize;
    offset+=sizeof(RID);
    memcpy(page->data+offset,record->data, slotSize-sizeof(RID));
    markDirty(bm, page);
    unpinPage(bm, page);
    return RC_OK;
}

RC insertLeafNode(BTreeHandle *tree, BT_KeyPosition *kp, Value *key, RID *rid) {
	BM_PageHandle *page = (BM_PageHandle *) malloc(sizeof(BM_PageHandle));
	RC rc;

	// get N
	int n;
	rc = -99;
	rc = getN(tree, &n);

	if (rc != RC_OK) {
		return rc;
	}

	// pin the page
	rc = pinPage(BM, page, kp->nodePage);

	if (rc != RC_OK) {
		return rc;
	}

	int *ip = (int *) page->data;

	// get current keys
	int currentKeys = ip[1];

	// i is the index of key and key pointer, starts from 0
	int i;
	for (i = currentKeys - 1; i > kp->keyPos - 1; i--) {
		ip[4 + i + 1] = ip[4 + i];
		ip[4 + n + 2 * (i + 1)] = ip[4 + n + 2 * i];
		ip[4 + n + 2 * (i + 1) + 1] = ip[4 + n + 2 * i + 1];
	}

	// After the for loop, the i is pointing the left of the key position
	// put the key and RID into the right slot
	ip[4 + i + 1] = key->v.intV;
	ip[4 + n + 2 * (i + 1)] = rid->page;
	ip[4 + n + 2 * (i + 1) + 1] = rid->slot;

	// increment current
	ip[1]++;

	// unpin the page
	rc = -99;
	rc = unpinPage(BM, page);

	if (rc != RC_OK) {
		return rc;
	}

	free(page);

	return RC_OK;
}

RC deleteOldKey(BTreeHandle *tree, BT_KeyPosition *kp) {
	BM_PageHandle *page = (BM_PageHandle *) malloc(sizeof(BM_PageHandle));
	RC rc;

	// get N
	int n;
	rc = -99;
	rc = getN(tree, &n);

	if (rc != RC_OK) {
		return rc;
	}

	// pin the page
	rc = pinPage(BM, page, kp->nodePage);

	if (rc != RC_OK) {
		return rc;
	}

	int *ip = (int *) page->data;

	// get current keys
	int currentKeys = ip[1];

	int i;
	for (i = kp->keyPos + 1; i < currentKeys; i++) {
		ip[4 + i] = ip[4 + i + 1];
		ip[4 + n + 2 * i] = ip[4 + n + 2 * (i + 1)];
		ip[4 + n + 2 * i + 1] = ip[4 + n + 2 * (i + 1) + 1];
	}

	// After the for loop, the i is pointing the currentKeys position
	// unset the key and keyPointer of currentKeys position
	ip[4 + i] = 0;
	ip[4 + n + 2 * i] = 0;
	ip[4 + n + 2 * i + 1] = 0;

	// decrement current
	ip[1]--;

	// unpin the page
	rc = -99;
	rc = unpinPage(BM, page);

	if (rc != RC_OK) {
		return rc;
	}

	free(page);

	return RC_OK;
}

RC openTable (RM_TableData *rel, char *name){
    page=(BM_PageHandle *)calloc(PAGE_SIZE,sizeof(BM_PageHandle));
    Schema *schema;
    bm=MAKE_POOL();
    initBufferPool(bm,name, 3, RS_FIFO, NULL);
    pinPage(bm, page,0);
    schema=stringToSchema(page->data);
    rel->name=name;
    rel->schema=schema;
    rel->mgmtData=bm;
    return RC_OK;
}

/*
 * This function is used to update a Record,
 * updating on a deleted page is not possible
 */
RC updateRecord (RM_TableData *rel, Record *record)
{
	//Find the data to be updated
	if(record->id.page > 0 && record->id.page <=  totalPages)
	{
		BM_PageHandle *page = MAKE_PAGE_HANDLE();

		int pageNum, slotNum;

		// Setting record id and slot number
		pageNum = record->id.page;
		slotNum = record->id.slot;

		//Compare if the record is a deleted Record,
		//return update not possible for deleted records (EC 401)

		if(strncmp(record->data, "DELETED_RECORD", 14) == 0)
			return RC_RM_UPDATE_NOT_POSSIBLE_ON_DELETED_RECORD;

		//Take the serailized updated record data
		char *record_str = serializeRecord(record, rel->schema);

		//pin page
		pinPage(((RM_RecordMgmt *)rel->mgmtData)->bm, page, record->id.page);

		//set the new fields, or the entire modified page
		memset(page->data, '\0', strlen(page->data));
		sprintf(page->data, "%s", record_str);

		//free the temp data
		free(record_str);

		//mark the page as dirty
		markDirty(((RM_RecordMgmt *)rel->mgmtData)->bm, page);

		//unpin the page, after use
		unpinPage(((RM_RecordMgmt *)rel->mgmtData)->bm, page);

		//force write onto the page, as it is modified page now
		forcePage(((RM_RecordMgmt *)rel->mgmtData)->bm, page);

		//printf("record data in update: %s\n", page->data);

		free(page);		//free page, avoid leaks
		return RC_OK;
	}
	else
	{
		return RC_RM_NO_MORE_TUPLES;	//return the data to be modfied not found
	}

	return RC_OK;
}

void removeFromTail(TableManager *tableManager, Pager *dp, int pageno){
      BM_PageHandle pageHandler;
        Pager *pagerData;

        pinPage(&tableManager->pool, &pageHandler, (PageNumber)dp->prev);
        pagerData= (Pager*) pageHandler.data;
        markDirty(&tableManager->pool, &pageHandler);
        pagerData->next= 0;
        unpinPage(&tableManager->pool, &pageHandler);
        dp->next=dp->prev= 0;

}

void
testClock(void)
{
    // expected results
    const char *poolContents[]= {
        
    "[3x0],[-1 0],[-1 0],[-1 0]",
    "[3x0],[2 0],[-1 0],[-1 0]",
    "[3x0],[2 0],[0 0],[-1 0]",
    "[3x0],[2 0],[0 0],[8 0]",
    "[4 0],[2 0],[0 0],[8 0]",
    "[4 0],[2 0],[0 0],[8 0]",
    "[4 0],[2 0],[5 0],[8 0]",
    "[4 0],[2 0],[5 0],[0 0]",
    "[9 0],[2 0],[5 0],[0 0]",
    "[9 0],[8 0],[5 0],[0 0]",
    "[9 0],[8 0],[3x0],[0 0]",
    "[9 0],[8 0],[3x0],[2 0]"
    };
    const int orderRequests[]= {3,2,0,8,4,2,5,0,9,8,3,2};
        
    int i;
    int snapshot = 0;
    BM_BufferPool *bm = MAKE_POOL();
    BM_PageHandle *h = MAKE_PAGE_HANDLE();
    testName = "Testing CLOCK page replacement";

    CHECK(createPageFile("testbuffer.bin"));
    createDummyPages(bm, 100);
    CHECK(initBufferPool(bm, "testbuffer.bin", 4, RS_CLOCK, NULL));
    
    for (i=0;i<11;i++)
    {
        pinPage(bm,h,orderRequests[i]);
	if(orderRequests[i] == 3)
		markDirty(bm,h);
        unpinPage(bm,h);
        ASSERT_EQUALS_POOL(poolContents[snapshot++], bm, "check pool content using pages");
    }
    
    forceFlushPool(bm);
    // check number of write IOs
    ASSERT_EQUALS_INT(2, getNumWriteIO(bm), "check number of write I/Os");
    ASSERT_EQUALS_INT(10, getNumReadIO(bm), "check number of read I/Os");
    
    CHECK(shutdownBufferPool(bm));
    CHECK(destroyPageFile("testbuffer.bin"));
    
    free(bm);
    free(h);
    TEST_DONE();
}

/*
 * This function is used to Open a Created Table,
 * for any operation to be performed, the table has to be opened first
 * Also store the Schema related info by deserializing the schema,
 * in the Table Data
 */
RC openTable (RM_TableData *rel, char *name)
{
	//Record Management to store record attributes
	RM_RecordMgmt *rm_mgmt = (RM_RecordMgmt*)malloc(sizeof(RM_RecordMgmt));

	FILE *fptr;	//FILE pointer
	fptr = fopen(name, "r+");	//Open the PageFile (Table)

	//Get the total Number of Pages in the page File
	char* readHeader;
	readHeader = (char*)calloc(PAGE_SIZE,sizeof(char));

	fgets(readHeader,PAGE_SIZE,fptr);

	char* totalPage;
	totalPage = readHeader;
	totalPages = atoi(totalPage);	//convert to integer

	//Make a Buffer Pool
	rm_mgmt->bm = MAKE_POOL();

	//Make a Page Handle
	BM_PageHandle *page = MAKE_PAGE_HANDLE();

	//Initialize the BufferPool
	initBufferPool(rm_mgmt->bm,name,6,RS_FIFO,NULL);

	//Pin the Page = 0, which has the Schema Information
	pinPage(rm_mgmt->bm,page,0);

	//FreePages are stored in an array
	rm_mgmt->freePages = (int*)malloc(sizeof(int));
	rm_mgmt->freePages[0] = totalPages;

	//initialize the table data attributes

	//Deserialzing the Schema gives us the Relation information (i.e. Schema info)
	rel->schema = deserializeSchema(page->data);

	//store the name of the schema
	rel->name = name;

	//store the record management details
	rel->mgmtData = rm_mgmt;

	//Free the temp. memory allocations
	free(readHeader);
	free(page);

	return RC_OK;
}