Split __wt_row_ikey into a (rare) raw alloc flavor and the (much more common) WT_REF flavor. Don't free instantiated keys in parent pages.

7 files changed, 53 insertions, 50 deletions

diff --git a/src/btree/bt_handle.c b/src/btree/bt_handle.c
index 299849ad365..c12c3686add 100644
--- a/src/btree/bt_handle.c
+++ b/src/btree/bt_handle.c
@@ -453,8 +453,7 @@ __btree_tree_open_empty(WT_SESSION_IMPL *session, int creation)
 		ref->page = NULL;
 		ref->addr = NULL;
 		ref->state = WT_REF_DELETED;
-		WT_ERR(__wt_row_ikey_incr(
-		    session, root, 0, "", 1, &ref->key.ikey));
+		WT_ERR(__wt_row_ikey_incr(session, root, 0, "", 1, ref));
 		break;
 	WT_ILLEGAL_VALUE_ERR(session);
 	}
diff --git a/src/btree/bt_page.c b/src/btree/bt_page.c
index b5140beb792..3b0e6bcdab8 100644
--- a/src/btree/bt_page.c
+++ b/src/btree/bt_page.c
@@ -611,7 +611,7 @@ __inmem_row_int(WT_SESSION_IMPL *session, WT_PAGE *page, size_t *sizep)
 
 			WT_ERR(__wt_row_ikey_incr(session, page,
 			    WT_PAGE_DISK_OFFSET(page, cell),
-			    current->data, current->size, &ref->key.ikey));
+			    current->data, current->size, ref));
 
 			*sizep += sizeof(WT_IKEY) + current->size;
 			break;
diff --git a/src/btree/bt_slvg.c b/src/btree/bt_slvg.c
index 1cf616a2f6b..d6c20556a9a 100644
--- a/src/btree/bt_slvg.c
+++ b/src/btree/bt_slvg.c
@@ -1858,8 +1858,7 @@ __slvg_row_build_internal(
 			WT_ERR(__slvg_row_build_leaf(session, trk, ref, ss));
 		} else {
 			WT_ERR(__wt_row_ikey_incr(session, page, 0,
-			    trk->row_start.data, trk->row_start.size,
-			    &ref->key.ikey));
+			    trk->row_start.data, trk->row_start.size, ref));
 
 			WT_ERR(__slvg_ovfl_ref_all(session, trk));
 		}
@@ -1981,8 +1980,8 @@ __slvg_row_build_leaf(
 	 */
 	rip = page->pg_row_d + skip_start;
 	WT_ERR(__wt_row_leaf_key(session, page, rip, key, 0));
-	WT_ERR(__wt_row_ikey_incr(session,
-	    ref->home, 0, key->data, key->size, &ref->key.ikey));
+	WT_ERR(__wt_row_ikey_incr(
+	    session, ref->home, 0, key->data, key->size, ref));
 
 	/* Set the referenced flag on overflow pages we're using. */
 	if (trk->trk_ovfl_cnt != 0)
diff --git a/src/btree/bt_split.c b/src/btree/bt_split.c
index d424db61ff9..195d7405caa 100644
--- a/src/btree/bt_split.c
+++ b/src/btree/bt_split.c
@@ -281,11 +281,12 @@ __split_ref_deepen_move(WT_SESSION_IMPL *session,
 	if (parent->type == WT_PAGE_ROW_INT) {
 		if ((ikey = __wt_ref_key_instantiated(ref)) == NULL) {
 			__wt_ref_key(parent, ref, &key, &size);
-			WT_RET(__wt_row_ikey(
-			    session, 0, key, size, &ref->key.ikey));
+			WT_RET(__wt_row_ikey(session, 0, key, size, ref));
 			ikey = ref->key.ikey;
-		} else
+		} else {
 			WT_RET(__split_ovfl_key_cleanup(session, parent, ref));
+			*parent_decrp += sizeof(WT_IKEY) + ikey->size;
+		}
 		*child_incrp += sizeof(WT_IKEY) + ikey->size;
 	}
 
@@ -375,7 +376,6 @@ static int
 __split_deepen(WT_SESSION_IMPL *session, WT_PAGE *parent, uint32_t children)
 {
 	WT_DECL_RET;
-	WT_IKEY *ikey;
 	WT_PAGE *child;
 	WT_PAGE_INDEX *alloc_index, *child_pindex, *pindex;
 	WT_REF **alloc_refp;
@@ -454,8 +454,7 @@ __split_deepen(WT_SESSION_IMPL *session, WT_PAGE *parent, uint32_t children)
 		ref->addr = NULL;
 		if (parent->type == WT_PAGE_ROW_INT) {
 			__wt_ref_key(parent, *parent_refp, &p, &size);
-			WT_ERR(
-			    __wt_row_ikey(session, 0, p, size, &ref->key.ikey));
+			WT_ERR(__wt_row_ikey(session, 0, p, size, ref));
 			parent_incr += sizeof(WT_IKEY) + size;
 		} else
 			ref->key.recno = (*parent_refp)->key.recno;
@@ -586,16 +585,6 @@ __split_deepen(WT_SESSION_IMPL *session, WT_PAGE *parent, uint32_t children)
 	 * fails, we don't roll back that change, because threads may already
 	 * be using the new index.
 	 */
-	for (parent_refp = pindex->index, i = 0;
-	    i < pindex->entries;
-	    parent_refp++, i++)
-		if ((ikey = __wt_ref_key_instantiated(*parent_refp)) != NULL) {
-			size = sizeof(WT_IKEY) + ikey->size;
-			WT_ERR(__split_safe_free(
-			    session, split_gen, 0, ikey, size));
-			parent_decr += size;
-		}
-
 	size = sizeof(WT_PAGE_INDEX) + pindex->entries * sizeof(WT_REF *);
 	WT_ERR(__split_safe_free(session, split_gen, 0, pindex, size));
 	parent_decr += size;
@@ -771,8 +760,8 @@ __wt_multi_to_ref(WT_SESSION_IMPL *session,
 	case WT_PAGE_ROW_INT:
 	case WT_PAGE_ROW_LEAF:
 		ikey = multi->key.ikey;
-		WT_RET(__wt_row_ikey(session, 0,
-		    WT_IKEY_DATA(ikey), ikey->size, &ref->key.ikey));
+		WT_RET(__wt_row_ikey(
+		    session, 0, WT_IKEY_DATA(ikey), ikey->size, ref));
 		incr += sizeof(WT_IKEY) + ikey->size;
 		break;
 	default:
@@ -1185,8 +1174,7 @@ __wt_split_insert(WT_SESSION_IMPL *session, WT_REF *ref, int *splitp)
 	} else
 		WT_ERR(__wt_row_leaf_key(
 		    session, page, &page->pg_row_d[0], key, 1));
-	WT_ERR(__wt_row_ikey(
-	    session, 0, key->data, key->size, &child->key.ikey));
+	WT_ERR(__wt_row_ikey(session, 0, key->data, key->size, child));
 	parent_incr += sizeof(WT_REF) + sizeof(WT_IKEY) + key->size;
 	__wt_scr_free(session, &key);
 
@@ -1205,7 +1193,7 @@ __wt_split_insert(WT_SESSION_IMPL *session, WT_REF *ref, int *splitp)
 	child->state = WT_REF_MEM;
 	WT_ERR(__wt_row_ikey(session, 0,
 	    WT_INSERT_KEY(moved_ins), WT_INSERT_KEY_SIZE(moved_ins),
-	    &child->key.ikey));
+	    child));
 	parent_incr +=
 	    sizeof(WT_REF) + sizeof(WT_IKEY) + WT_INSERT_KEY_SIZE(moved_ins);
 
diff --git a/src/btree/row_key.c b/src/btree/row_key.c
index 292ed8ef85c..f2868afe13a 100644
--- a/src/btree/row_key.c
+++ b/src/btree/row_key.c
@@ -439,7 +439,7 @@ next:		switch (direction) {
 		(void)__wt_row_leaf_key_info(
 		    page, copy, &ikey, &cell, NULL, NULL);
 		if (ikey == NULL) {
-			WT_ERR(__wt_row_ikey(session,
+			WT_ERR(__wt_row_ikey_alloc(session,
 			    WT_PAGE_DISK_OFFSET(page, cell),
 			    keyb->data, keyb->size, &ikey));
 
@@ -462,15 +462,37 @@ err:	__wt_scr_free(session, &tmp);
 }
 
 /*
+ * __wt_row_ikey_alloc --
+ *	Instantiate a key in a WT_IKEY structure.
+ */
+int
+__wt_row_ikey_alloc(WT_SESSION_IMPL *session,
+    uint32_t cell_offset, const void *key, size_t size, WT_IKEY **ikeyp)
+{
+	WT_IKEY *ikey;
+
+	/*
+	 * Allocate memory for the WT_IKEY structure and the key, then copy
+	 * the key into place.
+	 */
+	WT_RET(__wt_calloc(session, 1, sizeof(WT_IKEY) + size, &ikey));
+	ikey->size = WT_STORE_SIZE(size);
+	ikey->cell_offset = cell_offset;
+	memcpy(WT_IKEY_DATA(ikey), key, size);
+	*ikeyp = ikey;
+	return (0);
+}
+
+/*
  * __wt_row_ikey_incr --
  *	Instantiate a key in a WT_IKEY structure and increment the page's
  * memory footprint.
  */
 int
 __wt_row_ikey_incr(WT_SESSION_IMPL *session, WT_PAGE *page,
-    uint32_t cell_offset, const void *key, size_t size, void *ikeyp)
+    uint32_t cell_offset, const void *key, size_t size, WT_REF *ref)
 {
-	WT_RET(__wt_row_ikey(session, cell_offset, key, size, ikeyp));
+	WT_RET(__wt_row_ikey(session, cell_offset, key, size, ref));
 
 	__wt_cache_page_inmem_incr(session, page, sizeof(WT_IKEY) + size);
 
@@ -483,36 +505,30 @@ __wt_row_ikey_incr(WT_SESSION_IMPL *session, WT_PAGE *page,
  */
 int
 __wt_row_ikey(WT_SESSION_IMPL *session,
-    uint32_t cell_offset, const void *key, size_t size, void *dest)
+    uint32_t cell_offset, const void *key, size_t size, WT_REF *ref)
 {
 	WT_IKEY *ikey;
 
-	/*
-	 * Allocate memory for the WT_IKEY structure and the key, then copy
-	 * the key into place.
-	 */
-	WT_RET(__wt_calloc(session, 1, sizeof(WT_IKEY) + size, &ikey));
-	ikey->size = WT_STORE_SIZE(size);
-	ikey->cell_offset = cell_offset;
-	memcpy(WT_IKEY_DATA(ikey), key, size);
+	WT_RET(__wt_row_ikey_alloc(session, cell_offset, key, size, &ikey));
 
 #ifdef HAVE_DIAGNOSTIC
 	{
-	WT_IKEY **ikeyp;
-
 	uintptr_t oldv;
 
-	ikeyp = dest;
-	oldv = (uintptr_t)*ikeyp;
+	oldv = (uintptr_t)ref->key.ikey;
 	WT_DIAGNOSTIC_YIELD;
 
-	/* We should never overwrite an instantiated key. */
+	/*
+	 * We should never overwrite an instantiated key, and we should
+	 * never instantiate a key after a split.
+	 */
 	WT_ASSERT(session, oldv == 0 || (oldv & WT_IK_FLAG) != 0);
+	WT_ASSERT(session, ref->state != WT_REF_SPLIT);
 	WT_ASSERT(session,
-	    WT_ATOMIC_CAS8(*ikeyp, (WT_IKEY *)oldv, ikey));
+	    WT_ATOMIC_CAS8(ref->key.ikey, (WT_IKEY *)oldv, ikey));
 	}
 #else
-	*(WT_IKEY **)dest = ikey;
+	ref->key.ikey = ikey;
 #endif
 	return (0);
 }
diff --git a/src/include/extern.h b/src/include/extern.h
index 7b9e7f13185..c6628ffdc1c 100644
--- a/src/include/extern.h
+++ b/src/include/extern.h
@@ -163,8 +163,9 @@ extern int __wt_col_search(WT_SESSION_IMPL *session, uint64_t recno, WT_REF *lea
 extern int __wt_row_leaf_keys(WT_SESSION_IMPL *session, WT_PAGE *page);
 extern int __wt_row_leaf_key_copy( WT_SESSION_IMPL *session, WT_PAGE *page, WT_ROW *rip, WT_ITEM *key);
 extern int __wt_row_leaf_key_work(WT_SESSION_IMPL *session, WT_PAGE *page, WT_ROW *rip_arg, WT_ITEM *keyb, int instantiate);
-extern int __wt_row_ikey_incr(WT_SESSION_IMPL *session, WT_PAGE *page, uint32_t cell_offset, const void *key, size_t size, void *ikeyp);
-extern int __wt_row_ikey(WT_SESSION_IMPL *session, uint32_t cell_offset, const void *key, size_t size, void *dest);
+extern int __wt_row_ikey_alloc(WT_SESSION_IMPL *session, uint32_t cell_offset, const void *key, size_t size, WT_IKEY **ikeyp);
+extern int __wt_row_ikey_incr(WT_SESSION_IMPL *session, WT_PAGE *page, uint32_t cell_offset, const void *key, size_t size, WT_REF *ref);
+extern int __wt_row_ikey(WT_SESSION_IMPL *session, uint32_t cell_offset, const void *key, size_t size, WT_REF *ref);
 extern int __wt_page_modify_alloc(WT_SESSION_IMPL *session, WT_PAGE *page);
 extern int __wt_row_modify(WT_SESSION_IMPL *session, WT_CURSOR_BTREE *cbt, WT_ITEM *key, WT_ITEM *value, WT_UPDATE *upd, int is_remove);
 extern int __wt_row_insert_alloc(WT_SESSION_IMPL *session, WT_ITEM *key, u_int skipdepth, WT_INSERT **insp, size_t *ins_sizep);
diff --git a/src/reconcile/rec_write.c b/src/reconcile/rec_write.c
index 1ff197bdb9e..7f2639059fe 100644
--- a/src/reconcile/rec_write.c
+++ b/src/reconcile/rec_write.c
@@ -5090,7 +5090,7 @@ __rec_split_row(WT_SESSION_IMPL *session, WT_RECONCILE *r, WT_PAGE *page)
 
 	for (multi = mod->mod_multi,
 	    bnd = r->bnd, i = 0; i < r->bnd_next; ++multi, ++bnd, ++i) {
-		WT_RET(__wt_row_ikey(session, 0,
+		WT_RET(__wt_row_ikey_alloc(session, 0,
 		    bnd->key.data, bnd->key.size, &multi->key.ikey));
 
 		if (bnd->skip == NULL) {