Fix B-tree split algorithm

Instead of naive picking of counts, do detailed consideration of free space in
pages.

Author: Alexander Korotkov

contrib/in_memory/btree.c

@@ -26,6 +26,10 @@
 #include "btree_private.h"
 #include "fdw.h"
 
+static int get_split_item_size(Page p, OffsetNumber newoffset,
+				LocationIndex newitem_size, bool replace, OffsetNumber offset);
+static OffsetNumber find_split_location(Page p, OffsetNumber offset,
+				LocationIndex tuplesize, bool replace, float target_ratio);
 static PageNumber btree_split(BTreeDescr *desc, Page p, OffsetNumber *offset,
 			bool *place_right, Pointer tupleheader, Pointer tuple,
 			OffsetNumber tuplesize, bool replace, uint32 *state, CommitSeqNo *csn);
@@ -54,6 +58,117 @@ init_btree(BTreeDescr *desc)
 	init_meta_page(GET_PAGE(desc->meta), 1);
 }
 
+/*
+ * Get size of item during split consideration.
+ */
+static int
+get_split_item_size(Page p, OffsetNumber newoffset, LocationIndex newitem_size,
+					bool replace, OffsetNumber offset)
+{
+	if (offset < newoffset)
+		return BTREE_PAGE_GET_ITEM_SIZE(p, offset) + sizeof(OffsetNumber);
+	else if (offset == newoffset)
+		return newitem_size + sizeof(OffsetNumber);
+	else if (replace)
+		return BTREE_PAGE_GET_ITEM_SIZE(p, offset) + sizeof(OffsetNumber);
+	else
+		return BTREE_PAGE_GET_ITEM_SIZE(p, offset - 1) + sizeof(OffsetNumber);
+}
+
+/*
+ * Find appropriate location to split.  Return number of tuples to be placed
+ * to the left page.
+ */
+static OffsetNumber
+find_split_location(Page p, OffsetNumber offset, LocationIndex tuplesize,
+					bool replace, float target_ratio)
+{
+	int				left_free_space;
+	int				right_free_space;
+	int				left_bound;
+	int				right_bound;
+	int				first_data_key = BTREE_FIRST_DATA_KEY(p);
+	int				count;
+	bool			left_bounded = false;
+	bool			right_bounded = false;
+	bool			leaf = PAGE_IS_LEAF(p);
+	LocationIndex	header_size = leaf ? OLeafTupleHeaderSize : OInternalTupleHeaderSize;
+	LocationIndex	newitem_size = sizeof(OffsetNumber) + header_size + tuplesize;
+	LocationIndex	item_size;
+
+	/*
+	 * Covert ratio of fillfactor to ratio of free space.
+	 */
+	target_ratio = target_ratio / (1.0 - target_ratio);
+
+	count = BTREE_PAGE_ITEMS_COUNT(p) - first_data_key + (replace ? 0 : 1);
+	left_free_space = right_free_space = IN_MEMORY_BLCKSZ - offsetof(BTreePageHeader, items);
+
+	/*
+	 * Left and right pages initially contain one item each.  Left page also
+	 * reserves space for high key.  For leafs, We assume that high key
+	 * couldn't be wider than than source tuple.
+	 */
+	left_bound = 1;
+	left_free_space -= get_split_item_size(p, offset, newitem_size,
+										   replace, first_data_key);
+	left_free_space -= get_split_item_size(p, offset, newitem_size,
+										   replace, first_data_key + 1);
+
+	right_bound = count - 1;
+	right_free_space -= get_split_item_size(p, offset, newitem_size,
+											replace, first_data_key + count - 1);
+	if (!PAGE_IS_RIGHTMOST(p))
+		right_free_space -= BTREE_PAGE_GET_ITEM_SIZE(p, BTREE_HIKEY) + sizeof(OffsetNumber);
+
+	Assert(left_free_space >= 0 && right_free_space >= 0);
+
+	/*
+	 * Iterate shifting left bound upper and right bound lower until those
+	 * bounds meet each other.
+	 */
+	while (left_bound < right_bound)
+	{
+		if (right_bounded || (!left_bounded &&
+			(float) left_free_space * target_ratio > (float) right_free_space))
+		{
+			Assert(!left_bounded);
+			item_size = get_split_item_size(p, offset, newitem_size, replace,
+											first_data_key + left_bound + 1);
+			if (left_free_space >= item_size)
+			{
+				left_free_space -= item_size;
+				left_bound++;
+			}
+			else
+			{
+				left_bounded = true;
+			}
+		}
+		else
+		{
+			Assert(!right_bounded);
+			item_size = get_split_item_size(p, offset, newitem_size, replace,
+											first_data_key + right_bound - 1);
+			if (right_free_space >= item_size)
+			{
+				right_free_space -= item_size;
+				right_bound--;
+			}
+			else
+			{
+				right_bounded = true;
+			}
+		}
+	}
+
+	Assert(left_bound == right_bound);
+	return left_bound;
+}
+
+/*
+ * Split B-tree page into two.
+ */
 static PageNumber
 btree_split(BTreeDescr *desc, Page p, OffsetNumber *offset, bool *place_right,
 			Pointer tupleheader, Pointer tuple, LocationIndex tuplesize,
@@ -85,7 +200,10 @@ btree_split(BTreeDescr *desc, Page p, OffsetNumber *offset, bool *place_right,
 	{
 		right_count = count / 2;
 	}
-	left_count = (count - right_count) + BTREE_FIRST_DATA_KEY(p);
+	left_count = find_split_location(p, *offset, tuplesize, replace,
+									 was_rightmost ? 0.9 : 0.5);
+	right_count = count - left_count;
+	left_count += BTREE_FIRST_DATA_KEY(p);
 	if (*offset < left_count)
 	{
 		*place_right = false;
@@ -162,27 +280,20 @@ btree_split(BTreeDescr *desc, Page p, OffsetNumber *offset, bool *place_right,
 	new_header->csn = *csn;
 	new_header->undoPos = undo_pos;
 
-	/* Insert tuple */
+	/* Insert new tuple to the right page if needed */
 	if (*place_right)
 	{
 		if (!replace)
 			add_page_item(new_page, *offset, MAXALIGN(tuplesize) + header_size);
 		else
 			resize_page_item(new_page, *offset, MAXALIGN(tuplesize) + header_size);
 		tuple_ptr = BTREE_PAGE_GET_ITEM(new_page, *offset);
+		memcpy(tuple_ptr, tupleheader, header_size);
+		tuple_ptr += header_size;
+		memcpy(tuple_ptr, tuple, tuplesize);
 	}
-	else
-	{
-		if (!replace)
-			add_page_item(p, *offset, MAXALIGN(tuplesize) + header_size);
-		else
-			resize_page_item(p, *offset, MAXALIGN(tuplesize) + header_size);
-		tuple_ptr = BTREE_PAGE_GET_ITEM(p, *offset);
-	}
-	memcpy(tuple_ptr, tupleheader, header_size);
-	tuple_ptr += header_size;
-	memcpy(tuple_ptr, tuple, tuplesize);
 
+	/* Update high key of left page */
 	first_data_key = BTREE_FIRST_DATA_KEY(new_page);
 	rightbound_key = BTREE_PAGE_GET_ITEM(new_page, first_data_key) + header_size;
 	if (leaf)
@@ -197,13 +308,33 @@ btree_split(BTreeDescr *desc, Page p, OffsetNumber *offset, bool *place_right,
 	}
 
 	if (was_rightmost)
+	{
+		if (!(*place_right))
+			(*offset)++;
 		add_page_item(p, BTREE_HIKEY, MAXALIGN(rightbound_key_size));
+	}
 	else
+	{
 		resize_page_item(p, BTREE_HIKEY, MAXALIGN(rightbound_key_size));
+	}
 
 	memcpy(BTREE_PAGE_GET_ITEM(p, BTREE_HIKEY),
 		   rightbound_key, rightbound_key_size);
 
+	/* Insert new tuple to the left page if needed */
+	if (!(*place_right))
+	{
+		if (!replace)
+			add_page_item(p, *offset, MAXALIGN(tuplesize) + header_size);
+		else
+			resize_page_item(p, *offset, MAXALIGN(tuplesize) + header_size);
+		tuple_ptr = BTREE_PAGE_GET_ITEM(p, *offset);
+		memcpy(tuple_ptr, tupleheader, header_size);
+		tuple_ptr += header_size;
+		memcpy(tuple_ptr, tuple, tuplesize);
+	}
+
+
 #ifdef NOT_USED
 	/* Remove leftmost key from the page */
 	if (!leaf)

contrib/in_memory/expected/in_memory.out

@@ -5231,3 +5231,7 @@ SELECT * FROM im_test1;
   3 | xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
 (3 rows)
 
+-- split location bug
+TRUNCATE im_test1;
+INSERT INTO im_test1 (SELECT id, id || repeat('x', 25) FROM generate_series(2, 16, 1) id);
+INSERT INTO im_test1 VALUES(1, repeat('x', 280));

contrib/in_memory/sql/in_memory.sql

@@ -720,3 +720,8 @@ DELETE FROM im_test1 WHERE id = 3;
 INSERT INTO im_test1 SELECT id, repeat('x', 280) FROM generate_series(3, 4) id;
 ROLLBACK;
 SELECT * FROM im_test1;
+
+-- split location bug
+TRUNCATE im_test1;
+INSERT INTO im_test1 (SELECT id, id || repeat('x', 25) FROM generate_series(2, 16, 1) id);
+INSERT INTO im_test1 VALUES(1, repeat('x', 280));