Merge branch 'PGPROEE10' into PGPROEE10_wait_sampling

Author: dmpgpro

contrib/in_memory/Makefile

@@ -41,7 +41,7 @@ submake-isolation:
 submake-in_memory:
 	$(MAKE) -C $(top_builddir)/contrib/in_memory
 
-REGRESSCHECKS = in_memory tuple_limit subtrans ddl_forbid non_coercible
+REGRESSCHECKS = in_memory tuple_limit subtrans ddl_forbid non_coercible collation
 
 regresscheck: | submake-regress submake-in_memory temp-install
 	$(pg_regress_check) \
@@ -60,5 +60,5 @@ isolationcheck: | submake-isolation submake-in_memory temp-install
 
 temp-install: EXTRA_INSTALL=contrib/in_memory
 
-standby-check: submake-in_memory temp-install
+prove-check: submake-in_memory temp-install
 	$(prove_check)

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,33 +280,26 @@ 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, tuplesize + header_size);
+			add_page_item(new_page, *offset, MAXALIGN(tuplesize) + header_size);
 		else
-			resize_page_item(new_page, *offset, tuplesize + header_size);
+			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, tuplesize + header_size);
-		else
-			resize_page_item(p, *offset, 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)
 	{
 		rightbound_key = desc->tuple_make_key(desc, rightbound_key);
-		rightbound_key_size = MAXALIGN(desc->key_len(desc, rightbound_key));
+		rightbound_key_size = desc->key_len(desc, rightbound_key);
 	}
 	else
 	{
@@ -197,13 +308,33 @@ btree_split(BTreeDescr *desc, Page p, OffsetNumber *offset, bool *place_right,
 	}
 
 	if (was_rightmost)
-		add_page_item(p, BTREE_HIKEY, rightbound_key_size);
+	{
+		if (!(*place_right))
+			(*offset)++;
+		add_page_item(p, BTREE_HIKEY, MAXALIGN(rightbound_key_size));
+	}
 	else
-		resize_page_item(p, BTREE_HIKEY, rightbound_key_size);
+	{
+		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)
@@ -275,7 +406,8 @@ insert_to_leaf(BTreeStack *stack,
 
 	while (true)
 	{
-		PageFitsItemResult fit = page_fits_item(p, offset, tuplen + tupheaderlen, replace);
+		LocationIndex			itemsize = MAXALIGN(tuplen) + tupheaderlen;
+		PageFitsItemResult		fit = page_fits_item(p, offset, itemsize, replace);
 
 		if (fit != PageNotFit)
 		{
@@ -286,26 +418,49 @@ insert_to_leaf(BTreeStack *stack,
 			{
 				compact_page(desc, p, &state);
 				offset = btree_page_binary_search(desc, p, tuple, BTreeTuple);
+
+				/*
+				 * If we were going to replace the tuple, then we should check
+				 * it's not gone after compaction.
+				 */
+				if (replace)
+				{
+					Assert(PAGE_IS_LEAF(p)); /* We place tuples only on leaf pages */
+
+					if (offset < header->itemsCount)
+					{
+						Pointer curTuple = BTREE_PAGE_GET_ITEM(p, offset) +
+										   OLeafTupleHeaderSize;
+						if (desc->cmp(desc, tuple, BTreeTuple, curTuple, BTreeTuple))
+							replace = false;
+					}
+					else
+					{
+						replace = false;
+					}
+				}
 			}
 
 			state = modify_page(p, state);
 			if (!replace)
 			{
-				add_page_item(p, offset, tuplen + tupheaderlen);
+				add_page_item(p, offset, itemsize);
 
 				/* Increment number of tuples in progress */
 				if (level == 0)
 					header->nInProgress++;
 			}
 			else
 			{
+				int old_item_size = BTREE_PAGE_GET_ITEM_SIZE(p, offset);
+
 				/*
 				 * We should be able to undo this action without memory
 				 * allocation.  Thus we shouldn't cut space occupied by this
 				 * page item.
 				 */
-				if (tuplen + tupheaderlen > BTREE_PAGE_GET_ITEM_SIZE(p, offset))
-					resize_page_item(p, offset, tuplen + tupheaderlen);
+				if (itemsize > old_item_size)
+					resize_page_item(p, offset, itemsize);
 
 				if (level == 0)
 				{
@@ -324,8 +479,8 @@ insert_to_leaf(BTreeStack *stack,
 						 * tuple.
 						 */
 						if (prev->deleted)
-							header->nVacatedBytes -= BTREE_PAGE_GET_ITEM_SIZE(p, offset)
-													 + sizeof(OffsetNumber);
+							header->nVacatedBytes -= (old_item_size
+													  + sizeof(OffsetNumber));
 					}
 				}
 			}
@@ -656,6 +811,7 @@ btree_modify_internal(BTreeDescr *desc, BTreeModifyAction action,
 				}
 
 				undo_pos = preserve_undo_record(OLeafTupleHeaderSize);
+
 				if (!UndoPosIsValid(undo_pos))
 				{
 					unlock_page(p, state);
@@ -697,6 +853,12 @@ btree_modify_internal(BTreeDescr *desc, BTreeModifyAction action,
 				{
 					/* If we deleted it ourselves, revert this deletion */
 					apply_undo(desc, p, offset, false);
+
+					/*
+					 * We've already reverted the previous delete, so don't
+					 * consider it as sefl update anymore.
+					 */
+					self_update = NoSelfUpdate;
 				}
 				replace = true;
 			}
@@ -747,7 +909,7 @@ btree_modify_internal(BTreeDescr *desc, BTreeModifyAction action,
 		}
 	}
 
-	tuplen = MAXALIGN(desc->tuple_len(desc, tuple));
+	tuplen = desc->tuple_len(desc, tuple);
 	insert_to_leaf(&stack, tuple, tuplen, &leaf_header, replace, self_update, undo_pos);
 
 	return result;