btree.c

基于btree索引算法的数据库代码

		while(key_sz_left)
		{
			assert(NULL == pg_ovfl);
			pg_ovfl = PagerGetPage(node->btree_mgr->hpgr, pgno_ovfl);
			if(NULL == pg_ovfl)
				goto btree_mgr_cell_get_data_err_;
			
			rb = PageHeadMakeReadable(pg_ovfl);
			if(NULL == rb)
				goto btree_mgr_cell_get_data_err_;

			if(key_sz_left > BTREE_OVERFLOW_PAYLOAD(node->btree_mgr))
			{
				array_to_uint_as_big_endian(rb, BTREE_OVERFLOW_HDR_SZ, pgno_ovfl);
				PagerReleasePage(pg_ovfl);
				pg_ovfl = NULL;

				key_sz_left -= BTREE_OVERFLOW_PAYLOAD(node->btree_mgr);
			}
			else
			{
				if(key_sz_left < BTREE_OVERFLOW_PAYLOAD(node->btree_mgr))
				{
					assert(data_sz >= key_sz_left);

					memcpy(data, rb + BTREE_OVERFLOW_HDR_SZ + key_sz_left, key_sz_left);
					data += key_sz_left;
					data_sz -= key_sz_left;
				}

				PagerReleasePage(pg_ovfl);
				pg_ovfl = PagerGetPage(node->btree_mgr->hpgr, pgno_ovfl);
				if(NULL == pg_ovfl)
					goto btree_mgr_cell_get_data_err_;

				rb = PageHeadMakeReadable(pg_ovfl);
				if(NULL == rb)
					goto btree_mgr_cell_get_data_err_;

				break;
			}
		}
	}
	else if(cell_info->key_sz <= cell_info->local_sz || (node->node_flag & e_btree_page_flag_intkey))
	{
		unsigned int copy_sz = cell_info->local_sz - cell_info->key_sz * ((node->node_flag & e_btree_page_flag_intkey) ? 0 : 1);

		assert(data_sz > copy_sz);

		/* 说明data在本页也有部分存储 */

		if(copy_sz)
		{
			memcpy(data,
				cell_info->cell + cell_info->head_sz + cell_info->key_sz * ((node->node_flag & e_btree_page_flag_intkey) ? 0 : 1),
				copy_sz);

			data += copy_sz;
			data_sz -= copy_sz;
		}

		pg_ovfl = PagerGetPage(node->btree_mgr->hpgr, cell_info->overflow_pgno);
		if(NULL == pg_ovfl)
			goto btree_mgr_cell_get_data_err_;

		rb = PageHeadMakeReadable(pg_ovfl);
		if(NULL == rb)
			goto btree_mgr_cell_get_data_err_;

		rb = PageHeadMakeReadable(pg_ovfl);
		if(NULL == rb)
			goto btree_mgr_cell_get_data_err_;
	}
	else
		assert(0);

	assert(pg_ovfl && PageHeadMakeReadable(pg_ovfl) == rb);

	while(data_sz)
	{
		assert(pg_ovfl);

		if(data_sz > BTREE_OVERFLOW_PAYLOAD(node->btree_mgr))
		{
			memcpy(data, rb + BTREE_OVERFLOW_HDR_SZ, BTREE_OVERFLOW_PAYLOAD(node->btree_mgr));

			data += BTREE_OVERFLOW_PAYLOAD(node->btree_mgr);
			data_sz -= BTREE_OVERFLOW_PAYLOAD(node->btree_mgr);

			PagerReleasePage(pg_ovfl);

			array_to_uint_as_big_endian(rb, BTREE_OVERFLOW_HDR_SZ, pgno_ovfl);
			pg_ovfl = PagerGetPage(node->btree_mgr->hpgr, pgno_ovfl);
			if(NULL == pg_ovfl)
				goto btree_mgr_cell_get_data_err_;
			rb = PageHeadMakeReadable(pg_ovfl);
			if(NULL == rb)
				goto btree_mgr_cell_get_data_err_;
		}
		else
		{
			memcpy(data, rb + BTREE_OVERFLOW_HDR_SZ, data_sz);
			data_sz = 0;
		}
	}

	return 0;

btree_mgr_cell_get_data_err_:

	if(pg_ovfl)
		PagerReleasePage(pg_ovfl);

	return -1;
}

/**
 * @brief 从cell里取出key
 */
static __INLINE__ int btree_mgr_cell_get_key(btree_node * node,
											 btree_cellinfo * cell_info,
											 unsigned char * key, unsigned int key_sz)
{
	assert(cell_info);

	/* 如果是整形key,赋值返回 */
	if(node->node_flag & e_btree_page_flag_intkey)
		return 0;

	/* 如果没有发生溢出,拷贝返回 */
	if(cell_info->local_sz == cell_info->payload_sz)
	{
		assert(key && key_sz >= cell_info->key_sz);
		assert(((sizeof(unsigned int)) * ((node->node_flag & e_btree_page_flag_leaf) ? 0 : 1 ) + 
			4 * ((node->node_flag & e_btree_page_flag_hasdata) ? 1 : 0) + 4) == cell_info->head_sz);

		memcpy(key, 
			cell_info->cell + cell_info->head_sz,
			cell_info->key_sz);

		return 0;
	}

	assert(node && node->btree_mgr);

	{
		btree_mgr_t * btree_mgr = node->btree_mgr;

		assert(key && key_sz >= cell_info->key_sz);

		/* 如果key都在本节点页存储,拷贝返回 */
		if(cell_info->key_sz <= cell_info->local_sz)
		{
			assert((4 * ((node->node_flag & e_btree_page_flag_leaf) ? 0 : 1) + 
				4 * ((node->node_flag & e_btree_page_flag_hasdata) ? 1 : 0) + 4 + 4) == cell_info->head_sz);

			memcpy(key,
				cell_info->cell + cell_info->head_sz,
				cell_info->key_sz);

			return 0;
		}
		else
		{
			/* 如果部key在溢出页中也有存储,将key_sz的内容全部拷贝出来 */
			unsigned int left = cell_info->key_sz - cell_info->local_sz;
			unsigned int pgno_ovfl = cell_info->overflow_pgno;
			HPAGE_HEAD pg_ovfl = NULL;

			memcpy(key,
				cell_info->cell + 4 * ((node->node_flag & e_btree_page_flag_leaf) ? 0 : 1) + 
				4 * ((node->node_flag & e_btree_page_flag_hasdata) ? 1 : 0) + 4 + 4,
				cell_info->local_sz);

			key += cell_info->local_sz;

			assert(pgno_ovfl);
			assert(left);

			while(left)
			{
				const unsigned char * rb = NULL;
				assert(NULL == pg_ovfl);

				pg_ovfl = PagerGetPage(btree_mgr->hpgr, pgno_ovfl);
				if(NULL == pg_ovfl)
					return -1;

				rb = PageHeadMakeReadable(pg_ovfl);
				assert(rb);

				if(left > BTREE_OVERFLOW_PAYLOAD(btree_mgr))
				{
					memcpy(key, &rb[BTREE_OVERFLOW_HDR_SZ], BTREE_OVERFLOW_PAYLOAD(btree_mgr));

					key += BTREE_OVERFLOW_PAYLOAD(btree_mgr);
					left -= BTREE_OVERFLOW_PAYLOAD(btree_mgr);

					PagerReleasePage(pg_ovfl);
					pg_ovfl = NULL;

					array_to_uint_as_big_endian(rb, sizeof(unsigned int), pgno_ovfl);
				}
				else
				{
					memcpy(key, &rb[BTREE_OVERFLOW_HDR_SZ], left);
					left = 0;

					PagerReleasePage(pg_ovfl);
					pg_ovfl = NULL;

					break;
				}
			}

			assert(!left);

			return 0;
		}
	}
}

/**
 * @brief 解析一个打包好的cell
 */
static __INLINE__ int btree_mgr_parse_cell(btree_node * node,
										   const unsigned char * cell, const unsigned int cell_buf_sz,
										   btree_cellinfo * cell_info)
{
	/**
	 * 一条记录的分布如下
	 *    SIZE    DESCRIPTION
	 *      4     Page number of the left child. Omitted if leaf flag is set.
	 *      4     Number of bytes of data. Omitted if the zerodata flag is set.
	 *      4     Number of bytes of key. Or the key itself if intkey flag is set.
	 *      4     First page of the overflow chain.  Omitted if no overflow
	 *      *     Payload
	 */

	unsigned int payload_sz = 0;
	unsigned int cell_sz = 0;

	assert(node && cell_info && cell);

	memset(cell_info, 0, sizeof(*cell_info));
	cell_info->cell = (unsigned char *)cell;

	if(!(node->node_flag & e_btree_page_flag_leaf))
	{
		/* 如果不是叶节点 */
		array_to_uint_as_big_endian(cell, sizeof(unsigned int), cell_info->lchild_pgno);
		cell_sz += sizeof(unsigned int);
		cell += sizeof(unsigned int);
	}

	if(node->node_flag & e_btree_page_flag_hasdata)
	{
		/* 如果节点含有数据域 */
		array_to_uint_as_big_endian(cell, sizeof(unsigned int), cell_info->data_sz);
		cell_sz += sizeof(unsigned int);
		cell += sizeof(unsigned int);
		payload_sz += cell_info->data_sz;
	}

	/* 解析key域 */
	array_to_uint_as_big_endian(cell, sizeof(unsigned int), cell_info->key_sz);
	cell_sz += sizeof(unsigned int);
	cell += sizeof(unsigned int);

	/* 如果不是整形的key */
	if(!(node->node_flag & e_btree_page_flag_intkey))
		payload_sz += cell_info->key_sz;

	/* 根据payload_sz的值,判断当前cell是否是溢出的 */
	if(payload_sz > node->btree_mgr->max_local)
	{
		unsigned int surplus = payload_sz % BTREE_OVERFLOW_PAYLOAD(node->btree_mgr);
		if(surplus <= node->btree_mgr->max_local)
			cell_info->local_sz = surplus;
		else
			cell_info->local_sz = node->btree_mgr->min_local;

		/* 查找overflow页号 */
		array_to_uint_as_big_endian(cell, sizeof(unsigned int), cell_info->overflow_pgno);
		cell_sz += sizeof(unsigned int);
		cell += sizeof(unsigned int);
	}
	else
		cell_info->local_sz = payload_sz;

	cell_info->head_sz = cell_sz;
	cell_sz += cell_info->local_sz;

	cell_info->payload_sz = payload_sz;
	cell_info->cell_sz = cell_sz;

	return 0;
}

/**
 * @brief 将cell"指针"指向的内容解析出来
 */
static __INLINE__ int btree_mgr_parse_cellptr(btree_node * node, unsigned int cell_ptr, btree_cellinfo * cell_info)
{
	assert(cell_info);
	assert(node && node->pg && node->read_buf);
	assert(node->read_buf == PageHeadMakeReadable(node->pg));

	{
		const unsigned char * cell = &node->read_buf[cell_ptr];

		assert(PagePtrIsInpage(node->pg, cell));

		btree_mgr_parse_cell(node, cell, node->btree_mgr->pg_sz - (cell - node->read_buf), cell_info);		
	}
	
	return 0;	
}

/**
 * @brief 获取指定idx的cell"指针"
 */
#define btree_mgr_find_cellptr(_n, _i, _cp) do{\
		assert((_n) && (_n)->pg && (_n)->pg_no);\
		assert((_n)->read_buf);\
		assert((_n)->read_buf == PageHeadMakeReadable((_n)->pg));\
		assert((_i) < (_n)->ncell);\
		get_2byte(&(_n)->read_buf[BTREE_CELL_PTR_OFFSET + (_i) * BTREE_CELL_PTR_SIZE], BTREE_CELL_PTR_SIZE, (_cp));\
		assert((_cp));\
	}while(0)

/**
 * @brief 解析相应idx对应cell的信息
 */
#define btree_mgr_parse_idx(_n, _i, _ci) do{\
		unsigned int __cell_ptr = 0;\
		assert((_n) && (_ci));\
		assert((_i) < _n->ncell);\
		btree_mgr_find_cellptr((_n), (_i), __cell_ptr);\
		btree_mgr_parse_cellptr((_n), __cell_ptr, (_ci));\
	}while(0)

/**
 * @brief 取得某个节点指定idx的lchild
 */
#define btree_mgr_get_idx_lchild(_n, _i, _lchild) do{\
		btree_cellinfo __ci;\
		assert((_n));\
		assert((_i) < (_n)->ncell);\
		btree_mgr_parse_idx(_n, _i, &__ci);\
		_lchild = __ci.lchild_pgno;\
	}while(0)

/**
 * @brief 填充cell info信息,计算cell_sz的大小(包含cell头大于,以及local payload,不包含溢出的payload)
 */
static __INLINE__ unsigned int btree_mgr_fill_cellinfo(btree_node * node,
													   unsigned int key_sz, unsigned data_sz,
													   btree_cellinfo * cell_info)
{
	/* key是一定存在的,其它根据页的属性以及payload的值有可能被忽略 */
	unsigned int cell_sz = sizeof(unsigned int);
	unsigned int payload_sz = 0;

	assert(node && node->btree_mgr);
	assert(cell_info);

	memset(cell_info, 0, sizeof(*cell_info));
	cell_info->key_sz = key_sz;

	/* 是否是叶子节点 */
	if(!(node->node_flag & e_btree_page_flag_leaf))
		cell_sz += sizeof(unsigned int);

	/* 关键字是否为整形 */
	if(!(node->node_flag & e_btree_page_flag_intkey))
		payload_sz += key_sz;

	/* 是否包含数据域 */
	if(node->node_flag & e_btree_page_flag_hasdata)
	{
		cell_info->data_sz = data_sz;
 
		cell_sz += sizeof(unsigned int);
		payload_sz += data_sz;
	}

	/* payload是否溢出 */
	if(payload_sz <= node->btree_mgr->max_local)
	{
		cell_info->head_sz = cell_sz;
		cell_sz += payload_sz;
		cell_info->local_sz = payload_sz;
	}
	else
	{
		/* todo:此段逻辑与btree_mgr_parse_cellptr中的对应的部分可以再做一次提炼 */
		unsigned surplus = payload_sz % BTREE_OVERFLOW_PAYLOAD(node->btree_mgr);

		/* 加入溢出页页号存储空间 */
		cell_sz += sizeof(unsigned int);
		cell_info->head_sz = cell_sz;

		if(surplus <= node->btree_mgr->max_local)
			cell_info->local_sz = surplus;
		else
			cell_info->local_sz = node->btree_mgr->min_local;

		cell_sz += cell_info->local_sz;
	}
	
	cell_info->payload_sz = payload_sz;
	return cell_info->cell_sz = cell_sz;
}

/**
 * @brief 打包用户的数据,形成一个cell
 */
static __INLINE__ int btree_mgr_pack_cell(btree_node * node,
										  btree_cellinfo * cell_info,
										  unsigned char * cell, const unsigned int cell_sz,
										  const void * key, const unsigned int key_sz,
										  const void * data, const unsigned int data_sz)
{
	/**
	 * 一条记录的分布如下
	 *    SIZE    DESCRIPTION
	 *      4     Page number of the left child. Omitted if leaf flag is set.
	 *      4     Number of bytes of data. Omitted if the zerodata flag is set.
	 *      4     Number of bytes of key. Or the key itself if intkey flag is set.
	 *      4     First page of the overflow chain.  Omitted if no overflow
	 *      *     Payload
	 */

	int ret = 0;

	btree_mgr_t * btree_mgr = NULL;

	/* 记录溢出节点的临时指针 */
	HPAGE_HEAD pg_ovfl = NULL;

	/* 记录payload的大小 */
	unsigned int payload_sz = 0;

	/* 临时指针,指向接收payload的存储空间 */
	unsigned char * pPayload = NULL;

	/* 临时存储当前用于存储payload的空间剩余大小 */
	unsigned int space_left = 0;

	/* 临时指针,指向为溢出页号预留的存储空间 */
	unsigned char * pPrior = NULL;

	/* 临时指针,初始值指向key,如果key为整形,则指向data(如果有data的话) */
	unsigned char * pSrc = NULL;
	unsigned int Src_sz = 0;

	assert(node && node->btree_mgr);
	assert(cell && cell_sz && cell_info);
	assert(cell_sz == cell_info->cell_sz);
	assert(data_sz == cell_info->data_sz);
	assert(key_sz == cell_info->key_sz);

	cell_info->cell = cell;

	/* 如果不是叶节点,要打包左孩子的页索引 */
	if(!(node->node_flag & e_btree_page_flag_leaf))
	{
		assert(cell_info->lchild_pgno);

		uint_to_big_endian(cell_info->lchild_pgno, cell, sizeof(unsigned int));
		cell += sizeof(sizeof(unsigned int));

		assert((unsigned int)(cell - cell_info->cell) <= cell_sz);
	}

	/* 如果含有数据域,要打包数据域的长度 */
	if(node->node_flag & e_btree_page_flag_hasdata)
	{
		assert(data && data_sz);

		uint_to_big_endian(data_sz, cell, sizeof(unsigned int));
		cell += sizeof(sizeof(unsigned int));

		assert((unsigned int)(cell - cell_info->cell) < cell_sz);